MongoDB Change Streams

可用性

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

• ストレージエンジン

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

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

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

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

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

接続

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

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

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

コレクション、データベース、または配置の監視

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

変更ストリームのパフォーマンスに関する考慮事項

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

変更ストリームを開く

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

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

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

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

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

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

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

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

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

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

変更ストリーム出力の修正

更新操作のための完全なドキュメントの検索

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

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

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

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

変更ストリームの再開

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

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

npx mongodb-resumetoken-decoder <RESUME TOKEN>

{
   "_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")
    }
}

{
   "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 })
}

{
   "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" ] }

イベント通知

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

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

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

照合

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

Change Streams と孤立したドキュメント

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

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

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

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

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

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

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

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

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

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

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

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

    注意

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

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

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

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

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

その他の考慮事項

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

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

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

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

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

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

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

• 変更前イメージをリクエストするには、db.collection.watch() で、fullDocumentBeforeChange を required または 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 ロールが必要です。

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

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

クラスターの initialSyncMethod パラメータが fileCopyBased の場合、変更ストリームリスナーには影響しません。

initialSyncMethod が logical で、新しく同期されたノードで変更ストリームが開かれ、 論理的な最初の同期の完了より前の点でのイベントが読み取られる場合、変更前と変更後のイメージが欠落している可能性があります。