$changeStreamSplitLargeEvent（集計ステージ）

このセクションのサンプルシナリオでは、 myCollectionという名前の新しいコレクションを持つ$changeStreamSplitLargeEventの使用を示します。

myCollectionを作成し、 16 MB 未満のデータを含むドキュメントを 1 つ挿入します。

db.myCollection.insertOne(   { _id: 0, largeField: "a".repeat( 16 * 1024 * 1024 - 1024 ) })

largeField 反復文字aが含まれています。

myCollectionに対してchangeStreamPreAndPostImagesを有効にします。これにより、変更ストリームは更新前（イメージ前）と更新後（イメージ後）のドキュメントを取得できます。

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

db.collection.watch()を使用して、 myCollectionへの変更を監視するための変更ストリーム カーソルを作成します。

myChangeStreamCursor = db.myCollection.watch(   [ { $changeStreamSplitLargeEvent: {} } ],   { fullDocument: "required", fullDocumentBeforeChange: "required" })

変更ストリーム イベント の場合、

• fullDocument: "required" には、ドキュメントの変更後のイメージが含まれます。

• fullDocumentBeforeChange: "required" には、ドキュメントの変更前のイメージが含まれます。

詳細については、$changeStream を参照してください。

myCollectionでドキュメントを更新します。これにより、ドキュメントの変更前と変更後のイメージを含む 変更ストリーム イベント も生成されます。

db.myCollection.updateOne(   { _id: 0 },   { $set: { largeField: "b".repeat( 16 * 1024 * 1024 - 1024 ) } })

largeField に繰り返しの文字bが含まれるようになりました。

next()メソッドを使用してmyChangeStreamCursorからフラグメントを取得し、そのフラグメントをfirstFragment 、 secondFragment 、 thirdFragmentという名前のオブジェクトに保存します。

const firstFragment = myChangeStreamCursor.next()const secondFragment = myChangeStreamCursor.next()const thirdFragment = myChangeStreamCursor.next()

firstFragment.splitEvent表示:

firstFragment.splitEvent

フラグメントの詳細を含む出力は次のようになります。

splitEvent: { fragment: 1, of: 3 }

同様に、 secondFragment.splitEventとthirdFragment.splitEventも次を返します。

splitEvent: { fragment: 2, of: 3 }splitEvent: { fragment: 3, of: 3 }

firstFragmentのオブジェクトキーを調べるには、次の手順に従います。

Object.keys( firstFragment )

出力:

[   '_id',   'splitEvent',   'wallTime',   'clusterTime',   'operationType',   'documentKey',   'ns',   'fullDocument']

firstFragment.fullDocumentのサイズをバイト単位で調べるには次のようにします。

bsonsize( firstFragment.fullDocument )

出力:

16776223

secondFragment には、サイズが約 16 MB のfullDocumentBeforeChange変更前のイメージが含まれています。 次の例は、 secondFragmentのオブジェクトキーを示しています。

Object.keys( secondFragment )

出力:

[ '_id', 'splitEvent', 'fullDocumentBeforeChange' ]

thirdFragment には、サイズが約 16 MB のupdateDescriptionフィールドが含まれています。 次の例は、 thirdFragmentのオブジェクトキーを示しています。

Object.keys( thirdFragment )

出力:

[ '_id', 'splitEvent', 'updateDescription' ]

MongoDB .NET/ C#ドライバーを使用して$changeStreamSplitLargeEvent ステージを集計パイプラインに追加するには、 PipelineDefinitionオブジェクトで UnionWith() メソッドを呼び出します。

次の例では、16 MB を超えるイベントをフラグメントに分割し、変更ストリームカーソルで順番に返すパイプラインステージを作成します。各フラグメントの splitEventフィールドには、フラグメントインデックスと合計数が表示されます。

[BsonIgnoreExtraElements]public class LargeDocument{    [BsonId]    public int Id { get; set; }    public string LargeField { get; set; } = null!;}

var client = new MongoClient("<connection-string>");
var db = client.GetDatabase("change_stream_test");var collection = db.GetCollection<LargeDocument>("largeEventCollection");
collection.InsertOne(new LargeDocument{    Id = 0,    LargeField = new string('a', 16 * 1024 * 1024 - 1024)});
db.RunCommand<BsonDocument>(new BsonDocument{    { "collMod", "largeEventCollection" },    { "changeStreamPreAndPostImages", new BsonDocument("enabled", true) }});
var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<LargeDocument>>()    .ChangeStreamSplitLargeEvent();
var watchOptions = new ChangeStreamOptions{    FullDocument = ChangeStreamFullDocumentOption.Required,    FullDocumentBeforeChange = ChangeStreamFullDocumentBeforeChangeOption.Required};
using var cursor = collection.Watch(pipeline, watchOptions);
collection.UpdateOne(    Builders<LargeDocument>.Filter.Eq(d => d.Id, 0),    Builders<LargeDocument>.Update.Set(d => d.LargeField, new string('b', 16 * 1024 * 1024 - 1024)));
while (cursor.MoveNext()){    foreach (var fragment in cursor.Current)    {        Console.WriteLine(fragment.BackingDocument["splitEvent"].ToJson());    }}

{ "fragment" : 1, "of" : 3 }{ "fragment" : 2, "of" : 3 }{ "fragment" : 3, "of" : 3 }

watch() メソッドと aggregate() メソッドの両方を使用して、 $changeStreamSplitLargeEvent操作を実行できます。MongoDB Collectionオブジェクトの watch() メソッドに集計パイプラインを渡すと、$changeStreamSplitLargeEvent は ChangeStreamCursor を返します。集計パイプラインを aggregate() メソッドに渡すと、$changeStreamSplitLargeEvent から が返されます。AggregationCursor

次の例では、16 MB を超えるイベントをフラグメントに分割し、ChangeStreamCursor で順番に返します。

const pipeline = [{ changeStreamSplitLargeEvent: {} }];
const changeStream = collection.watch(pipeline);return changeStream;