$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 オブジェクトで ChangeStreamSplitLargeEvent() メソッドを呼び出します。

次の例では、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;