Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Menggunakan Apache Kafka sebagai target AWS Database Migration Service

Anda dapat menggunakan AWS DMS untuk memigrasikan data ke cluster Apache Kafka. Apache Kafka adalah platform streaming terdistribusi. Anda dapat menggunakan Apache Kafka untuk menelan dan memproses data streaming secara langsung.

AWS juga menawarkan Amazon Managed Streaming untuk Apache Kafka (Amazon MSK) untuk digunakan sebagai AWS DMS target. Amazon MSK adalah layanan streaming Apache Kafka yang terkelola penuh yang menyederhanakan implementasi dan pengelolaan instans Apache Kafka. Ini bekerja dengan versi Apache Kafka open-source, dan Anda mengakses instance Amazon MSK sebagai AWS DMS target persis seperti instance Apache Kafka lainnya. Untuk informasi lebih lanjut, lihat Apa itu Amazon MSK? dalam Panduan Developer: Amazon Managed Streaming for Apache Kafka.

Sebuah klaster Kafka menyimpan aliran catatan dalam kategori yang disebut topik yang dibagi menjadi partisi. Partisi adalah urutan catatan data (pesan) yang diidentifikasi secara unik secara unik dalam suatu topik. Partisi dapat didistribusikan ke beberapa broker dalam sebuah klaster untuk mengaktifkan pemrosesan paralel catatan topik. Untuk informasi lebih lanjut tentang topik dan partisi dan distribusi mereka di Apache Kafka, lihat Topik dan log dan Distribusi.

Cluster Kafka Anda dapat berupa instans Amazon MSK, cluster yang berjalan pada instans Amazon EC2, atau cluster lokal. Instans Amazon MSK atau cluster pada instans Amazon EC2 dapat berada di VPC yang sama atau yang berbeda. Jika cluster Anda berada di lokasi, Anda dapat menggunakan server nama lokal Anda sendiri untuk instans replikasi untuk menyelesaikan nama host cluster. Untuk informasi tentang menyiapkan server nama untuk instance replikasi Anda, lihatMenggunakan server nama on-premise Anda sendiri. Untuk informasi selengkapnya tentang menyiapkan jaringan, lihatMenyiapkan jaringan untuk instans replikasi.

Saat menggunakan cluster Amazon MSK, pastikan grup keamanannya mengizinkan akses dari instance replikasi Anda. Untuk informasi tentang mengubah grup keamanan untuk cluster Amazon MSK, lihat Meng ubah grup keamanan cluster Amazon MSK.

AWS Database Migration Service menerbitkan catatan ke topik Kafka menggunakan JSON. Selama konversi, AWS DMS membuat serial setiap catatan dari basis data sumber ke dalam pasangan atribut-nilai dalam format JSON.

Untuk memigrasikan data Anda dari sumber data apa pun yang didukung ke klaster Kafka target, Anda menggunakan pemetaan objek. Dengan pemetaan objek, Anda menentukan bagaimana struktur catatan data dalam topik target. Anda juga menentukan kunci partisi untuk setiap tabel, yang digunakan Apache Kafka untuk mengelompokkan data ke dalam partisi.

Saat ini, AWS DMS mendukung satu topik per tugas. Untuk satu tugas dengan beberapa tabel, semua pesan masuk ke satu topik. Setiap pesan menyertakan bagian metadata yang mengidentifikasi skema target dan tabel. AWS DMS versi 3.4.6 dan yang lebih tinggi mendukung replikasi multitopik menggunakan pemetaan objek. Untuk informasi selengkapnya, lihat Replikasi multitopik menggunakan pemetaan objek.

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

Menghubungkan ke Kafka menggunakan Keamanan Lapisan Pengangkutan (TLS)

Sebuah klaster Kafka menerima koneksi aman menggunakan Keamanan Lapisan Pengangkutan (TLS). Dengan AWS DMS, Anda dapat menggunakan salah satu dari tiga opsi protokol keamanan berikut untuk mengamankan koneksi titik akhir Kafka.

Menggunakan enkripsi SSL dengan Amazon MSK atau klaster Kafka yang dikelola sendiri

Anda dapat menggunakan enkripsi SSL untuk mengamankan koneksi titik akhir ke Amazon MSK atau klaster Kafka yang dikelola sendiri. Saat Anda menggunakan metode autentikasi enkripsi SSL, klien memvalidasi identitas server melalui sertifikat server. Kemudian koneksi terenkripsi dibuat antara server dan klien.

"KafkaSettings": {
    "SecurityProtocol": "ssl-encryption", 
}
                

Menggunakan autentikasi SSL

Anda dapat menggunakan autentikasi SSL untuk mengamankan koneksi titik akhir ke Amazon MSK atau klaster Kafka yang dikelola sendiri.

Untuk mengaktifkan autentikasi klien dan enkripsi menggunakan otentikasi SSL untuk menyambung ke Amazon MSK, lakukan hal berikut:

Menggunakan SASL-SSL otentikasi untuk terhubung ke Amazon MSK

Metode The Simple Authentication and Security Layer (SASL) menggunakan nama pengguna dan kata sandi untuk memvalidasi identitas klien, dan membuat koneksi terenkripsi antara server dan klien.

Untuk menggunakan SASL, pertama-tama buat nama pengguna dan kata sandi yang aman ketika mengatur Amazon MSK klaster Anda. Untuk deskripsi cara mengatur nama pengguna dan kata sandi yang aman untuk cluster Amazon MSK, lihat Men yiapkan SASL/SCRAM otentikasi untuk cluster Amazon MSK di Panduan Pengembang Amazon Managed Streaming untuk Apache Kafka.

Kemudian, ketika Anda membuat titik akhir target Kafka, setel pengaturan titik akhir protokol keamanan (SecurityProtocol) menggunakan pilihan sasl-ssl. Anda juga mengatur pilihan SaslUsername dan SaslPassword. Pastikan semua ini konsisten dengan nama pengguna dan kata sandi aman yang Anda buat ketika pertama kali menyiapkan klaster Amazon MSK Anda, seperti yang ditunjukkan dalam contoh JSON berikut.

                   
"KafkaSettings": {
    "SecurityProtocol": "sasl-ssl",
    "SaslUsername":"Amazon MSK cluster secure user name",
    "SaslPassword":"Amazon MSK cluster secure password"                    
}
                    

Menggunakan citra sebelumnya untuk melihat nilai-nilai asli dari baris CDC untuk Apache Kafka sebagai target

Ketika menulis pembaruan CDC ke target data-streaming seperti Kafka Anda dapat melihat nilai asli baris basis data sumber sebelum diubah oleh pembaruan.. Untuk memungkinkan hal ini AWS DMS , isi 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:

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)
}
        

Untuk pilihan BeforeImageSettings, hal berikut berlaku:

Menggunakan aturan transformasi citra sebelumnya

Sebagai alternatif untuk pengaturan tugas, Anda dapat menggunakan add-before-image-columns parameter, yang menerapkan aturan transformasi kolom. Dengan parameter ini, Anda dapat mengaktifkan pencitraan sebelumnya 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 sebelumnya. 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.

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.

Keterbatasan saat menggunakan Apache Kafka sebagai target AWS Database Migration Service

Batasan berikut berlaku saat menggunakan Apache Kafka sebagai target:

Menggunakan pemetaan objek untuk bermigrasi data ke topik Kafka

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

topik Kafka tidak memiliki struktur preset selain memiliki kunci partisi.

Untuk membuat aturan pemetaan objek, tetapkan 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 ketika bermigrasi dari basis data relasional ke sebuah topik Kafka. Jenis aturan ini menggunakan nilai taskResourceId.schemaName.tableName dari basis data relasional sebagai kunci partisi dalam topik Kafka dan membuat atribut untuk masing-masing kolom dalam basis data sumber.

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

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.

Untuk memigrasi informasi ini dari skema bernama Test ke topik Kafka, buatlah aturan untuk memetakan data ke topik 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": "DefaultMapToKafka",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customers"
            }
        }
    ]
}

Mengingat topik Kafka dan kunci partisi (dalam hal ini, taskResourceId.schemaName.tableName), berikut ini menggambarkan format catatan yang dihasilkan menggunakan data sampel kami di topik target Kafka:

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

Restrukturisasi data dengan pemetaan atribut

Anda dapat merestrukturisasi data saat migrasi ke topik kafka berlangsung 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": "TransformToKafka",
            "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 mengatur nilai konstan untukpartition-key, tentukan"partition-key-type: "constant", ini menetapkan nilai partisi keconstant. Sebagai contoh, Anda mungkin melakukan ini untuk memaksa semua data untuk disimpan dalam partisi 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": "TransformToKafka",
            "rule-action": "map-record-to-document",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customer"
            },
            "mapping-parameters": {
                "partition-key-type": "constant",
                "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}"
                    }
                ]
            }
        }
    ]
}

Replikasi multitopik menggunakan pemetaan objek

Secara default, AWS DMS tugas memigrasikan semua data sumber ke salah satu topik Kafka berikut:

Dengan versi AWS DMS engine 3.4.6 dan yang lebih tinggi, Anda dapat menggunakan kafka-target-topic atribut untuk memetakan setiap tabel sumber yang dimigrasi ke topik terpisah. Misalnya, aturan pemetaan objek berikut memigrasikan tabel sumber Customer dan Address ke topik Kafka customer_topic danaddress_topic, masing-masing. Pada saat yang sama, AWS DMS memigrasikan semua tabel sumber lainnya, termasuk Bills tabel dalam Test skema, ke topik yang ditentukan di titik akhir target.

{
    "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": "MapToKafka1",
            "rule-action": "map-record-to-record",
            "kafka-target-topic": "customer_topic",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customer" 
            },
            "partition-key-type": "constant"
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "3",
            "rule-name": "MapToKafka2",
            "rule-action": "map-record-to-record",
            "kafka-target-topic": "address_topic",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Address"
            },
            "partition-key-type": "constant"
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "4",
            "rule-name": "DefaultMapToKafka",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Bills"
            }
        }
    ]
}

Dengan menggunakan replikasi multitopik Kafka, Anda dapat mengelompokkan dan memigrasikan tabel sumber untuk memisahkan topik Kafka menggunakan tugas replikasi tunggal.

Secara default, AWS DMS mengubah nama topik Kafka target menjadi semua huruf kecil. Dengan versi AWS DMS mesin 3.4.6 dan yang lebih tinggi, Anda dapat menggunakan disable-kafka-topic-normalization atribut dalam aturan pemetaan objek untuk mempertahankan kasus asli nama topik Kafka target. Setel atribut ini true untuk AWS DMS mencegah mengubah nama topik menjadi huruf kecil. Nilai defaultnya adalah false (nama topik dikonversi menjadi huruf kecil). Contoh berikut menunjukkan cara menggunakan pengaturan ini.

{
    "rules": [
        {
            "rule-type": "object-mapping",
            "rule-id": "1",
            "rule-name": "MapToKafka",
            "rule-action": "map-record-to-record",
            "kafka-target-topic": "MyTopic",
            "disable-kafka-topic-normalization": true,
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customer"
            }
        }
    ]
}

Format pesan untuk Apache Kafka

Keluaran JSON hanya berupa daftar pasangan nilai-kunci.

Contoh pesan JSON berikut menggambarkan pesan tipe data dengan semua metadata tambahan.

{ 
   "data":{ 
      "id":100000161,
      "fname":"val61s",
      "lname":"val61s",
      "REGION":"val61s"
   },
   "metadata":{ 
      "timestamp":"2019-10-31T22:53:59.721201Z",
      "record-type":"data",
      "operation":"insert",
      "partition-key-type":"primary-key",
      "partition-key-value":"sbtest.sbtest_x.100000161",
      "schema-name":"sbtest",
      "table-name":"sbtest_x",
      "transaction-id":9324410911751,
      "transaction-record-id":1,
      "prev-transaction-id":9324410910341,
      "prev-transaction-record-id":10,
      "commit-timestamp":"2019-10-31T22:53:55.000000Z",
      "stream-position":"mysql-bin-changelog.002171:36912271:0:36912333:9324410911751:mysql-bin-changelog.002171:36912209"
   }
}
                

Contoh pesan JSON berikut menggambarkan pesan tipe kontrol.

{ 
   "control":{ 
      "table-def":{ 
         "columns":{ 
            "id":{ 
               "type":"WSTRING",
               "length":512,
               "nullable":false
            },
            "fname":{ 
               "type":"WSTRING",
               "length":255,
               "nullable":true
            },
            "lname":{ 
               "type":"WSTRING",
               "length":255,
               "nullable":true
            },
            "REGION":{ 
               "type":"WSTRING",
               "length":1000,
               "nullable":true
            }
         },
         "primary-key":[ 
            "id"
         ],
         "collation-name":"latin1_swedish_ci"
      }
   },
   "metadata":{ 
      "timestamp":"2019-11-21T19:14:22.223792Z",
      "record-type":"control",
      "operation":"create-table",
      "partition-key-type":"task-id",
      "schema-name":"sbtest",
      "table-name":"sbtest_t1"
   }
}