ApsaraDB for MongoDB:MongoDB change streams を使用してリアルタイムでデータ変更をキャプチャする

より多くの DDL イベントをリッスンする

前提条件

MongoDB 6.0 以降 (アップグレードガイド)。

手順

1. mongosh を使用してデータベースに接続します。

2. watch コマンドを実行し、showExpandedEvents: true を設定します:

   // mongo shell または mongosh v1.x
   cursor = db.getSiblingDB("test").watch([],
     {
       showExpandedEvents: true       // より多くの DDL イベントのリッスンを有効にする
     }
   );
   cursor.next();

結果の検証

1. 新しい mongosh ウィンドウで、db.createCollection("myCollection1") などの変更コマンドを実行します。

2. 元の SQL ウィンドウの出力で SQL 実行の詳細を表示できます。

   

pre-image を有効にする

MongoDB の pre-image は、ドキュメントが変更または削除される前の完全なスナップショットです。変更前のデータの元の値を記録します。

前提条件

MongoDB 6.0 以降 (アップグレードガイド)。

手順

1. データベースレベルで pre-image を有効にします:

   db.adminCommand({
     setClusterParameter: {
       changeStreamOptions: {
         preAndPostImages: { expireAfterSeconds: "off" } // "off" は oplog の保存期間が使用されることを示します
       }
     }
   })

2. コレクションレベルで pre-image を有効にします:

   説明

   データベース内のすべてのコレクションで pre-image を有効にするには、まずデータベースレベルで有効にする必要があります。次に、各コレクションに対して個別にこの機能を有効にする必要があります。

   既存のコレクションの変更

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

   新しいコレクションの作成時に指定

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

3. change stream リスナーを作成し、pre-image オプションを指定します:

   // ターゲットコレクションにリスナーを作成
   cursor = db.getCollection("myCollection").watch([],
     {
       fullDocument: 'required', // または 'whenAvailable'
       fullDocumentBeforeChange: 'required' // または 'whenAvailable'
     }
   )
   cursor.next();

   • required: サーバーは pre-image または post-image を返す必要があります。そうでない場合、エラーが報告されます。

   • whenAvailable: サーバーはイメージを返そうとしますが、保証はされません。

結果の検証

1. データベースレベルで pre-image が有効になっているかどうかを確認します:

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

   有効な場合、次の出力が返されます:

2. コレクションの構成を確認します:

   db.getCollectionInfos({name: "myCollection"}) // または db.runCommand({listCollections: 1})

   期待される出力: 返されたドキュメントで、"options" : { "changeStreamPreAndPostImages" : { "enabled" : true } } のようなフィールドを探します。

   

3. 別の mongosh ウィンドウで、myCollection 内のドキュメントを更新します。

4. cursor によって返されたイベントを監視します。これには、pre-image である fullDocumentBeforeChange フィールドが含まれている必要があります。

   

詳細については、「Change Streams with Document Pre- and Post-Images」をご参照ください。

post-image を有効にする

MongoDB の post-image は、変更が発生した後のドキュメントの完全なスナップショットです。変更後のドキュメントの完全な内容を記録します。

前提条件

MongoDB 3.6 以降 (アップグレードガイド)。

手順

watch コマンドを実行するときに、fullDocument: 'updateLookup' を設定します。

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

結果の検証

1. 別の mongosh ウィンドウで、myCollection にドキュメントを追加または更新します。

2. cursor によって返されたイベントを監視します。これには、post-image である fullDocument フィールドが含まれている必要があります。

大規模な変更イベント (> 16 MB) の処理

前提条件

MongoDB 7.0 以降 (アップグレードガイド)。

手順

watch() パイプラインに $changeStreamSplitLargeEvent ステージを含めることができます:

myChangeStreamCursor = db.myCollection.watch(
  [ { $changeStreamSplitLargeEvent: {} } ], // 分割ステージを追加
  {
    fullDocument: "required",
    fullDocumentBeforeChange: "required"
  }
)

結果の検証

• 非常に大きな配列を含むドキュメントの更新など、16 MB を超える変更イベントを引き起こす操作を実行します。

• 返されたイベントストリームを監視します。それは複数の連続した fragment イベントに分割され、最後の fragment イベントで終わります。

pre-image のストレージオーバーヘッドを削減する

デフォルトでは、pre-image は oplog とともに有効期限が切れます。より短い有効期限を設定して、ストレージスペースを節約できます:

db.adminCommand({
  setClusterParameter: {
    changeStreamOptions: {
      preAndPostImages: { expireAfterSeconds: 100 } // 単位: 秒
    }
  }
})