AI エージェント向け: ドキュメントインデックスは https://www.mongodb.com/ja-jp/docs/llms.txt で利用できます。すべてのページの markdown バージョンは、いずれかの URL パスに .md を追加することで利用できます。
Docs Menu

MongoDB Change Streams

変更ストリームを使用すると、アプリケーションは oplog を手動で追尾する以前の複雑さやリスクなしで、リアルタイムのデータ変更にアクセスできます。アプリケーションは変更ストリームを使用して、単一のコレクション、データベース、または配置全体のすべてのデータ変更をサブスクライブし、即座に対応できます。変更ストリームは集計フレームワークを使用するため、アプリケーションは特定の変更をフィルタリングすることも、通知を任意に変換することもできます。

注意

変更ストリームはデータベースイベントに制限されます。Atlas Stream Processing には、複数のデータイベントタイプの管理や、Atlas データベースと同じクエリAPIを使用した複雑なデータの処理ストリームなどの拡張機能があります。詳しくは、Atlas Stream Processing を参照してください。

MongoDB 5.1 以降では、変更ストリームが最適化され、リソースの使用効率が上がり、一部の集計パイプライン ステージの実行が高速化されています。

変更ストリームは、レプリカセットシャーディングされたクラスターで使用できます。

  • ストレージエンジン

    レプリカセットとシャーディングされたクラスターでは、WiredTiger ストレージエンジンを使用する必要があります。変更ストリームは、保管時の暗号化機能を使用する配置でも使用できます。

  • レプリカセット プロトコル バージョン

    レプリカセットとシャーディングされたクラスターには、レプリカセット プロトコル バージョン 1(pv1)を使用する必要があります。

  • 読み取り保証 (read concern) "過半数" 有効化

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

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

注意

時系列コレクションは、ドキュメントレベルで変更を追跡する代わりに最適化されたストレージ形式を使用するため、変更ストリームをサポートしていません。時系列コレクションを Atlas Stream Processing のソースとして使用することはできません。

Tip

詳細については時系列コレクションの制限を参照してください。

変更ストリームはStable API V 1に含まれています。 ただし、 showExpandedEventsオプションは Stable API V 1に含まれていません。

変更ストリームの接続は、+srv 接続オプションを使用して DNS シードリスト、または接続文字列でサーバーを個別にリストする方法のいずれかを使用できます。

変更ストリームへの接続を失うか、接続が切断された場合、ドライバーは、読み込み設定 (read preference) が一致するクラスター内の別のノードを介して変更ストリームへの接続を再確立しようとします。ドライバーが正しく読み込み設定 (read preference) されているノードを見つけられない場合、例外が発生します。

詳細については、「接続文字列 URI 形式」を参照してください。

変更ストリームを以下に対して開くことができます。

ターゲット
説明

コレクション

単一のコレクション(system コレクション、または admin データベース、local データベース、config データベース内のコレクションを除く)の変更ストリーム カーソルを開くことができます。

The examples on this page include both MongoDB driver and mongosh examples for opening and working with a change stream cursor for a single collection.

データベース

単一データベース(adminlocalconfig データベースを除く)の変更ストリーム カーソルを開いて、すべての非システム コレクションに対する変更を監視できます。

For the MongoDB driver method, refer to your driver documentation. See also the mongosh method db.watch().

配置

配置(レプリカセットまたはシャーディングされたクラスター)の変更ストリーム カーソルを開き、すべてのデータベース(adminlocalconfig を除く)でシステム以外のコレクションに加えられた変更を監視できます。

For the MongoDB driver method, refer to your driver documentation. See also the mongosh method Mongo.watch().

データベースに対して開かれたアクティブな変更ストリームの量が接続プールのサイズを超えると、通知のレイテンシが発生する可能性があります。各変更ストリームは、次のイベントを待機している間、getMore 操作によって接続を開いたまま保持します。レイテンシを回避するには、プールサイズがオープン中の変更ストリーム数より大きいことを確認してください。詳細については、maxPoolSize の設定を参照してください。

変更ストリームをシャーディングされたクラスター上で開くとき

  • mongos は、各シャードに個別の変更ストリームを作成します。この動作は、変更ストリームが特定のシャードキーの範囲を対象にしているかどうかに関わらず発生します。

  • mongos は変更ストリームの結果を受け取ると、その結果をソートおよびフィルタリングします。必要に応じて、mongosfullDocument ルックアップも実行します。

最高のパフォーマンスを得るには、変更ストリームでの $lookup クエリの使用を制限します。

変更ストリームを開くには

  • レプリカセットの場合は、任意のデータ保持ノードから変更ストリームを開きます。

  • シャーディングされたクラスターの場合は、mongos から変更ストリームを開きます。

次の例では、コレクションの変更ストリームを開き、カーソルを反復処理して変更ストリーム ドキュメントを検索します。[1]

変更ストリーム カーソルを反復処理して変更イベントを取得します。変更ストリームのドキュメント形式については、「変更ストリーム イベント」を参照してください。

変更ストリーム カーソルは、次のいずれかが発生するまで開いたままになります。

  • カーソルが明示的に閉じている。

  • 無効化イベントが発生している(コレクションの削除や名前の変更など)。

  • MongoDBデプロイへの接続が閉じているか、タイムアウトになっている。詳細については、「動作」を参照してください。

  • 配置がシャーディングされたクラスターの場合、シャードを削除すると、開いている変更ストリームのカーソルが閉じてしまい、閉じた変更ストリームのカーソルが完全に再開できなくなることがあります。

注意

閉じられていないカーソルのライフサイクルは言語に依存します。

[1] startAtOperationTime を指定すると、特定の時点でカーソルを開くことができます。過去の開始点を指定する場合、oplog の時間範囲内である必要があります。

Tip

The _id field of the change stream event document act as the resume token. Do not use the pipeline to modify or remove the change stream event's _id field.

MongoDB 4.2 以降、変更ストリームの集計パイプライン でイベントの _id フィールドが変更される場合、変更ストリームで例外がスローされるようになります。

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

デフォルトでは、変更ストリームは更新操作中にフィールドのデルタのみを返します。ただし、更新されたドキュメントの過半数がコミットした最新のバージョンを返すように変更ストリームを構成できます。

updateLookup操作は、コレクションから、シャードキーとドキュメント識別子で識別されるドキュメントを読み取ります。コレクションは名前によって識別され、変更ストリームが処理される時点で存在するコレクションデータを使用します。次のシナリオを検討してみましょう。

  • コレクションの名前が変更された場合、ドキュメントは返されません。

  • コレクションの名前が変更され、古い名前で新しいコレクションが作成された場合、新しいコレクションに対してルックアップ操作が実行されます 。一致するドキュメントが見つかった場合は、それが返されます。

警告

迅速な削除やトラフィックの急増が発生している状況では、$match フィルターで fullDocument: "updateLookup" を構成すると、「再開トークンが見つかりません」エラーが発生する可能性があります。これは、ドキュメントの削除によって fullDocumentフィールドがnull 値を返す場合に発生します。これは、一致するドキュメントがないためであり、変更ストリームが再開トークンを見つけられなくなります。

代わりに、fullDocumentBeforeChange: "whenAvailable"fullDocument: "whenAvailable" の変更前と変更後のイメージを使用してください。「Change Streams with イメージ前ドキュメントとイメージ後ドキュメント」セクションを参照してください。

注意

過半数がコミットした操作で、更新操作、lookup に、更新されたドキュメントを変更した操作が 1 つ以上ある場合、返される完全なドキュメント全体は、更新操作時のドキュメントとは大幅に異なる可能性があります。

ただし、変更ストリーム ドキュメントに含まれるデルタには、その変更ストリーム イベントに適用された監視対象コレクションの変更が常に正しく記述されています。

次のいずれかに当てはまる場合、更新イベントの fullDocument フィールドが欠落していることがあります。

  • ドキュメントが削除された場合、または更新と検索の間にコレクションが削除された場合。

  • 更新によって、そのコレクションのシャードキーで、少なくとも1つのフィールドの値が変更された場合。

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

変更ストリームは、カーソルを開くときに、resumeAfter または startAfter のいずれかに再開トークンを指定することによって再開できます。

警告

再開トークンを使用して変更ストリームを再開する場合は、最初にトークンを生成したときと同じパイプラインとオプションを使用します。別の変更ストリームパイプラインまたは別のオプションを使用すると、予期しない動作が発生したり、データの整合性に悪影響を影響たり、変更ストリームが再開できなくなったりする可能性があります。

カーソルを開くときに再開トークンを resumeAfter に渡すことで、特定のイベントの後に変更ストリームを再開できます。

再開トークンの詳細については、「再開トークン」を参照してください。

重要

  • oplog には、トークンまたはタイムスタンプ(タイムスタンプが過去のものの場合)に関連付けられた操作を見つけるのに十分な履歴が必要です。

  • 無効化イベント(コレクションの削除や名前の変更など)によってストリームが閉じられた後は、resumeAfter を使用して変更ストリームを再開することはできません。代わりに、startAfter を使用して、無効化イベント後に新しい変更ストリームを開始できます。

You can start a new change stream after a specific event by passing a resume token to startAfter when opening the cursor. Unlike resumeAfter, startAfter can resume notifications after an invalidate event by creating a new change stream.

再開トークンの詳細については、「再開トークン」を参照してください。

重要

  • oplog には、トークンまたはタイムスタンプ(タイムスタンプが過去のものの場合)に関連付けられた操作を見つけるのに十分な履歴が必要です。

再開トークンには次の 2 種類があります。

  • イベント トークン: 特定の変更イベントを識別します。変更ストリームカーソルは、変更イベントが発生するたびにイベントトークンを生成します。

  • 高値トークン: 関連する 変更イベントのない点を表します。サーバーは定期的に高値トークンを生成し、変更イベントが発生していない場合でもクラスター時間が経過したことを示します。

Tip

サーバーは高値の再開トークンのタイムスタンプを定期的に増やします。書き込み (write) 頻度の低いアイドル シャードでは、ユースケースによっては、この高度が十分な頻度で発生しないことがあります。ハイライト タイムスタンプをより頻繁に進めるには、appendOplogNote コマンドを使用してアイドル シャード上のoplogに noop エントリを書込みます。

再開トークンは、複数のソースにあります。

ソース
説明

各変更イベントの通知には、_id フィールドに再開トークンが含まれています。

$changeStream 集計ステージには、cursor.postBatchResumeToken フィールドに再開トークンが含まれています。

このフィールドは、aggregate コマンドを使用しているときにのみ表示されます。

getMore コマンドには、cursor.postBatchResumeToken フィールドに再開トークンが含まれています。

MongoDB 4.2 以降、変更ストリームの集計パイプライン でイベントの _id フィールドが変更される場合、変更ストリームで例外がスローされるようになります。

Tip

MongoDB provides a "snippet", an extension to mongosh, that decodes hex-encoded resume tokens.

You can install and run the resumetoken snippet from mongosh:

snippet install resumetoken
decodeResumeToken('<RESUME TOKEN>')

You can also run resumetoken from the command line (without using mongosh) if npm is installed on your system:

npx mongodb-resumetoken-decoder <RESUME TOKEN>

下記についての詳細は、参照先をご覧ください。

変更イベントの通知には、_id フィールドに再開トークンが含まれています。

{
"_id": {
"_data": "82635019A0000000012B042C0100296E5A1004AB1154ACACD849A48C61756D70D3B21F463C6F7065726174696F6E54797065003C696E736572740046646F63756D656E744B65790046645F69640064635019A078BE67426D7CF4D2000004"
},
"operationType": "insert",
"clusterTime": Timestamp({ "t": 1666193824, "i": 1 }),
"collectionUUID": new UUID("ab1154ac-acd8-49a4-8c61-756d70d3b21f"),
"wallTime": ISODate("2022-10-19T15:37:04.604Z"),
"fullDocument": {
"_id": ObjectId("635019a078be67426d7cf4d2"'),
"name": "Giovanni Verga"
},
"ns": {
"db": "test",
"coll": "names"
},
"documentKey": {
"_id": ObjectId("635019a078be67426d7cf4d2")
}
}

aggregateコマンドを使用する場合、$changeStream 集計ステージには cursor.postBatchResumeToken フィールドに再開トークンが含まれています。

{
"cursor": {
"firstBatch": [],
"postBatchResumeToken": {
"_data": "8263515EAC000000022B0429296E1404"
},
"id": Long("4309380460777152828"),
"ns": "test.names"
},
"ok": 1,
"$clusterTime": {
"clusterTime": Timestamp({ "t": 1666277036, "i": 1 }),
"signature": {
"hash": Binary(Buffer.from("0000000000000000000000000000000000000000", "hex"), 0),
"keyId": Long("0")
}
},
"operationTime": Timestamp({ "t": 1666277036, "i": 1 })
}

getMore コマンドにも、cursor.postBatchResumeToken フィールドに再開トークンが含まれています。

{
"cursor": {
"nextBatch": [],
"postBatchResumeToken": {
"_data": "8263515979000000022B0429296E1404"
},
"id": Long("7049907285270685005"),
"ns": "test.names"
},
"ok": 1,
"$clusterTime": {
"clusterTime": Timestamp( { "t": 1666275705, "i": 1 } ),
"signature": {
"hash": Binary(Buffer.from("0000000000000000000000000000000000000000", "hex"), 0),
"keyId": Long("0")
}
},
"operationTime": Timestamp({ "t": 1666275705, "i": 1 })
}

変更ストリームは、データの変更が確実であることが確認された後に下流のシステムに通知することで、依存しているビジネスシステムを持つアーキテクチャにメリットをもたらすことができます。たとえば、変更ストリームは、データ抽出・変換・ロード(ETL)サービス、クロスプラットフォーム同期、コラボレーション機能、通知サービスを実装する際、開発者の時間を節約するのに役立ちます。

自己管理型配置認可に認証を強制する配置の場合:

  • 特定のコレクションに対して変更ストリームを開くには、対応するコレクションに対して changeStream アクションと find アクションを許可する権限がアプリケーションに必要です。

    { resource: { db: <dbname>, collection: <collection> }, actions: [ "find", "changeStream" ] }
  • 単一のデータベースに対して変更ストリームを開くには、データベース内のすべての非 system コレクションに対して changeStream アクションと find アクションを許可する権限がアプリケーションに必要です。

    { resource: { db: <dbname>, collection: "" }, actions: [ "find", "changeStream" ] }
  • 配置の全体に対して変更ストリームを開くには、配置内のすべてのデータベースのすべての非 system コレクションに対して changeStream アクションとfind アクションを許可する権限がアプリケーションに必要です。

    { resource: { db: "", collection: "" }, actions: [ "find", "changeStream" ] }

変更ストリームは、レプリカセット内のデータを保持しているノードの過半数に反映されたデータ変更についてのみ通知します。変更ストリームは、障害発生時にも永続性が保証された、過半数コミット済みの変更に対してのみ通知がトリガーされることを保証します。

たとえば、レプリカセットが3つのノードで構成されており、プライマリに対して変更ストリーム カーソルが開かれているとします。クライアントが挿入操作を実行した場合、変更ストリームは、挿入がデータを保持しているノードの過半数に保存された後でのみ、データ変更をアプリケーションに通知します。

操作がトランザクションに関連付けられている場合、変更イベント ドキュメントには txnNumberlsid が含まれます。

明示的な照合が指定されていない限り、変更ストリームはsimple バイナリ比較を使用します。

MongoDB 5.3 以降では、範囲の移行 中に、孤立したドキュメントの更新に対して 変更ストリームイベントは生成されません。

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

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

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

  • db.createCollection()create、または collMod を使用し、コレクションに対して changeStreamPreAndPostImages を有効にします。例、collMod コマンドを使用する場合は次のようになります。

    db.runCommand( {
    collMod: <collection>,
    changeStreamPreAndPostImages: { enabled: true }
    } )

To check the current changeStreamPreAndPostImages setting for a collection, run db.getCollectionInfos():

db.getCollectionInfos( {
name: "<collection>"
} )[0].options.changeStreamPreAndPostImages
{ changeStreamPreAndPostImages: { enabled: true } }

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

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

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

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

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

      注意

      setClusterParameter コマンドはMongoDB Atlasクラスターではサポートされていません。すべてのコマンドに対する Atlas のサポートの詳細については、「 Atlas でサポートされていないコマンド 」を参照してください。

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

      db.adminCommand( { getClusterParameter: "changeStreamOptions" } )
    • expireAfterSecondsoff に設定すると、デフォルトの保持ポリシーが適用されます。対応する変更ストリーム イベントがoplog から削除されるまで、変更前と変更後のイメージは保持されます。

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

その他の考慮事項

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

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

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

    • updateDescription のような他の変更ストリーム イベントのフィールドが大きくない場合、最大 16 メビバイトのドキュメントの変更ストリーム出力では、変更後のイメージのみをリクエストします。

    • 次の場合、16 メビバイトまでのドキュメントの変更ストリーム出力で、変更前のイメージのみをリクエストします。

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

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

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

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

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

    • config.system.preimages のサイズを監視するには、シャーディングされたクラスターか、レプリカセットの mongod ノードに接続します。次に、以下のコマンドを実行します。

      use config
      db.system.preimages.totalSize()
      db.system.preimages.stats()

      注意

      これらのコマンドを実行するには、config.system.preimages コレクションに対する collStats 権限アクションが必要です。

      MongoDB Atlas 配置でこれらのコマンドを実行するには、atlasAdmin ロールが必要です。

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

重要

下位互換性のない機能

MongoDB 6.0 以降では、変更ストリームにドキュメントの変更前のイメージと変更後のイメージを使用している場合、以前の MongoDB バージョンにダウングレードする前に、collMod コマンドを使用して各コレクションの changeStreamPreAndPostImages を無効にする必要があります。

Tip

変更ストリーム出力の完全な例については、「Change Streams とドキュメントの変更前イメージおよび変更後イメージ」を参照してください。