Data Transmission Service:Sinkronisasi data dari instans kluster sharded ApsaraDB for MongoDB ke fungsi Function Compute

Apa yang diterima fungsi Anda

Setiap pemanggilan melewatkan objek dengan array Records. Setiap elemen dalam array merepresentasikan satu event perubahan:

Fungsi menerima dua kategori operasi:

• DDL — perubahan skema: CreateIndex, CreateCollection, DropIndex, DropCollection

• DML — perubahan data: INSERT, UPDATE, DELETE

db.createCollection("testCollection")

{
  "Records": [{
    "data": [{"doc": "{\"create\": \"testCollection\", \"idIndex\": {\"v\": 2, \"key\": {\"_id\": 1}, \"name\": \"_id_\"}}"}],
    "pkNames": ["_id"],
    "type": "DDL",
    "es": 1694056437000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": true,
    "table": "testCollection",
    "ts": 1694056437510
  }]
}

db.testCollection.drop()

{
  "Records": [{
    "data": [{"doc": "{\"drop\": \"testCollection\"}"}],
    "pkNames": ["_id"],
    "type": "DDL",
    "es": 1694056577000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": true,
    "table": "testCollection",
    "ts": 1694056577789
  }]
}

db.testCollection.createIndex({name: 1})

{
  "Records": [{
    "data": [{"doc": "{\"createIndexes\": \"testCollection\", \"v\": 2, \"key\": {\"name\": 1}, \"name\": \"name_1\"}"}],
    "pkNames": ["_id"],
    "type": "DDL",
    "es": 1694056670000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": true,
    "table": "testCollection",
    "ts": 1694056670719
  }]
}

db.testCollection.dropIndex({name: 1})

{
  "Records": [{
    "data": [{"doc": "{\"dropIndexes\": \"testCollection\", \"index\": \"name_1\"}"}],
    "pkNames": ["_id"],
    "type": "DDL",
    "es": 1694056817000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": true,
    "table": "$cmd",
    "ts": 1694056818035
  }]
}

// Masukkan batch
db.runCommand({insert: "user", documents: [{"name": "jack", "age": 20}, {"name": "lili", "age": 20}]})

// Masukkan tunggal
db.user.insert({"name": "jack", "age": 20})

{
  "Records": [
    {
      "data": [{"doc": "{\"_id\": {\"$oid\": \"64f9397f6e255f74d65a****\"}, \"name\": \"jack\", \"age\": 20}"}],
      "pkNames": ["_id"],
      "type": "INSERT",
      "es": 1694054783000,
      "database": "MongoDBTest",
      "id": 0,
      "isDdl": false,
      "table": "user",
      "ts": 1694054784427
    },
    {
      "data": [{"doc": "{\"_id\": {\"$oid\": \"64f9397f6e255f74d65a****\"}, \"name\": \"lili\", \"age\": 20}"}],
      "pkNames": ["_id"],
      "type": "INSERT",
      "es": 1694054783000,
      "database": "MongoDBTest",
      "id": 0,
      "isDdl": false,
      "table": "user",
      "ts": 1694054784428
    }
  ]
}

db.user.update({"name": "jack"}, {$set: {"age": 30}})

{
  "Records": [{
    "data": [{"doc": "{\"$set\": {\"age\": 30}}"}],
    "pkNames": ["_id"],
    "old": [{"doc": "{\"_id\": {\"$oid\": \"64f9397f6e255f74d65a****\"}}"}],
    "type": "UPDATE",
    "es": 1694054989000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": false,
    "table": "user",
    "ts": 1694054990555
  }]
}

db.user.remove({"name": "jack"})

{
  "Records": [{
    "data": [{"doc": "{\"_id\": {\"$oid\": \"64f9397f6e255f74d65a****\"}}"}],
    "pkNames": ["_id"],
    "type": "DELETE",
    "es": 1694055452000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": false,
    "table": "user",
    "ts": 1694055452852
  }]
}

Batasan

Tinjau batasan berikut sebelum membuat tugas sinkronisasi Anda.

Batasan cakupan:

• Hanya sinkronisasi data inkremental yang didukung. Sinkronisasi data penuh tidak didukung.

• Sinkronisasi lintas wilayah tidak didukung.

• DTS tidak dapat menyinkronkan data dari database admin, config, atau local.

• Pemetaan objek tidak didukung.

• Informasi transaksi tidak dipertahankan. Transaksi dikonversi menjadi catatan individual di tujuan.

Batasan database sumber:

• Sumber harus menggunakan arsitektur kluster sharded dengan tidak lebih dari 10 node Mongos.

• Sumber tidak boleh berupa kluster Azure Cosmos DB for MongoDB atau kluster elastis Amazon DocumentDB.

• Koleksi yang akan disinkronkan harus memiliki kendala PRIMARY KEY atau UNIQUE, dan semua field harus unik. Jika tidak, database tujuan mungkin berisi catatan data duplikat.

• Ukuran dokumen tunggal tidak boleh melebihi 16 MB. Dokumen yang melebihi batas ini tidak dapat ditulis ke fungsi tujuan dan akan memicu error. Gunakan fitur ekstrak, transformasi, dan muat (ETL) untuk memfilter field besar jika diperlukan.

• Sinkronkan hingga 1.000 koleksi per tugas. Untuk jumlah koleksi yang lebih banyak, buat beberapa tugas atau lakukan sinkronisasi pada tingkat database.

• Instans sumber tidak boleh diskalakan selama tugas sinkronisasi sedang berjalan.

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

• Jika balancer database sumber aktif, tugas mungkin mengalami penundaan.

• Jika database sumber adalah database MongoDB yang dikelola sendiri dengan arsitektur kluster sharded, atur parameter Access Method ke Express Connect, VPN Gateway, Smart Access Gateway, atau Cloud Enterprise Network (CEN).

Batasan penulisan:

• Untuk operasi INSERT, data yang dimasukkan harus berisi kunci shard.

• Untuk operasi UPDATE, kunci shard tidak boleh dimodifikasi.

• Konfigurasikan hanya satu tugas DTS per fungsi tujuan. Beberapa tugas yang menulis ke fungsi yang sama dapat menyebabkan error data.

Persyaratan oplog dan change stream:

• Oplog harus diaktifkan dan menyimpan data log minimal selama 7 hari, ATAU change stream harus diaktifkan untuk mencakup perubahan minimal 7 hari terakhir. Jika kedua kondisi tersebut tidak terpenuhi, DTS mungkin gagal menangkap perubahan sumber, yang dapat menyebabkan inkonsistensi atau kehilangan data yang tidak dicakup oleh perjanjian tingkat layanan (SLA) DTS.

Batasan change stream (jika berlaku):

• Change stream memerlukan MongoDB 4.0 atau versi lebih baru.

• Sinkronisasi dua arah tidak didukung saat menggunakan change stream.

• Untuk kluster Amazon DocumentDB non-elastis, gunakan change stream: atur Migration Method ke ChangeStream dan Architecture ke Sharded Cluster.

Database baru:

• DTS tidak menyinkronkan data inkremental dari database yang dibuat setelah tugas sinkronisasi dimulai.

Operasi yang didukung

Operasi yang ditangkap bergantung pada metode migrasi.

Menggunakan oplog (direkomendasikan):

• CREATE COLLECTION, CREATE INDEX

• DROP DATABASE, DROP COLLECTION, DROP INDEX

• RENAME COLLECTION

• INSERT, UPDATE, dan DELETE tingkat dokumen

Menggunakan aliran perubahan:

• DROP DATABASE, DROP COLLECTION

• RENAME COLLECTION

• INSERT, UPDATE, dan DELETE tingkat dokumen

Penagihan

Sinkronisasi data inkremental dikenai biaya. Untuk detail harga, lihat Ikhtisar penagihan.

DTS menagih instans selama periode percobaan koneksi ulang.

Prasyarat

Sebelum memulai, pastikan Anda telah memiliki:

• Instans kluster sharded ApsaraDB for MongoDB yang sedang berjalan. Lihat Buat instans kluster sharded

• Layanan dan fungsi Function Compute, dengan Handler Type diatur ke Event Handler. Lihat Buat fungsi dengan cepat

• Akun database pada instans ApsaraDB for MongoDB sumber dengan izin baca pada database sumber, admin, dan local. Lihat Kelola izin pengguna database MongoDB

  Catatan

  Jika Anda menggunakan ChangeStream sebagai metode sinkronisasi inkremental, akun database sumber memerlukan izin baca Change Streams tingkat instans (seperti readAnyDatabase). Jika sumber adalah instans ApsaraDB for MongoDB dengan akun kustom, Anda juga harus memberikan izin baca akun tersebut pada database admin. Untuk detailnya, lihat Izin akun root yang ditentukan saat pembuatan instans.

Buat tugas sinkronisasi

Langkah 4: Pilih objek untuk disinkronkan

Di langkah Configure Objects, konfigurasikan hal berikut:

Parameter	Deskripsi
Synchronization Types	Tetap ke Incremental Data Synchronization. Tidak dapat diubah
Data Format	Tetap ke Canal Json. Untuk deskripsi field, lihat bagian Canal Json pada Format data topik kluster Kafka
Source Objects	Pilih database atau koleksi untuk disinkronkan, lalu klik untuk memindahkannya ke Selected Objects
Selected Objects	Tinjau objek yang dipilih. Klik untuk menghapus objek. Klik kanan objek untuk mengonfigurasi granularitas sinkronisasi (tingkat database atau koleksi)

Langkah selanjutnya

• Tulis penanganan event Function Compute Anda untuk memproses muatan Canal JSON. Gunakan field isDdl untuk membedakan logika DDL dan DML, serta deserialisasi string doc di setiap elemen data untuk mengakses field dokumen.

• Untuk mengurangi beban dokumen dengan field besar pada fungsi Anda, konfigurasikan filter ETL sebelum data mencapai fungsi. Lihat Konfigurasikan ETL dalam tugas migrasi data atau sinkronisasi data.

• Untuk memantau kesehatan sinkronisasi, konfigurasikan peringatan di Advanced Settings atau setelah tugas dibuat. Lihat Konfigurasikan pemantauan dan peringatan saat membuat tugas DTS.

• Jika tugas gagal, dukungan teknis DTS akan berusaha memulihkannya dalam waktu 8 jam. Selama pemulihan, tugas mungkin dimulai ulang dan parameter tugas (bukan parameter database) mungkin disesuaikan. Untuk daftar parameter yang mungkin dimodifikasi, lihat Modifikasi parameter instans.