ApsaraDB for MongoDB:Tangkap perubahan data secara real time menggunakan MongoDB change streams

Dengarkan event DDL tambahan

Prasyarat

MongoDB 6.0 atau yang lebih baru (Panduan peningkatan).

Prosedur

1. Gunakan mongosh untuk menghubungkan ke database.

2. Jalankan perintah watch dengan showExpandedEvents: true:

   // mongo shell atau mongosh v1.x
   cursor = db.getSiblingDB("test").watch([],
     {
       showExpandedEvents: true       // Aktifkan pendengaran untuk lebih banyak event DDL
     }
   );
   cursor.next();

Verifikasi Hasil

1. Di jendela SQL baru, jalankan pernyataan perubahan, seperti db.createCollection("myCollection1").

2. Periksa jendela SQL asli untuk output terkait pernyataan yang dieksekusi.

   mg xxx test> cursor = db.getSiblingDB("test").watch([],
   ... {
   ...     showExpandedEvents: true     // Aktifkan pendengaran event DDL tambahan
   ... }
   ... );
   ... cursor.next();
   Warning: Jika tidak ada dokumen dalam batch, next akan block. Gunakan tryNext jika Anda ingin memeriksa apakah ada dokumen tanpa menunggu.
   {
     _id: {
       _data: '8268777614000000012B042C0100296E5A100434919C64814940F3A61F9D97A3F60D5C463C6Fxxx03C63726561746500466F70657274696F6E4465737372697074696F6E0046466964496E6465787800461E76002B04466B657990046E5F6964002B02003C6E616D65003C5F69645F000000004'
     },
     operationType: 'create',
     clusterTime: Timestamp({ t: 1752659476, i: 1 }),
     collectionUUID: UUID('34919c64-8149-40f3-a61f-9xxx'),
     wallTime: ISODate('2025-07-16T09:51:16.043Z'),
     ns: { db: 'test', coll: 'myCollection1' },
     operationDescription: { idIndex: { v: 2, key: { _id: 1 }, name: '_id_' } }
   }

Aktifkan pre-image

Pre-image di MongoDB adalah snapshot lengkap dari dokumen sebelum dimodifikasi atau dihapus. Pre-image mencatat nilai asli sebelum perubahan terjadi.

Prasyarat

MongoDB 6.0 atau yang lebih baru (Panduan peningkatan).

Prosedur

1. Aktifkan pre-image di tingkat database:

   db.adminCommand({
     setClusterParameter: {
       changeStreamOptions: {
         preAndPostImages: { expireAfterSeconds: "off" } // "off" menggunakan periode retensi oplog
       }
     }
   })

2. Aktifkan pre-image di tingkat koleksi:

   Catatan

   Untuk mengaktifkan pre-image untuk semua koleksi dalam database, pertama-tama aktifkan di tingkat database. Kemudian aktifkan secara terpisah untuk setiap koleksi dalam database tersebut.

   Modifikasi koleksi yang sudah ada

   db.runCommand({
     collMod: "myCollection",
     changeStreamPreAndPostImages: { enabled: true }
   })

   Tentukan saat membuat koleksi baru

   db.createCollection("myCollection", { changeStreamPreAndPostImages: { enabled: true }})

3. Buat listener change stream (tentukan opsi pre-image):

   // Buat listener pada koleksi target
   cursor = db.getCollection("myCollection").watch([],
     {
       fullDocument: 'required', // atau 'whenAvailable'
       fullDocumentBeforeChange: 'required' // atau 'whenAvailable'
     }
   )
   cursor.next();

   • required: Server harus mengembalikan pre/post-image. Jika tidak, server mengembalikan error.

   • whenAvailable: Server mencoba mengembalikan image tetapi tidak menjaminnya.

Verifikasi Hasil

1. Periksa apakah pengaturan tingkat database diaktifkan:

   db.adminCommand( { getClusterParameter: "changeStreamOptions" } )

   Jika berhasil, perintah mengembalikan output seperti berikut:

   test> db.adminCommand( { getClusterParameter: "changeStreamOptions" } )
   {
     clusterParameters: [
       {
         _id: 'changeStreamOptions',
         clusterParameterTime: Timestamp({ t: 1752655937, i: 1 }),
         preAndPostImages: { expireAfterSeconds: Long('100') }
       }
     ],
     ok: 1,
     '$clusterTime': {
       clusterTime: Timestamp({ t: 1752656717, i: 1 }),
       signature: {
         hash: Binary.createFromBase64('xxx=', 0),
         keyId: Long('xxx')
       }
     },
     operationTime: Timestamp({ t: 1752656717, i: 1 })
   }

2. Periksa konfigurasi koleksi:

   db.getCollectionInfos({name: "myCollection"}) // atau db.runCommand({listCollections: 1})

   Output yang diharapkan: Temukan bidang serupa dengan "options" : { "changeStreamPreAndPostImages" : { "enabled" : true } } dalam dokumen yang dikembalikan.

   test> db.getCollectionInfos({name: "myCollection"})
   [
     {
       name: 'myCollection',
       type: 'collection',
       options: { changeStreamPreAndPostImages: { enabled: true } },
       info: {
         readOnly: false,
         uuid: UUID('55ed1b7c-7575-4ba3-8afa-xxx')
       },
       idIndex: { v: 2, key: { _id: 1 }, name: '_id_' }
     }
   ]

3. Di jendela mongosh lain, perbarui dokumen dalam myCollection.

4. Amati event yang dikembalikan oleh cursor. Event tersebut harus mencakup bidang fullDocumentBeforeChange (dokumen sebelum perubahan).

   ... cursor.next();
   {
     _id: {
       _data: '8268786B15000000012B042C0100296E5A100455ED1B7C75754BA38AFA0C5B56A360BC463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B65790046xxx697CDDF6EE279xxx'
     },
     operationType: 'update',
     clusterTime: Timestamp({ t: 1752722197, i: 1 }),
     wallTime: ISODate('2025-07-17T03:16:37.626Z'),
     fullDocument: {
       _id: ObjectId('6878697cddfxxx'),
       name: 'test',
       count: 111
     },
     ns: { db: 'test', coll: 'myCollection' },
     documentKey: { _id: ObjectId('6878697cddfxxx') },
     updateDescription: {
       updatedFields: { count: 111 },
       removedFields: [],
       truncatedArrays: []
     },
     fullDocumentBeforeChange: { _id: ObjectId('6878697cddf6xxx'), name: 'test' }
   }

Untuk informasi lebih lanjut, lihat Change Streams with Document Pre- and Post-Images.

Aktifkan post-image

Post-image di MongoDB adalah snapshot lengkap dari dokumen setelah perubahan terjadi. Post-image mencatat konten dokumen lengkap setelah perubahan.

Prasyarat

MongoDB 3.6 atau yang lebih baru (Panduan peningkatan).

Prosedur

Saat menjalankan perintah watch, atur fullDocument: 'updateLookup'.

cursor = db.getSiblingDB("test").myCollection.watch([], 
  {
    fullDocument: 'updateLookup'
  }
);
cursor.next();
              

Verifikasi Hasil

1. Di jendela mongosh lain, sisipkan atau perbarui dokumen dalam myCollection.

2. Amati event yang dikembalikan oleh cursor. Event tersebut harus mencakup bidang fullDocument (dokumen setelah perubahan).

   xxx [primary] test> cursor = db.getSiblingDB("test").myCollection.watch([],
   ... {
   ...     fullDocument: 'updateLookup'
   ... }
   ... );
   ... cursor.next();
   [...
   {
     _id: {
       _data: 'xxx100296E5A10040D6C3FBC28484F08240555576066945463C6F7065726174696F6E54797065003C757064617465500046646F63756D656E744B6579005F6964004B657390046645F696400646881ADE741E7D4B638ED7CA1000004'
     },
     operationType: 'update',
     clusterTime: Timestamp({ t: 1753329166, i: 1 }),
     wallTime: ISODate('2025-07-24T03:52:46.966Z'),
     fullDocument: {
       _id: ObjectId('xxx8ed7ca1'),
       name: 'test1',
       age: 12,
       count: 2222
     },
     ns: { db: 'test', coll: 'myCollection' },
     documentKey: { _id: ObjectId('xxx7ca1') },
     updateDescription: {
       updatedFields: { count: 2222 },
       removedFields: [],
       truncatedArrays: []
     }
   }]

Atasi event perubahan yang sangat besar (>16 MB)

Prasyarat

MongoDB 7.0 atau yang lebih baru (Panduan peningkatan).

Prosedur

Sertakan tahap $changeStreamSplitLargeEvent dalam pipeline perintah watch():

myChangeStreamCursor = db.myCollection.watch(
  [ { $changeStreamSplitLargeEvent: {} } ], // Tambahkan tahap split
  {
    fullDocument: "required",
    fullDocumentBeforeChange: "required"
  }
)

Verifikasi Hasil

• Lakukan operasi yang menghasilkan event perubahan lebih besar dari 16 MB (misalnya, memperbarui dokumen yang berisi array sangat besar).

• Amati bahwa aliran event yang dikembalikan dibagi menjadi beberapa event fragment berturut-turut, diakhiri dengan event fragment terakhir.

Kurangi overhead penyimpanan pre-image

Secara default, pre-image kedaluwarsa bersamaan dengan oplog. Atur waktu kedaluwarsa yang lebih singkat untuk menghemat ruang:

db.adminCommand({
  setClusterParameter: {
    changeStreamOptions: {
      preAndPostImages: { expireAfterSeconds: 100 } // Satuan: detik
    }
  }
})

1. Mengapa log lambat change stream selalu menampilkan COLLSCAN?

Perilaku ini normal dan diharapkan. Tidak diperlukan optimasi. Kursor change stream pada akhirnya membaca dari local.oplog.rs (satu-satunya tempat yang mencatat semua modifikasi instans). Koleksi ini tidak memiliki indeks, sehingga selalu melakukan COLLSCAN. Hal ini tidak dapat dihindari dan tidak memberikan ruang untuk optimasi.

Jika Anda menggunakan COLLSCAN sebagai kata kunci untuk memfilter log lambat saat mencari kueri yang perlu dioptimalkan, saring juga kata kunci $changeStream.

Hanya selidiki log lambat terkait change stream jika Anda mengalami masalah performa (misalnya, peningkatan latensi dalam konsumsi downstream).

2. Mengapa, pada instans sharded, saya melihat kursor change stream pada shard selain shard utama saat mendengarkan koleksi unsharded?

Ini mempersiapkan kemungkinan operasi shardCollection. Anda dapat mengonversi koleksi unsharded dengan listener aktif menjadi koleksi sharded kapan saja, mendistribusikan datanya ke seluruh shard (mirip dengan movePrimary). Membuat kursor change stream pada shard lain sebelumnya menangani skenario ini. Biasanya, kursor ini pada shard lain tidak mengembalikan event perubahan dan menimbulkan overhead performa minimal.

Demikian pula, untuk menangani kemungkinan operasi addShard/removeShard, saat Anda menyiapkan listener pada instans sharded, mongos juga membuat kursor yang sesuai pada Config Server.

3. Mengapa log lambat untuk kursor change stream muncul di node primary meskipun saya menentukan readPreference:secondary saat membuat kursor?

Kursor menjadi "tersemat" ke node tertentu setelah dibuat dan tidak secara otomatis bermigrasi saat peran node berubah. Setelah kursor change stream dibuat, kursor tersebut tetap berada pada node mongod tertentu (primary atau secondary) dan terus mengonsumsi melalui getMore. Setelah event seperti alih bencana primary/secondary, pengubahan ukuran instans, atau migrasi, kursor yang awalnya berada di secondary mungkin berakhir di primary.

Jika Anda tidak ingin beban ini tetap berada di node primary, bersihkan kursor change stream terkait menggunakan killCursors. Logika konsumen downstream Anda kemudian dapat memulihkan kursor ke node secondary menggunakan resumeToken dan readPreference. Operasi ini tidak menyebabkan kehilangan atau gangguan event.

db.runCommand(
   {
     killCursors: <collection>,
     cursors: [ <cursor id1>, ... ], comment: <any>
   }
)
db.getSiblingDB("<testDB>").runCommand( { killCursors: "<testColl>", cursors: [NumberLong("2452840976689696187") ] } ) 

4. Mengapa saya melihat banyak log lambat tentang change streams di mongos yang berlangsung 1000 ms?

2020-08-26T04:34:45.045+0000 I COMMAND [conn21283] command altconfig-b2b-perf.oplog command: getMore \{ getMore: 3513599116181216748, collection: "oplog", $db: "altconfig-b2b-perf", $clusterTime: { clusterTime: Timestamp(1598416483, 1), signature: { hash: BinData(0, EC3841EB1FB7A34F897688BB5983E32E2ADF6763), keyId: 6855385090000683010 } }, lsid: \{ id: UUID("72864d11-c0f8-48bb-823e-de5f18a9c409") } } originatingCommand: \{ find: "oplog", filter: { timestamp: { $gte: new Date(1597895390009) } }, tailable: true, awaitData: true, $db: "altconfig-b2b-perf", $clusterTime: \{ clusterTime: Timestamp(1597895487, 1), signature: { hash: BinData(0, AAAE33D5688F935C70469CBDB8EB1F6882749A40), keyId: 6855385090000683010 } }, lsid: \{ id: UUID("72864d11-c0f8-48bb-823e-de5f18a9c409") } } nShards:1 cursorid:3513599116181216748 numYields:0 nreturned:0 reslen:237 protocol:op_msg 1000ms

Ini adalah perilaku normal di versi MongoDB sebelum 6.0. Ini menunjukkan timeout tunggu, bukan bottleneck performa. Dalam arsitektur kluster sharded, mongos membuat kursor change stream pada shard dengan tailable:true, awaitData:true, dan maxTimeMS:1000 untuk mengembalikan sebanyak mungkin event perubahan dalam 1000 ms.

Di versi utama sebelum 6.0, mongos mencatat log lambat 1000 ms ini, yang dapat menyesatkan pengguna. MongoDB telah mengoptimalkan perilaku ini. Untuk detailnya, lihat SERVER-50559.