Data Transmission Service:Sinkronisasi data dari ApsaraDB for MongoDB ke Message Queue for Apache Kafka

Batasan database sumber

• Server sumber harus memiliki bandwidth keluar yang mencukupi. Bandwidth yang tidak mencukupi akan mengurangi kecepatan sinkronisasi.

• Saat mengonfigurasi pemetaan nama untuk koleksi, satu tugas dapat menyinkronkan hingga 1.000 koleksi. Melebihi batas ini akan menyebabkan error permintaan. Untuk menyinkronkan lebih dari 1.000 koleksi, buat beberapa tugas atau sinkronkan seluruh database tanpa pemetaan nama.

• Untuk sumber kluster sharded: bidang _id di setiap koleksi yang disinkronkan harus unik, atau ketidakkonsistenan data dapat terjadi.

• Untuk sumber kluster sharded: jumlah node mongos tidak boleh melebihi 10, dan instans tidak boleh berisi dokumen yatim (orphaned documents). Lihat topik FAQ untuk cara menghapus dokumen yatim.

• Instans ApsaraDB for MongoDB standalone, kluster Azure Cosmos DB for MongoDB, dan kluster elastis Amazon DocumentDB tidak didukung sebagai sumber.

• DTS tidak dapat terhubung ke MongoDB melalui titik akhir SRV.

• Oplog harus diaktifkan dan menyimpan data log selama minimal 7 hari, ATAU change streams harus diaktifkan agar DTS dapat berlangganan perubahan data dari 7 hari terakhir. Jika kedua kondisi ini tidak terpenuhi, DTS mungkin gagal mengambil perubahan sumber, dan kehilangan data atau ketidakkonsistenan dapat terjadi di luar cakupan perjanjian tingkat layanan (SLA).

  Penting

  - Gunakan oplog untuk mencatat perubahan database sumber (direkomendasikan). - Change streams memerlukan MongoDB 4.0 atau lebih baru. Sinkronisasi dua arah tidak didukung saat menggunakan change streams. - Untuk kluster Amazon DocumentDB non-elastis, aktifkan change streams dan atur Migration Method ke ChangeStream serta Architecture ke Sharded Cluster.

• Selama sinkronisasi data penuh, jangan mengubah skema database atau koleksi, atau mengubah data bertipe ARRAY.

• Untuk sumber kluster sharded, jangan menjalankan perintah berikut selama sinkronisasi: shardCollection, reshardCollection, unshardCollection, moveCollection, atau movePrimary. Perintah-perintah ini dapat menyebabkan ketidakkonsistenan data.

• Jika database sumber adalah instans MongoDB yang menggunakan arsitektur kluster sharded dan balancer database sumber melakukan penyeimbangan data, latensi dapat terjadi pada instans tersebut.

Batasan lainnya

• Hanya sinkronisasi tingkat koleksi yang didukung.

• Database admin, config, dan local tidak dapat disinkronkan.

• Rekaman tunggal yang melebihi 10 MB akan menyebabkan tugas gagal.

• Transaksi tidak dipertahankan. DTS mengonversi setiap transaksi menjadi rekaman individual di tujuan.

• Jika node broker ditambahkan atau dihapus di instans Kafka tujuan saat tugas DTS sedang berjalan, restart tugas DTS tersebut.

• Pastikan DTS dapat terhubung ke instans sumber dan tujuan. Misalnya, pengaturan keamanan instans database dan parameter listeners serta advertised.listeners dalam file server.properties instans Kafka yang dikelola sendiri tidak membatasi akses dari DTS.

• Jalankan sinkronisasi selama jam sepi ketika pemanfaatan CPU pada database sumber dan tujuan berada di bawah 30%.

• DTS secara otomatis mencoba ulang instans yang gagal dan telah berjalan kurang dari 7 days. Sebelum mengalihkan trafik ke tujuan, hentikan atau lepas instans sinkronisasi untuk mencegah pemulihan otomatis menimpa data tujuan.

• DTS menghitung latensi sinkronisasi data inkremental berdasarkan stempel waktu data terbaru yang telah disinkronkan di database tujuan dan stempel waktu saat ini di database sumber. Jika tidak ada operasi pembaruan yang dilakukan pada database sumber dalam jangka waktu lama, latensi sinkronisasi mungkin tidak akurat. Jika latensi tugas sinkronisasi data terlalu tinggi, Anda dapat melakukan operasi pembaruan pada database sumber untuk memperbarui latensi tersebut.

• Jika tugas DTS gagal dijalankan, dukungan teknis DTS akan mencoba memulihkan tugas tersebut dalam waktu 8 jam. Selama pemulihan, tugas mungkin di-restart, dan parameter tugas mungkin dimodifikasi. Hanya parameter tugas DTS yang mungkin dimodifikasi. Parameter database tidak diubah. Parameter yang mungkin dimodifikasi mencakup tetapi tidak terbatas pada parameter dalam bagian "Modify instance parameters".

• Untuk sumber kluster sharded dengan Oplog sebagai metode sinkronisasi inkremental, DTS tidak menjamin urutan penulisan lintas shard ke topik Kafka target.

• Untuk sumber kluster sharded, nonaktifkan balancer MongoDB selama sinkronisasi data penuh. Aktifkan kembali hanya setelah sinkronisasi penuh selesai dan sinkronisasi inkremental dimulai. Lihat Mengelola balancer ApsaraDB for MongoDB.

Langkah 1: Buka halaman Sinkronisasi Data

Gunakan Konsol DTS atau Konsol DMS.

DTS console

1. Login ke Konsol DTS.Konsol DTS

2. Di panel navigasi kiri, klik Data Synchronization.

3. Di pojok kiri atas, pilih wilayah tempat tugas sinkronisasi berada.

Konsol DMS

Langkah-langkah dapat berbeda tergantung pada mode Konsol DMS. Lihat Mode simple dan Sesuaikan tata letak dan gaya Konsol DMS.

1. Login ke Konsol DMS.Konsol DMS

2. Di bilah navigasi atas, arahkan kursor ke Data + AI lalu pilih DTS (DTS) > Data Synchronization.

3. Dari daftar drop-down di sebelah kanan Data Synchronization Tasks, pilih wilayah.

Langkah 2: Konfigurasi database sumber dan tujuan

Klik Create Task, lalu konfigurasikan parameter yang dijelaskan dalam tabel berikut.

Umum

Database sumber

Database tujuan

Langkah 3: Uji konektivitas

Klik Test Connectivity and Proceed di bagian bawah halaman.

DTS secara otomatis menambahkan blok CIDR servernya ke pengaturan keamanan database sumber dan tujuan, jika diizinkan. Untuk pengaturan manual, lihat Tambahkan blok CIDR server DTS.

Untuk database yang dikelola sendiri yang tidak menggunakan Alibaba Cloud Instance sebagai metode akses, klik Test Connectivity di kotak dialog CIDR Blocks of DTS Servers.

Langkah 4: Konfigurasi objek yang akan disinkronkan

Pada langkah Configure Objects, atur parameter berikut.

Langkah 5: Konfigurasi pengaturan lanjutan

Klik Next: Advanced Settings dan konfigurasikan parameter berikut.

Langkah 6: Jalankan pemeriksaan awal

Klik Next: Save Task Settings and Precheck.

Untuk melihat pratinjau parameter API untuk konfigurasi ini, arahkan kursor ke tombol tersebut lalu klik Preview OpenAPI parameters sebelum melanjutkan.

Tugas tidak dapat dimulai hingga pemeriksaan awal berhasil.

Jika pemeriksaan awal gagal, klik View Details di sebelah setiap item yang gagal, selesaikan masalahnya, lalu klik Precheck Again.

Jika muncul peringatan untuk suatu item: perbaiki peringatan yang tidak dapat diabaikan sebelum melanjutkan. Untuk peringatan yang dapat diabaikan, klik Confirm Alert Details, lalu klik Ignore di dialog, konfirmasi, dan klik Precheck Again. Mengabaikan peringatan dapat menyebabkan ketidakkonsistenan data.

Langkah 7: Beli dan mulai instans

1. Tunggu hingga Success Rate mencapai 100%, lalu klik Next: Purchase Instance.

2. Di halaman buy, konfigurasikan parameter berikut.

   Parameter	Deskripsi
   Billing Method	Subscription: bayar di muka; lebih hemat biaya untuk penggunaan jangka panjang. Pay-as-you-go: ditagih per jam; cocok untuk penggunaan jangka pendek. Lepaskan instans saat tidak lagi digunakan untuk menghentikan penagihan.
   Resource Group Settings	Kelompok sumber daya untuk instans sinkronisasi. Default: default resource group. Lihat Apa itu Resource Management?
   Instance Class	Tingkat kecepatan sinkronisasi. Lihat Kelas instans untuk instansi sinkronisasi data.
   Subscription Duration	Tersedia untuk penagihan Subscription. Opsi: 1–9 bulan, 1 tahun, 2 tahun, 3 tahun, atau 5 tahun.

3. Baca dan terima Data Transmission Service (Pay-as-you-go) Service Terms.

4. Klik Buy and Start, lalu klik OK di dialog konfirmasi.

Tugas akan muncul di daftar tugas. Pantau perkembangannya di sana.

Pilih skenario Anda

Bidang pesan Canal JSON

Ketiga skenario menggunakan envelope Canal JSON tingkat atas yang sama. Bidang berikut muncul di setiap pesan.

Skenario 1: Oplog

Atur Migration Method ke Oplog.

{
    "data": [{
        "person": {
            "skills": ["database", "ai"],
            "name": "testName",
            "age": 18
        },
        "_id": {
            "$oid": "67d27da49591697476e1****"
        },
        "cid": "a"
    }],
    "database": "kafkadb",
    "es": 1741847972000,
    "gtid": null,
    "id": 174184797200000****,
    "isDdl": false,
    "mysqlType": null,
    "old": null,
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741847973438,
    "type": "INSERT"
}

{
    "data": [{
        "$set": {
            "person.age": 20
        }
    }],
    "database": "kafkadb",
    "es": 1741848051000,
    "gtid": null,
    "id": 174184805100000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848051984,
    "type": "UPDATE"
}

{
    "data": [{
        "$set": {
            "salary": 100.0
        }
    }],
    "database": "kafkadb",
    "es": 1741848146000,
    "gtid": null,
    "id": 174184814600000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848147734,
    "type": "UPDATE"
}

{
    "data": [{
        "$unset": {
            "salary": true
        }
    }],
    "database": "kafkadb",
    "es": 1741848207000,
    "gtid": null,
    "id": 174184820700000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848208186,
    "type": "UPDATE"
}

{
    "data": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "database": "kafkadb",
    "es": 1741848289000,
    "gtid": null,
    "id": 174184828900000****,
    "isDdl": false,
    "mysqlType": null,
    "old": null,
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848289798,
    "type": "DELETE"
}

{
    "data": null,
    "database": "kafkadb",
    "es": 1741847893000,
    "gtid": null,
    "id": 1741847893000000005,
    "isDdl": true,
    "mysqlType": null,
    "old": null,
    "pkNames": null,
    "serverId": null,
    "sql": {
        "drop": "kafka_test"
    },
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741847893760,
    "type": "DDL"
}

Skenario 2: ChangeStream — hanya bidang yang diperbarui

Atur Migration Method ke ChangeStream. Atur Obtain the entire document after it is updated ke No.

Pesan insert dan delete identik dengan Skenario 1. Pesan update hanya berisi bidang yang berubah.

{
    "data": [{
        "$set": {
            "person.age": 20
        }
    }],
    "database": "kafkadb",
    "es": 1741848051000,
    "gtid": null,
    "id": 174184805100000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848052912,
    "type": "UPDATE"
}

{
    "data": [{
        "$unset": {
            "salary": 1
        }
    }],
    "database": "kafkadb",
    "es": 1741848207000,
    "gtid": null,
    "id": 174184820700000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848209142,
    "type": "UPDATE"
}

Skenario 3: ChangeStream — dokumen lengkap saat pembaruan

Atur Migration Method ke ChangeStream. Atur Obtain the entire document after it is updated ke Yes.

Event update mengirimkan dokumen lengkap setelah perubahan, bukan hanya bidang yang berubah.

{
    "data": [{
        "person": {
            "skills": ["database", "ai"],
            "name": "testName",
            "age": 20
        },
        "_id": {
            "$oid": "67d27da49591697476e1****"
        },
        "cid": "a"
    }],
    "database": "kafkadb",
    "es": 1741848051000,
    "gtid": null,
    "id": 174184805100000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848052219,
    "type": "UPDATE"
}

{
    "data": [{
        "person": {
            "skills": ["database", "ai"],
            "name": "testName",
            "age": 20
        },
        "_id": {
            "$oid": "67d27da49591697476e1****"
        },
        "salary": 100.0,
        "cid": "a"
    }],
    "database": "kafkadb",
    "es": 1741848146000,
    "gtid": null,
    "id": 174184814600000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848147327,
    "type": "UPDATE"
}

{
    "data": [{
        "person": {
            "skills": ["database", "ai"],
            "name": "testName",
            "age": 20
        },
        "_id": {
            "$oid": "67d27da49591697476e1****"
        },
        "cid": "a"
    }],
    "database": "kafkadb",
    "es": 1741848207000,
    "gtid": null,
    "id": 174184820700000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848208401,
    "type": "UPDATE"
}

Kasus khusus: missing fullDocument di ChangeStream

Saat bidang fullDocument pada event pembaruan ChangeStream tidak tersedia — misalnya, saat dokumen berpindah antar shard dalam koleksi sharded — pesan yang dikirimkan kembali ke perilaku Oplog (pembaruan parsial dengan operator $set atau $unset).

Contoh: pembaruan koleksi sharded dengan fullDocument tidak tersedia

Data dasar sumber:

use admin
db.runCommand({ enablesharding:"dts_test" })

Perubahan inkremental sumber:

use dts_test
sh.shardCollection("dts_test.cstest",{"name":"hashed"})
db.cstest.insert({"_id":1,"name":"a"})
db.cstest.updateOne({"_id":1,"name":"a"},{$set:{"name":"b"}})

{
    "data": [{
        "$set": {
            "name": "b"
        }
    }],
    "database": "dts_test",
    "es": 1740720994000,
    "gtid": null,
    "id": 174072099400000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "name": "a",
        "_id": 1.0
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "cstest",
    "ts": 1740721007099,
    "type": "UPDATE"
}