変更ストリーム

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

注意

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

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

可用性

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

ストレージエンジン
レプリカセットとシャーディングされたクラスターでは、WiredTiger ストレージエンジンを使用する必要があります。変更ストリームは、保管時の暗号化機能を使用する配置でも使用できます。
レプリカセットプロトコルバージョン
レプリカセットとシャーディングされたクラスターは、レプリカセットプロトコルバージョン 1（pv1）を使用する必要があります。
読み取り保証 (read concern) "過半数" 有効化
変更ストリームは、 "majority"の読み取り保証 (read concern) のサポートに関係なく使用できます。つまり、変更ストリームを使用するには、読み取り保証（read concern） majorityのサポートを有効にする（デフォルト）か無効にするかを選択できます。

注意

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

Tip

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

Stable API でのサポート

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

接続

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

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

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

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

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

ターゲット	説明
コレクション	単一のコレクション（`system` コレクション、または `admin` データベース、`local` データベース、`config` データベース内のコレクションを除く）の変更ストリームカーソルを開くことができます。このページの例では、MongoDB ドライバーを使用して、単一のコレクションの変更ストリームカーソルを開いて操作します。`mongosh` メソッド `db.collection.watch()` も参照してください。
データベース	単一データベース（`admin`、`local`、`config` データベースを除く）の変更ストリームカーソルを開いて、すべての非システムコレクションに対する変更を監視できます。 MongoDB ドライバーメソッドについては、「ドライバーのドキュメント」を参照してください。 `mongosh`メソッド`db.watch()`も参照してください。
配置	配置（レプリカセットまたはシャーディングされたクラスター）の変更ストリームカーソルを開き、すべてのデータベース（`admin`、`local`、`config` を除く）でシステム以外のコレクションに加えられた変更を監視できます。 MongoDB ドライバーメソッドについては、「ドライバーのドキュメント」を参照してください。 `mongosh`メソッド`Mongo.watch()`も参照してください。

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

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

シャーディングされたクラスターの考慮事項

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

mongos は、各シャードに個別の変更ストリームを作成します。この動作は、変更ストリームが特定のシャードキーの範囲を対象にしているかどうかに関わらず発生します。
mongos は変更ストリームの結果を受け取ると、その結果をソートおよびフィルタリングします。必要に応じて、mongos はfullDocument ルックアップも実行します。

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

変更ストリームを開く

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

レプリカセットの場合は、任意のデータ保持ノードから変更ストリームを開きます。
シャーディングされたクラスターの場合は、mongos から変更ストリームを開きます。

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

➤ 右上の言語の選択のドロップダウンメニューを使用して、このページの例の言語を設定します。

以下のC の例ではMongoDBレプリカセットに接続し、 inventoryコレクションを含むデータベースにアクセスしていることを前提としています。

mongoc_collection_t *collection;
bson_t *pipeline = bson_new ();
bson_t opts = BSON_INITIALIZER;
mongoc_change_stream_t *stream;
const bson_t *change;
const bson_t *resume_token;
bson_error_t error;
collection = mongoc_database_get_collection (db, "inventory");
stream = mongoc_collection_watch (collection, pipeline, NULL /* opts */);
mongoc_change_stream_next (stream, &change);
if (mongoc_change_stream_error_document (stream, &error, NULL)) {
   MONGOC_ERROR ("%s\n", error.message);
}
mongoc_change_stream_destroy (stream);

以下のC# の例では、 MongoDBレプリカセットに接続し、 inventoryコレクションを含むデータベースにアクセスしていることを前提としています。

var cursor = inventory.Watch();
while (cursor.MoveNext() && cursor.Current.Count() == 0) { } // keep calling MoveNext until we've read the first batch
var next = cursor.Current.First();
cursor.Dispose();

以下の Go の例では、MongoDBレプリカセットに接続し、inventory コレクションを含むデータベースにアクセスしていることを前提としています。

cs, err := coll.Watch(ctx, mongo.Pipeline{})
assert.NoError(t, err)
defer cs.Close(ctx)
ok := cs.Next(ctx)
next := cs.Current

以下の Java の例では、MongoDB レプリカセットに接続し、inventory コレクションを含むデータベースにアクセスしていることを前提としています。

MongoCursor<ChangeStreamDocument<Document>> cursor = inventory.watch().iterator();
ChangeStreamDocument<Document> next = cursor.next();

以下の Kotlin の例では、 MongoDB レプリカセットに接続し、かつinventoryコレクションを含むデータベースにアクセスできることを前提としています。これらのタスクの完了の詳細については、「 Kotlin コルーチンドライバーのデータベースとコレクション」のガイドを参照してください。

val job = launch {
    val changeStream = collection.watch()
    changeStream.collect {
        println("Received a change event: $it")
    }
}

以下の例では MongoDB レプリカセットに接続し、アクセスしているデータベースに inventory コレクションが含まれていることを前提としています。

cursor = db.inventory.watch()
document = await cursor.next()

以下の Node.js の例では、MongoDB レプリカセットに接続し、アクセスしているデータベースに inventory コレクションが含まれていることを前提としています。

次の例では、ストリームを使用して変更イベントを処理しています。

const collection = db.collection('inventory');
const changeStream = collection.watch();
changeStream
  .on('change', next => {
    // process next document
  })
  .once('error', () => {
    // handle error
  });

代わりに、イテレーターを使用して変更イベントを処理することもできます。

const collection = db.collection('inventory');
const changeStream = collection.watch();
const next = await changeStream.next();

changeStream により EventEmitter が拡張されます。

以下の例では、MongoDB レプリカセットに接続し、 inventory コレクションを含むデータベースにアクセスしていることを前提としています。

$changeStream = $db->inventory->watch();
$changeStream->rewind();
$firstChange = $changeStream->current();
$changeStream->next();
$secondChange = $changeStream->current();

以下の Python の例では、 MongoDB レプリカセットに接続し、かつinventoryコレクションを含むデータベースにアクセスしていることを前提としています。

cursor = db.inventory.watch()
next(cursor)

以下の例では、MongoDB レプリカセットに接続し、 inventory コレクションを含むデータベースにアクセスしていることを前提としています。

cursor = inventory.watch.to_enum
cursor.next

以下のSwift （Async）例ではMongoDBレプリカセットに接続し、アクセスしているデータベースに inventoryコレクションが含まれていることを前提としています。

let inventory = db.collection("inventory")
// Option 1: retrieve next document via next()
let next = inventory.watch().flatMap { cursor in
    cursor.next()
}
// Option 2: register a callback to execute for each document
let result = inventory.watch().flatMap { cursor in
    cursor.forEach { event in
        // process event
        print(event)
    }
}

以下の Swift（Sync）例では MongoDB レプリカセットに接続し、アクセスしているデータベースに inventory コレクションが含まれていることを前提としています。

let inventory = db.collection("inventory")
let changeStream = try inventory.watch()
let next = changeStream.next()

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

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

カーソルが明示的に閉じている。
無効化イベントが発生している（コレクションの削除や名前の変更など）。
MongoDBデプロイへの接続が閉じているか、タイムアウトになっている。詳細については、「動作」を参照してください。
配置がシャーディングされたクラスターの場合、シャードを削除すると、開いている変更ストリームのカーソルが閉じることがあります。閉じた変更ストリームのカーソルは完全に再開できない場合があります。

注意

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

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