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

Apa yang diterima fungsi Anda

Setiap pemanggilan meneruskan 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
  }]
}

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

// Single insert
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 merupakan 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 berjalan.

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

• Jika penyeimbang (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, atau Smart Access Gateway atau Cloud Enterprise Network (CEN).

Tulis kendala:

• 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 kesalahan 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.

Pembatasan 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

Buat tugas sinkronisasi

Langkah 4: Pilih objek untuk disinkronkan

Pada langkah Configure Objects, konfigurasi 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 dalam 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, konfigurasi filter ETL sebelum data mencapai fungsi. Lihat Konfigurasi ETL dalam tugas migrasi atau sinkronisasi data.

• Untuk memantau kesehatan sinkronisasi, konfigurasi peringatan di Advanced Settings atau setelah tugas dibuat. Lihat Konfigurasi 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.