Menggunakan Amazon Kinesis Data Streams sebagai target AWS Database Migration Service

Anda dapat menggunakannya AWS DMS untuk memigrasikan data ke aliran data Amazon Kinesis. Amazon Kinesis data streams adalah bagian dari layanan Amazon Kinesis Data Streams. Anda dapat menggunakan Kinesis data stream untuk mengumpulkan dan memproses sejumlah besar catatan data secara langsung.

Kinesis data stream adalah serangkaian serpihan. Serpihan secara unik diidentifikasi urutan catatan data dalam stream. Untuk informasi lebih lanjut tentang serpihan di Amazon Kinesis Data Streams, lihat Serpihan di Panduan Developer Amazon Kinesis Data Streams.

AWS Database Migration Service menerbitkan catatan ke aliran data Kinesis menggunakan JSON. Selama konversi, AWS DMS membuat serial setiap catatan dari basis data sumber ke dalam pasangan atribut-nilai dalam format JSON atau format pesan JSON_UNFORMATTED. Format pesan JSON_UNFORMATTED adalah string JSON baris tunggal dengan pembatas baris baru. Ini memungkinkan Amazon Data Firehose untuk mengirimkan data Kinesis ke tujuan Amazon S3, dan kemudian menanyakannya menggunakan berbagai mesin kueri termasuk Amazon Athena.

Anda menggunakan pemetaan objek untuk memigrasi data Anda dari sumber data mana pun yang didukung ke aliran target. Dengan pemetaan objek, Anda menentukan bagaimana struktur catatan data dalam aliran. Anda juga menentukan kunci partisi untuk setiap tabel, yang digunakan Kinesis Data Streams untuk mengelompokkan data ke dalam serpihannya.

Saat AWS DMS membuat tabel pada titik akhir target Kinesis Data Streams, itu membuat tabel sebanyak di titik akhir database sumber. AWS DMS juga menetapkan beberapa nilai parameter Kinesis Data Streams. Biaya pembuatan tabel tergantung pada jumlah data dan jumlah tabel yang akan dimigrasi.

catatan

Opsi Mode SSL di AWS DMS konsol atau API tidak berlaku untuk beberapa streaming data dan layanan NoSQL seperti Kinesis dan DynamoDB. Mereka aman secara default, jadi AWS DMS menunjukkan pengaturan mode SSL sama dengan tidak ada (Mode SSL = Tidak Ada). Anda tidak perlu menyediakan konfigurasi tambahan untuk titik akhir Anda untuk menggunakan SSL. Misalnya, saat menggunakan Kinesis sebagai titik akhir target, Kinesis sudah aman secara default. Semua panggilan API ke Kinesis menggunakan SSL, jadi tidak perlu opsi SSL tambahan di titik akhir. AWS DMS Anda dapat menempatkan data dengan aman dan mengambil data melalui titik akhir SSL menggunakan protokol HTTPS, yang AWS DMS digunakan secara default saat menghubungkan ke Kinesis Data Stream.

Pengaturan titik akhir Kinesis Data Streams

Saat Anda menggunakan titik akhir target Kinesis Data Streams, Anda bisa mendapatkan detail transaksi dan kontrol KinesisSettings menggunakan opsi di API. AWS DMS

Anda dapat mengatur pengaturan koneksi dengan cara berikut:

• Di AWS DMS konsol, menggunakan pengaturan titik akhir.

• Di CLI, menggunakan kinesis-settings opsi perintah. CreateEndpoint

Dalam CLI, gunakan parameter permintaan dari pilihan kinesis-settings berikut:

catatan

Support untuk pengaturan titik akhir IncludeNullAndEmpty tersedia pada AWS DMS versi 3.4.1 dan setelahnya. Tetapi dukungan untuk pengaturan titik akhir berikut lainnya untuk target Kinesis Data Streams tersedia di. AWS DMS

• MessageFormat - Format keluaran untuk catatan yang dibuat pada titik akhir. Format pesan adalah JSON (default) atau JSON_UNFORMATTED (satu baris tanpa tab).

• IncludeControlDetails - Menampilkan informasi kontrol terperinci untuk definisi tabel, definisi kolom, serta perubahan tabel dan kolom pada pesan keluaran Kinesis. Default adalah false.

• IncludeNullAndEmpty - Sertakan NULL dan kolom kosong dalam target. Default adalah false.

• IncludePartitionValue - Menampilkan nilai partisi dalam pesan keluaran Kinesis, kecuali tipe partisi schema-table-type. Default adalah false.

• IncludeTableAlterOperations - Termasuk setiap operasi bahasa definisi data (DDL) yang mengubah tabel dalam data kontrol, seperti rename-table, drop-table, add-column, drop-column, dan rename-column. Default adalah false.

• IncludeTransactionDetails - Menyediakan informasi transaksi detail dari basis data sumber. Informasi ini mencakup stempel waktu melakukan, posisi log, dan nilai-nilai untuk transaction_id, previous_transaction_id, dan transaction_record_id (catatan offset dalam transaksi). Default adalah false.

• PartitionIncludeSchemaTable - Skema prefiks dan nama tabel untuk nilai partisi, ketika tipe partisi primary-key-type. Melakukan hal ini meningkatkan distribusi data di antara serpihan Kinesis. Sebagai contoh, anggaplah bahwa skema SysBench memiliki ribuan tabel dan setiap tabel hanya memiliki rentang terbatas untuk kunci primer. Dalam kasus ini, kunci primer yang sama dikirim dari ribuan tabel ke serpihan yang sama, yang menyebabkan throttling. Default adalah false.

Contoh berikut menunjukkan bahwa kinesis-settings digunakan dengan sebuah contoh perintah create-endpoint yang dikeluarkan menggunakan perintah AWS CLI.

aws dms create-endpoint --endpoint-identifier=$target_name --engine-name kinesis --endpoint-type target
--region us-east-1 --kinesis-settings ServiceAccessRoleArn=arn:aws:iam::333333333333:role/dms-kinesis-role,
StreamArn=arn:aws:kinesis:us-east-1:333333333333:stream/dms-kinesis-target-doc,MessageFormat=json-unformatted,
IncludeControlDetails=true,IncludeTransactionDetails=true,IncludePartitionValue=true,PartitionIncludeSchemaTable=true,
IncludeTableAlterOperations=true

Pengaturan tugas beban penuh multithread

Untuk membantu meningkatkan kecepatan transfer, AWS DMS mendukung beban penuh multithreaded ke instance target Kinesis Data Streams. DMS mendukung multithreading ini dengan pengaturan tugas yang meliputi berikut ini:

• MaxFullLoadSubTasks - Gunakan pilihan ini untuk menunjukkan jumlah maksimum tabel sumber yang dimuat secara paralel. DMS memuat setiap tabel ke dalam tabel target Kinesis yang sesuai menggunakan subtugas khusus. Default adalah 8; nilai maksimum adalah 49.

• ParallelLoadThreads— Gunakan opsi ini untuk menentukan jumlah utas yang AWS DMS digunakan untuk memuat setiap tabel ke dalam tabel target Kinesis. Nilai maksimum untuk target Kinesis Data Streams adalah 32. Anda dapat meminta untuk meningkatkan batas maksimum ini.

• ParallelLoadBufferSize - Gunakan pilihan ini untuk menentukan jumlah catatan maksimum yang disimpan di buffer yang digunakan oleh thread beban paralel untuk memuat data ke target Kinesis. Nilai default adalah 50. Nilai maksimumnya adalah 1.000. Gunakan pengaturan ini dengan ParallelLoadThreads. ParallelLoadBufferSize hanya berlaku bila ada lebih dari satu thread.

• ParallelLoadQueuesPerThread - Gunakan pilihan ini untuk menentukan jumlah antrean yang diakses setiap thread secara bersamaan untuk mengambil catatan data dari antrean dan menghasilkan beban batch untuk target. Default adalah 1. Tetapi untuk target Kinesis dengan berbagai ukuran muatan, rentang yang valid adalah 5-512 antrean per thread.

Pengaturan beban tugas CDC multithread

Anda dapat meningkatkan performa perubahan pengambilan data (CDC) untuk titik akhir target streaming data secara langsung seperti Kinesis menggunakan pengaturan tugas untuk mengubah perilaku PutRecords Panggilan API. Untuk melakukan ini, Anda dapat menentukan jumlah thread yang terjadi bersamaan, antrean per thread, dan jumlah catatan yang disimpan dalam buffer menggunakan pengaturan tugas ParallelApply*. Misalnya, Anda ingin melakukan beban CDC dan menerapkan 128 thread secara paralel. Anda juga ingin mengakses 64 antrean per thread, dengan 50 catatan disimpan per buffer.

Untuk mempromosikan kinerja CDC, AWS DMS mendukung pengaturan tugas ini:

• ParallelApplyThreads— Menentukan jumlah thread bersamaan yang AWS DMS digunakan selama beban CDC untuk mendorong catatan data ke titik akhir target Kinesis. Nilai default adalah nol (0) dan nilai maksimum adalah 32.

• ParallelApplyBufferSize - Menentukan jumlah maksimum catatan yang disimpan di setiap antrean buffer untuk thread serentak untuk mendorong ke titik akhir target Kinesis selama beban CDC. Nilai default adalah 100 dan nilai maksimum adalah 1.000. Gunakan pilihan ini saat ParallelApplyThreads menentukan lebih dari satu thread.

• ParallelApplyQueuesPerThread - Menentukan jumlah antrean yang diakses oleh setiap thread untuk mengambil catatan data dari antrean dan menghasilkan beban batch untuk titik akhir Kinesis selama CDC. Nilai default adalah 1 dan nilai maksimum adalah 512.

Saat menggunakan pengaturan tugas ParallelApply*, default partition-key-type-nya adalah primary-key dari tabel, bukan schema-name.table-name.

Menggunakan citra sebelumnya untuk melihat nilai asli dari baris CDC untuk Kinesis data stream sebagai target

Ketika menulis pembaruan CDC ke target data-streaming seperti Kinesis Anda dapat melihat nilai asli baris basis data sumber sebelum diubah oleh pembaruan. Untuk memungkinkan hal ini, AWS DMS mengisi gambar sebelum peristiwa pembaruan berdasarkan data yang disediakan oleh mesin database sumber.

Mesin basis data sumber yang berbeda memberikan jumlah informasi yang berbeda untuk citra sebelum:

• Oracle menyediakan update untuk kolom hanya jika mereka berubah.

• PostgreSQL menyediakan data hanya untuk kolom yang merupakan bagian dari kunci primer (berubah atau tidak). Untuk menyediakan data untuk semua kolom (diubah atau tidak), Anda perlu mengatur REPLICA_IDENTITY ke FULL bukanDEFAULT. Perhatikan bahwa Anda harus memilih REPLICA_IDENTITY pengaturan dengan hati-hati untuk setiap tabel. Jika Anda mengatur REPLICA_IDENTITY keFULL, semua nilai kolom ditulis ke write-ahead logging (WAL) terus menerus. Hal ini dapat menyebabkan masalah kinerja atau sumber daya dengan tabel yang sering diperbarui.

• MySQL umumnya menyediakan data untuk semua kolom kecuali untuk tipe data BLOB dan CLOB (diubah atau tidak).

Untuk mengaktifkan pencitraan sebelumnya dengan tujuan menambahkan nilai asli dari basis data sumber untuk keluaran AWS DMS , gunakan pengaturan tugas BeforeImageSettings atau parameter add-before-image-columns. Parameter ini menerapkan aturan transformasi kolom.

BeforeImageSettings menambahkan atribut JSON baru untuk setiap operasi pembaruan dengan nilai yang dikumpulkan dari sistem basis data sumber, seperti yang ditunjukkan berikut.

"BeforeImageSettings": {
    "EnableBeforeImage": boolean,
    "FieldName": string,  
    "ColumnFilter": pk-only (default) / non-lob / all (but only one)
}
        

catatan

Hanya berlaku BeforeImageSettings untuk AWS DMS tugas yang berisi komponen CDC, seperti beban penuh ditambah tugas CDC (yang memigrasi data yang ada dan mereplikasi perubahan yang sedang berlangsung), atau ke tugas hanya CDC (yang hanya mereplikasi perubahan data). Jangan terapkan BeforeImageSettings pada tugas-tugas yang beban penuh saja.

Untuk pilihan BeforeImageSettings, hal berikut berlaku:

• Atur pilihan EnableBeforeImage ke true untuk mengaktifkan pencitraan sebelum. Default adalah false.

• Gunakan pilihan FieldName untuk memberikan nama ke atribut JSON baru. Ketika EnableBeforeImage adalah true, FieldName diperlukan dan tidak dapat kosong.

• Pilihan ColumnFilter menentukan kolom untuk menambahkan dengan menggunakan pencitraan sebelumnya. Untuk menambahkan kolom yang hanya merupakan bagian dari kunci primer tabel saja, gunakan nilai default, pk-only. Untuk menambahkan kolom apa pun yang memiliki nilai citra sebelum, gunakan all. Perhatikan bahwa gambar sebelum tidak berisi kolom dengan tipe data LOB, seperti CLOB atau BLOB.

  "BeforeImageSettings": {
      "EnableBeforeImage": true,
      "FieldName": "before-image",
      "ColumnFilter": "pk-only"
    }
  

catatan

Target Amazon S3 tidak mendukung BeforeImageSettings. Untuk target S3, gunakan hanya aturan transformasi add-before-image-columns untuk melakukan pencitraan sebelumnya selama CDC.

Menggunakan aturan transformasi citra sebelumnya

Sebagai alternatif untuk pengaturan tugas, Anda dapat menggunakan parameter add-before-image-columns, yang menerapkan aturan transformasi kolom. Dengan parameter ini, Anda dapat mengaktifkan pencitraan sebelum selama CDC pada target data streaming seperti Kinesis.

Dengan menggunakan add-before-image-columns dalam aturan transformasi, Anda dapat menerapkan kontrol yang lebih halus dari hasil citra sebelum. Aturan transformasi memungkinkan Anda untuk menggunakan pencari objek yang memberikan Anda kontrol atas tabel yang dipilih untuk aturan tersebut. Dan juga, Anda dapat merangkai aturan transformasi, yang memungkinkan aturan yang berbeda untuk diterapkan ke tabel yang berbeda. Anda kemudian dapat memanipulasi kolom yang dihasilkan dengan menggunakan aturan lain.

catatan

Jangan gunakan parameter add-before-image-columns bersamaan dengan pengaturan tugas BeforeImageSettings dalam tugas yang sama. Sebaliknya, gunakan salah satu saja, parameter atau pengaturan, tetapi tidak keduanya, untuk satu tugas.

Sebuah tipe peraturan transformation dengan parameter add-before-image-columns untuk kolom harus menyediakan before-image-def bagian. Berikut ini adalah sebuah contoh.

    {
      "rule-type": "transformation",
      …
      "rule-target": "column",
      "rule-action": "add-before-image-columns",
      "before-image-def":{
        "column-filter": one-of  (pk-only / non-lob / all),
        "column-prefix": string,
        "column-suffix": string,
      }
    }

Nilai dari column-prefix ditambahkan sebelum nama kolom, dan nilai default column-prefix adalah BI_. Nilai dari column-suffix ditambahkan sebelum nama kolom, dan default kosong. Jangan atur column-prefix dan column-suffix ke string kosong.

Pilih satu nilai untuk column-filter. Untuk menambahkan kolom yang merupakan bagian dari kunci primer tabel saja, pilih pk-only. Untuk menambahkan kolom yang bukan tipe LOB saja, gunakan non-lob. Atau pilih all untuk menambahkan kolom yang memiliki nilai citra-sebelum.

Contoh untuk aturan transformasi citra sebelumnya

Aturan transformasi dalam contoh berikut menambahkan kolom baru yang disebut BI_emp_no pada target. Jadi pernyataan seperti UPDATE employees SET emp_no = 3 WHERE emp_no = 1; mengisi BI_emp_no field dengan 1. Ketika Anda menulis pembaruan CDC untuk target Amazon S3, kolom BI_emp_no memungkinkan untuk memberitahu baris asli yang diperbarui.

{
  "rules": [
    {
      "rule-type": "selection",
      "rule-id": "1",
      "rule-name": "1",
      "object-locator": {
        "schema-name": "%",
        "table-name": "%"
      },
      "rule-action": "include"
    },
    {
      "rule-type": "transformation",
      "rule-id": "2",
      "rule-name": "2",
      "rule-target": "column",
      "object-locator": {
        "schema-name": "%",
        "table-name": "employees"
      },
      "rule-action": "add-before-image-columns",
      "before-image-def": {
        "column-prefix": "BI_",
        "column-suffix": "",
        "column-filter": "pk-only"
      }
    }
  ]
}

Untuk informasi tentang penggunaan peraturan tindakan add-before-image-columns, lihat Aturan dan tindakan transformasi.

Prasyarat untuk menggunakan aliran data Kinesis sebagai target AWS Database Migration Service

Peran IAM untuk menggunakan aliran data Kinesis sebagai target AWS Database Migration Service

Sebelum Anda mengatur aliran data Kinesis sebagai target AWS DMS, pastikan Anda membuat peran IAM. Peran ini harus memungkinkan AWS DMS untuk mengasumsikan dan memberikan akses ke aliran data Kinesis yang sedang dimigrasikan ke. Kumpulan izin akses minimum ditampilkan dalam kebijakan IAM berikut.

{
   "Version": "2012-10-17",
   "Statement": [
   {
     "Sid": "1",
     "Effect": "Allow",
     "Principal": {
        "Service": "dms.amazonaws.com"
     },
   "Action": "sts:AssumeRole"
   }
]
} 

        

Peran yang Anda gunakan untuk migrasi ke Kinesis data stream harus memiliki izin berikut.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStream",
                "kinesis:PutRecord",
                "kinesis:PutRecords"
            ],
            "Resource": "arn:aws:kinesis:region:accountID:stream/streamName"
        }
    ]
}

Mengakses aliran data Kinesis sebagai target AWS Database Migration Service

Di AWS DMS versi 3.4.7 dan yang lebih tinggi, untuk terhubung ke titik akhir Kinesis, Anda harus melakukan salah satu hal berikut:

• Konfigurasikan DMS untuk menggunakan titik akhir VPC. Untuk informasi tentang mengonfigurasi DMS untuk menggunakan titik akhir VPC, lihat. Mengkonfigurasi titik akhir VPC sebagaiAWSSumber DMS dan titik akhir target

• Konfigurasikan DMS untuk menggunakan rute publik, yaitu membuat instance replikasi Anda menjadi publik. Untuk informasi tentang contoh replikasi publik, lihat. Instans replikasi publik dan privat

Keterbatasan saat menggunakan Kinesis Data Streams sebagai target AWS Database Migration Service

Batasan berikut berlaku saat menggunakan Kinesis Data Streams sebagai target:

• AWS DMS menerbitkan setiap pembaruan ke satu catatan dalam database sumber sebagai satu catatan data dalam aliran data Kinesis tertentu terlepas dari transaksi. Namun, Anda dapat menyertakan detail transaksi untuk setiap data catatan dengan menggunakan parameter yang relevan dari API KinesisSettings.

• Mode LOB penuh tidak didukung.

• Ukuran LOB maksimum yang didukung adalah 1 MB.

• Kinesis Data Streams tidak mendukung deduplikasi. Aplikasi yang mengonsumsi data dari aliran perlu menangani duplikasi catatan. Untuk informasi lebih lanjut, lihat Menangani catatan duplikat di Panduan Developer Amazon Kinesis Data Streams.

• AWS DMS mendukung dua bentuk berikut untuk kunci partisi:

  • SchemaName.TableName: Kombinasi skema dan nama tabel.

  • ${AttributeName}: Nilai salah satu field di JSON, atau kunci primer dari tabel dalam basis data sumber.

• Untuk informasi tentang mengenkripsi data at rest Anda saat berada dalam Kinesis Data Streams, lihat Perlindungan data dalam Kinesis Data Streams di Panduan Developer AWS Key Management Service .

• BatchApply tidak didukung untuk titik akhir Kinesis. Menggunakan penerapan Batch (misalnya, pengaturan tugas metadata target BatchApplyEnabled) untuk target Kinesis dapat mengakibatkan hilangnya data.

• Target kinesis hanya didukung untuk aliran data Kinesis di AWS akun yang sama dan sama dengan instance replikasi Wilayah AWS .

• Saat bermigrasi dari sumber MySQL, BeforeImage data tidak menyertakan tipe data CLOB dan BLOB. Untuk informasi selengkapnya, lihat Menggunakan citra sebelumnya untuk melihat nilai asli dari baris CDC untuk Kinesis data stream sebagai target.

• AWS DMS tidak mendukung migrasi nilai tipe BigInt data dengan lebih dari 16 digit. Untuk mengatasi batasan ini, Anda dapat menggunakan aturan transformasi berikut untuk mengonversi BigInt kolom menjadi string. Untuk informasi selengkapnya tentang aturan transformasi, lihat Aturan dan tindakan transformasi.

  {
      "rule-type": "transformation",
      "rule-id": "id",
      "rule-name": "name",
      "rule-target": "column",
      "object-locator": {
          "schema-name": "valid object-mapping rule action",
          "table-name": "",
          "column-name": ""
      },
      "rule-action": "change-data-type",
      "data-type": {
          "type": "string",
          "length": 20
      }
  }

Menggunakan pemetaan objek untuk memigrasi data ke Kinesis data stream

AWS DMS menggunakan aturan pemetaan tabel untuk memetakan data dari sumber ke aliran data Kinesis target. Untuk memetakan data ke aliran target, Anda menggunakan jenis aturan pemetaan tabel yang disebut pemetaan objek. Anda menggunakan pemetaan objek untuk menentukan cara pencatatan data di peta sumber ke pencatatan data yang dipublikasikan ke Kinesis data stream.

Kinesis data stream tidak memiliki struktur preset selain memiliki kunci partisi. Dalam aturan pemetaan objek, nilai-nilai yang mungkin dari partition-key-type untuk catatan data adalah schema-table, transaction-id, primary-key, constant, dan attribute-name.

Untuk membuat aturan pemetaan objek, Anda tentukan rule-type sebagai object-mapping. Aturan ini menentukan jenis pemetaan objek yang ingin Anda gunakan.

Struktur untuk aturan tersebut adalah sebagai berikut.

{
    "rules": [
        {
            "rule-type": "object-mapping",
            "rule-id": "id",
            "rule-name": "name",
            "rule-action": "valid object-mapping rule action",
            "object-locator": {
                "schema-name": "case-sensitive schema name",
                "table-name": ""
            }
        }
    ]
}

AWS DMS saat ini mendukung map-record-to-record dan map-record-to-document sebagai satu-satunya nilai yang valid untuk rule-action parameter. Pengaturan ini memengaruhi nilai yang tidak dikecualikan sebagai bagian dari daftar exclude-columns atribut. map-record-to-documentNilai map-record-to-record dan menentukan bagaimana AWS DMS menangani catatan ini secara default. Nilai-nilai ini tidak mempengaruhi pemetaan atribut dengan cara apapun.

Gunakan map-record-to-record saat bermigrasi dari basis data relasional ke Kinesis data stream. Jenis aturan ini menggunakan nilai taskResourceId.schemaName.tableName dari basis data relasional sebagai kunci partisi dalam Kinesis data stream dan membuat atribut untuk setiap kolom dalam basis data sumber.

Saat menggunakanmap-record-to-record, perhatikan hal berikut:

• Pengaturan ini hanya memengaruhi kolom yang dikecualikan oleh exclude-columns daftar.

• Untuk setiap kolom tersebut, AWS DMS buat atribut yang sesuai dalam topik target.

• AWS DMS menciptakan atribut yang sesuai ini terlepas dari apakah kolom sumber digunakan dalam pemetaan atribut.

Gunakan map-record-to-document untuk menempatkan kolom sumber ke dalam satu, dokumen datar dalam aliran target yang sesuai menggunakan nama atribut “_doc”. AWS DMS menempatkan data ke dalam satu peta datar pada sumber yang disebut ”_doc“. Penempatan ini berlaku untuk kolom mana pun di tabel sumber yang tidak tercantum dalam daftar atribut exclude-columns.

Salah satu cara untuk memahami map-record-to-record adalah melihatnya beraksi. Untuk contoh ini, anggaplah bahwa Anda memulai dengan baris tabel basis data relasional dengan struktur dan data berikut.

FirstName	LastName	StoreId	HomeAddress	HomePhone	WorkAddress	WorkPhone	DateofBirth
Randy	Marsh	5	221B Baker Street	1234567890	31 Spooner Street, Quahog	9876543210	02/29/1988

Untuk memigrasi informasi ini dari skema bernama Test ke Kinesis data stream, Anda membuat aturan untuk memetakan data ke aliran target. Aturan berikut menggambarkan pemetaan.

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "1",
            "rule-action": "include",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "%"
            }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "2",
            "rule-name": "DefaultMapToKinesis",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customers"
            }
        }
    ]
}

Berikut ini menggambarkan format catatan yang dihasilkan dalam Kinesis data stream:

• StreamName: XXX

• PartitionKey: Test.Customers //schmaname.TableName

• Data: //Pesan JSON berikut

  
    {
       "FirstName": "Randy",
       "LastName": "Marsh",
       "StoreId":  "5",
       "HomeAddress": "221B Baker Street",
       "HomePhone": "1234567890",
       "WorkAddress": "31 Spooner Street, Quahog",
       "WorkPhone": "9876543210",
       "DateOfBirth": "02/29/1988"
    }   
                  
            

Namun, anggaplah Anda menggunakan aturan yang sama tetapi mengubah parameter rule-action untuk map-record-to-document dan mengecualikan kolom tertentu. Aturan berikut menggambarkan pemetaan.

{
	"rules": [
	   {
			"rule-type": "selection",
			"rule-id": "1",
			"rule-name": "1",
			"rule-action": "include",
			"object-locator": {
				"schema-name": "Test",
				"table-name": "%"
			}
		},
		{
			"rule-type": "object-mapping",
			"rule-id": "2",
			"rule-name": "DefaultMapToKinesis",
			"rule-action": "map-record-to-document",
			"object-locator": {
				"schema-name": "Test",
				"table-name": "Customers"
			},
			"mapping-parameters": {
				"exclude-columns": [
					"homeaddress",
					"homephone",
					"workaddress",
					"workphone"
				]
			}
		}
	]
}                

Dalam kasus ini, kolom tidak tercantum dalam parameter exclude-columns, FirstName, LastName, StoreId dan DateOfBirth, dipetakan ke _doc. Berikut ini menggambarkan format catatan yang dihasilkan.

       {
            "data":{
                "_doc":{
                    "FirstName": "Randy",
                    "LastName": "Marsh",
                    "StoreId":  "5",
                    "DateOfBirth": "02/29/1988"
                }
            }
        }
            

Restrukturisasi data dengan pemetaan atribut

Anda dapat merestrukturisasi data saat memigrasinya ke Kinesis data stream menggunakan peta atribut. Misalnya, Anda mungkin ingin menggabungkan beberapa field di sumber menjadi field tunggal dalam target. Peta atribut berikut menggambarkan bagaimana merestrukturisasi data.

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "1",
            "rule-action": "include",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "%"
            }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "2",
            "rule-name": "TransformToKinesis",
            "rule-action": "map-record-to-record",
            "target-table-name": "CustomerData",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customers"
            },
            "mapping-parameters": {
                "partition-key-type": "attribute-name",
                "partition-key-name": "CustomerName",
                "exclude-columns": [
                    "firstname",
                    "lastname",
                    "homeaddress",
                    "homephone",
                    "workaddress",
                    "workphone"
                ],
                "attribute-mappings": [
                    {
                        "target-attribute-name": "CustomerName",
                        "attribute-type": "scalar",
                        "attribute-sub-type": "string",
                        "value": "${lastname}, ${firstname}"
                    },
                    {
                        "target-attribute-name": "ContactDetails",
                        "attribute-type": "document",
                        "attribute-sub-type": "json",
                        "value": {
                            "Home": {
                                "Address": "${homeaddress}",
                                "Phone": "${homephone}"
                            },
                            "Work": {
                                "Address": "${workaddress}",
                                "Phone": "${workphone}"
                            }
                        }
                    }
                ]
            }
        }
    ]
}

Untuk menetapkan nilai konstan untuk partition-key, tentukan nilai partition-key. Misalnya, Anda mungkin melakukan ini untuk memaksa semua data untuk disimpan dalam serpihan tunggal. Pemetaan berikut menggambarkan pendekatan ini.

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "1",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "%"
            },
            "rule-action": "include"
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "1",
            "rule-name": "TransformToKinesis",
            "rule-action": "map-record-to-document",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customer"
            },
            "mapping-parameters": {
                "partition-key": {
                    "value": "ConstantPartitionKey"
                },
                "exclude-columns": [
                    "FirstName",
                    "LastName",
                    "HomeAddress",
                    "HomePhone",
                    "WorkAddress",
                    "WorkPhone"
                ],
                "attribute-mappings": [
                    {
                        "attribute-name": "CustomerName",
                        "value": "${FirstName},${LastName}"
                    },
                    {
                        "attribute-name": "ContactDetails",
                        "value": {
                            "Home": {
                                "Address": "${HomeAddress}",
                                "Phone": "${HomePhone}"
                            },
                            "Work": {
                                "Address": "${WorkAddress}",
                                "Phone": "${WorkPhone}"
                            }
                        }
                    },
                    {
                        "attribute-name": "DateOfBirth",
                        "value": "${DateOfBirth}"
                    }
                ]
            }
        }
    ]
}

catatan

Nilai partition-key untuk catatan kontrol yang ditujukan khusus untuk tabel tertentu adalah TaskId.SchemaName.TableName. Nilai partition-key untuk catatan kontrol yang ditujukan untuk tugas tertentu adalah TaskId dari catatan tersebut. Menentukan nilai partition-key dalam pemetaan objek tidak berdampak pada partition-key untuk catatan kontrol.

Format pesan untuk Kinesis Data Streams

Keluaran JSON hanya berupa daftar pasangan nilai-kunci. Format pesan JSON_UNFORMATTED adalah string JSON baris tunggal dengan pembatas baris baru.

AWS DMS menyediakan bidang cadangan berikut untuk mempermudah penggunaan data dari Kinesis Data Streams:

RecordType

Jenis catatan dapat berupa data atau kontrol. Catatan data mewakili baris yang sebenarnya dalam sumber. Kontrol catatan adalah untuk peristiwa-peristiwa penting di aliran, misalnya restart tugas.

Operasi

Untuk catatan data, operasinya dapat load, insert, update, atau delete.

Untuk catatan kontrol, operasi dapat berupacreate-table,rename-table,drop-table,change-columns,add-column,drop-column,rename-column, ataucolumn-type-change.

SchemaName

Skema sumber untuk catatan. Field ini bisa kosong untuk catatan kontrol.

TableName

Tabel sumber untuk catatan. Field ini bisa kosong untuk catatan kontrol.

Stempel waktu

Stempel waktu untuk saat pesan JSON dibuat. Field ini diformat dengan format ISO 8601.