db.collection.watch()

配置

db.collection.watch() は次のように、レプリカ セットおよびシャーディングされたクラスターの配置で利用できます。

• レプリカセットの場合、データを保持している任意のメンバーに対してdb.collection.watch()を発行できます。

• シャーディングされたクラスターの場合、 インスタンスでdb.collection.watch() mongosを発行する必要があります。

ストレージ エンジン

db.collection.watch()はWired Tiger ストレージ エンジンでのみ使用できます。

読み取り保証 (read concern)majority サポート

MongoDB 4.2 以降では、"majority" の読み取り保証 (read concern)のサポートに関係なく変更ストリームが利用可能になりました。つまり、変更ストリームを使用する際、読み取り保証 (read concern) のmajority サポートを有効にする（デフォルト）か無効にするかを選択できます。

MongoDB 4.0 以前では、変更ストリームは、"majority" 読み取り保証（read concern）サポートが有効（デフォルト）の場合にのみ使用できます。

再開可能性

MongoDBドライバーとは異なり、 mongoshはエラー後に変更ストリーム カーソルの再開を自動的に試行しません。 MongoDB ドライバーは、特定のエラーが発生後に 変更ストリーム カーソルの自動的な再開を1 回試行します。

db.collection.watch()は、oplog に保存されている情報を使用して変更イベントの説明を生成し、その操作に関連付けられた再開トークンを生成します。 resumeAfterまたはstartAfterオプションに渡される再開トークンによって識別される操作がすでにoplogから削除されている場合、 db.collection.watch()は変更ストリームを再開できません。

変更ストリームの再開の詳細については、「変更ストリームの再開」を参照してください。

アップデート 操作の完全なドキュメント検索

デフォルトでは、変更ストリーム カーソルはアップデート操作における特定のフィールドの変更またはデルタを返します。また、変更されたドキュメントのうち、過半数がコミットした最新のバージョンを検索して返すように変更ストリームを構成できます。アップデートと検索の間に他の書き込み操作が行われた場合、返されるドキュメントがアップデート実行時のドキュメントと大幅に異なる可能性があります。

アップデート操作中に適用された変更の数と完全なドキュメントのサイズによっては、アップデート操作における変更イベント ドキュメントのサイズが BSON ドキュメントの制限である 16 MB を超えるリスクがあります。サイズが超過した場合、サーバーで変更ストリーム カーソルが閉じられ、エラーが返されます。

変更ストリームを開く

次の操作は、 data.sensors コレクションに対し変更ストリーム カーソルを開きます。

watchCursor = db.getSiblingDB("data").sensors.watch()

カーソルを反復処理し、新しいイベントをチェックする。cursor.isClosed() メソッドを cursor.tryNext() メソッドと組み合わせて使用し、変更ストリーム カーソルが閉じられ、かつ最新のバッチにオブジェクトが残っていない場合にのみループが終了するようにします。

while (!watchCursor.isClosed()) {
  let next = watchCursor.tryNext()
  while (next !== null) {
    printjson(next);
    next = watchCursor.tryNext()
  }
}

変更ストリーム出力の詳細なドキュメントについては、 「変更イベント」を参照してください。

変更ストリームで updateLookup に fullDocument オプションを使用する

fullDocument オプションを "updateLookup" に設定すると、アップデートされた変更ストリーム イベントに関連するドキュメントについて、過半数がコミットした最新のバージョンを参照するよう、変更ストリーム カーソルに指示します。

次の操作は、 fullDocument : "updateLookup"オプションを使用して、data.sensors コレクションに対し変更ストリーム カーソルを開きます。

watchCursor = db.getSiblingDB("data").sensors.watch(
   [],
   { fullDocument : "updateLookup" }
)

カーソルを反復処理し、新しいイベントをチェックする。cursor.isClosed() メソッドを cursor.tryNext() メソッドと組み合わせて使用し、変更ストリーム カーソルが閉じられ、かつ最新のバッチにオブジェクトが残っていない場合にのみループが終了するようにします。

while (!watchCursor.isClosed()) {
  let next = watchCursor.tryNext()
  while (next !== null) {
    printjson(next);
    next = watchCursor.tryNext()
  }
}

どのアップデート操作も、変更イベントはドキュメント検索の結果を fullDocument フィールドに返します。

fullDocumentのアップデート出力の例については、「変更ストリーム アップデート イベント」を参照してください。

変更ストリーム出力の詳細なドキュメントについては、 「変更イベント」を参照してください。

変更ストリームにおけるドキュメントの変更前と変更後のイメージ

MongoDB 6.0 以降では、変更ストリーム イベントを使用して、変更前と変更後のドキュメントのバージョン（変更前とイメージと変更後のイメージ）を出力できます。

• 変更前のイメージとは、置換、更新、または削除される前のドキュメントです。挿入されたドキュメントには、変更前のイメージはありません。

• 変更後のイメージとは、挿入、置換、または更新された後のドキュメントです。削除されたドキュメントには、変更後のイメージはありません。

• db.createCollection() 、 create 、または collModを使用し、コレクションに対して changeStreamPreAndPostImages を有効にします。

変更ストリーム イベントにおいて、次の条件に当てはまる場合、変更前と変更後のイメージは使用できません。

• ドキュメントの更新または削除操作時に、コレクションにおいて有効になっていない場合。

• expireAfterSeconds で設定した、変更前と変更後のイメージ保持時間が経過した後に削除された場合。

  • 次の例では、クラスター全体でexpireAfterSecondsを100秒に設定します。

    use admin
    db.runCommand( {
       setClusterParameter:
          { changeStreamOptions: {
             preAndPostImages: { expireAfterSeconds: 100 }
          } }
    } )

  • 次の例では、expireAfterSeconds を含む現在の changeStreamOptions 設定を返します。

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

  • expireAfterSeconds を off に設定すると、デフォルトの保持ポリシーが適用されます。対応する変更ストリーム イベントがoplog から削除されるまで、変更前と変更後のイメージは保持されます。

  • 変更ストリーム イベントが oplog から削除されると、 expireAfterSeconds の変更前と変更後のイメージの保持時間にかかわらず、対応する変更前と変更後のイメージも削除されます。

その他の考慮事項

• 変更前と変更後のイメージを有効にすると、ストレージ容量が消費され、処理時間が増えます。変更前と変更後のイメージは、必要な場合のみ有効にしてください。

• 変更ストリーム イベントのサイズを 16 メガバイト未満に制限します。イベントのサイズを制限するには、次の方法があります。

  • ドキュメントのサイズを 8 MB に制限します。updateDescription のような他の変更ストリーム イベントのフィールドがそれほど大きくない場合、変更ストリーム出力で変更前と変更後のイメージを同時にリクエストできます。

  • updateDescription のような他の変更ストリーム イベントのフィールドがそれほど大きくない場合、ドキュメントの変更ストリーム出力では最大 16 MB の変更後のイメージのみをリクエストしてください。

  • 次の場合、ドキュメントの変更ストリーム出力では最大 16 MB の変更前のイメージのみをリクエストしてください。

    • ドキュメントのアップデートがドキュメントの構造または内容のごく一部にしか影響しない場合、そして

    • replace 変更イベントが発生しない場合。replace イベントには、常に変更後のイメージが含まれます。

• 変更前イメージをリクエストするには、db.collection.watch() で fullDocumentBeforeChange を required または whenAvailable に設定します。変更後イメージをリクエストするには、同じ方法で fullDocument を設定します。

• 変更前のイメージは config.system.preimages コレクションに書き込まれます。

  • config.system.preimages コレクションが大きくなる場合があります。コレクションのサイズを制限するには、前述のとおり、変更前のイメージに expireAfterSeconds 時間を設定します。

  • 変更前のイメージはバックグラウンド プロセスによって非同期で削除されます。

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

db.temperatureSensor.insertMany( [
   { "_id" : 0, "reading" : 26.1 },
   { "_id" : 1, "reading" : 25.9 },
   { "_id" : 2, "reading" : 24.3 },
   { "_id" : 3, "reading" : 22.4 },
   { "_id" : 4, "reading" : 24.6 }
] )

watchCursorFullDocumentBeforeChange = db.temperatureSensor.watch(
   [],
   { fullDocumentBeforeChange: "whenAvailable" }
)

while ( !watchCursorFullDocumentBeforeChange.isClosed() ) {
   if ( watchCursorFullDocumentBeforeChange.hasNext() ) {
      printjson( watchCursorFullDocumentBeforeChange.next() );
   }
}

db.temperatureSensor.updateOne(
   { _id: 2 },
   { $set: { reading: 22.1 } }
)

{
   "_id" : {
      "_data" : "82624B21...",
      "_typeBits" : BinData(0,"QA==")
   },
   "operationType" : "update",
   "clusterTime" : Timestamp(1649090957, 1),
   "ns" : {
      "db" : "test",
      "coll" : "temperatureSensor"
   },
   "documentKey" : {
      "_id" : 2
   },
   "updateDescription" : {
      "updatedFields" : {
         "reading" : 22.1
      },
      "removedFields" : [ ],
      "truncatedArrays" : [ ]
   },
   "fullDocumentBeforeChange" : {
      "_id" : 2,
      "reading" : 24.3
   }
}

watchCursorFullDocument = db.temperatureSensor.watch(
   [],
   { fullDocument: "whenAvailable" }
)

while ( !watchCursorFullDocument.isClosed() ) {
   if ( watchCursorFullDocument.hasNext() ) {
      printjson( watchCursorFullDocument.next() );
   }
}

db.temperatureSensor.updateOne(
   { _id: 1 },
   { $set: { reading: 29.5 } }
)

{
   "_id" : {
      "_data" : "8262474D...",
      "_typeBits" : BinData(0,"QA==")
   },
   "operationType" : "update",
   "clusterTime" : Timestamp(1648840090, 1),
   "fullDocument" : {
      "_id" : 1,
      "reading" : 29.5
   },
   "ns" : {
      "db" : "test",
      "coll" : "temperatureSensor"
   },
   "documentKey" : {
      "_id" : 1
   },
   "updateDescription" : {
      "updatedFields" : {
         "reading" : 29.5
      },
      "removedFields" : [ ],
      "truncatedArrays" : [ ]
   }
}

変更ストリームで集約パイプライン フィルターを使用する

次の操作は、集約パイプラインを使用して data.sensors コレクションに対する変更ストリーム カーソルを開き、insert イベントのみをフィルタリングします。

watchCursor = db.getSiblingDB("data").sensors.watch(
   [
      { $match : {"operationType" : "insert" } }
   ]
)

カーソルを反復処理し、新しいイベントをチェックする。cursor.isClosed() メソッドを cursor.hasNext() メソッドと組み合わせて使用し、変更ストリーム カーソルが閉じられ、かつ最新のバッチにオブジェクトが残っていない場合にのみループが終了するようにします。

while (!watchCursor.isClosed()){
   if (watchCursor.hasNext()){
      printjson(watchCursor.next());
   }
}

変更ストリーム カーソルは、 operationType が insert である変更イベントのみを返します。変更ストリームの出力における詳細なドキュメントについては、「変更イベント」を参照してください。

変更ストリームの再開

変更ストリーム カーソルによって返されるすべてのドキュメントには、_id フィールドとして再開トークンが含まれます。変更ストリームを再開するには、再開する変更イベントの _id ドキュメント全体を watch() の resumeAfter または startAfter オプションに渡します。

次の操作では、再開トークンを使用して data.sensors コレクションに対する変更ストリーム カーソルを再開します。再開トークンを生成した操作が、クラスターの oplog からロール オフされていないことを前提としています。

let watchCursor = db.getSiblingDB("data").sensors.watch();
let firstChange;
while (!watchCursor.isClosed()) {
   if (watchCursor.hasNext()) {
     firstChange = watchCursor.next();
     break;
   }
}
watchCursor.close();
let resumeToken = firstChange._id;
resumedWatchCursor = db.getSiblingDB("data").sensors.watch(
[],
   { resumeAfter : resumeToken }
)

カーソルを反復処理し、新しいイベントをチェックする。cursor.isClosed() メソッドを cursor.hasNext() メソッドと組み合わせて使用し、変更ストリーム カーソルが閉じられ、かつ最新のバッチにオブジェクトが残っていない場合にのみループが終了するようにします。

while (!resumedWatchCursor.isClosed()){
   if (resumedWatchCursor.hasNext()){
      print(resumedWatchCursor.next());
   }
}

変更ストリームの再開に関する詳細なドキュメントについては、「変更ストリームの再開」を参照してください。