/ /

データの変更を監視

Overview

このガイドでは、変更ストリームを使用してデータに対するリアルタイムの変更を監視する方法を学習できます。変更ストリームは、アプリケーションがコレクション、データベース、または配置のデータ変更をサブスクライブできる MongoDB Server の機能です。

Tip

Atlas Stream Processing

変更ストリームの代わりに、Atlas Stream Processing を使用してデータのストリームを処理および変換できます。データベースイベントのみを登録する変更ストリームとは異なり、Atlas Stream Processing は複数のデータイベント型を管理し、拡張データプロセシング機能を提供します。この機能の詳細については、 MongoDB AtlasドキュメントのAtlas Stream Processingを参照してください。

サンプルデータ

このガイドの例では、Atlasサンプルデータセットの sample_restaurants.restaurantsコレクションを使用します。MongoDB Atlasクラスターを無料で作成して、サンプルデータセットをロードする方法については、「 .NET/ C#ドライバーを使い始める」を参照してください。

このページの例では、次の Restaurant クラス、Address クラス、GradeEntry クラスをモデルとして使用します。

public class Restaurant
{
    public ObjectId Id { get; set; }
    public string Name { get; set; }
    [BsonElement("restaurant_id")]
    public string RestaurantId { get; set; }
    public string Cuisine { get; set; }
    public Address Address { get; set; }
    public string Borough { get; set; }
    public List<GradeEntry> Grades { get; set; }
}

public class Address
{
    public string Building { get; set; }
    [BsonElement("coord")]
    public double[] Coordinates { get; set; }
    public string Street { get; set; }
    [BsonElement("zipcode")]
    public string ZipCode { get; set; }
}

public class GradeEntry
{
    public DateTime Date { get; set; }
    public string Grade { get; set; }
    public float? Score { get; set; }
}

注意

restaurantsコレクションのドキュメントは、スニペットケースの命名規則を使用します。このガイドの例では、ConventionPack を使用してコレクション内のフィールドをパスカルケースに逆シリアル化し、Restaurantクラスのプロパティにマップします。

カスタム直列化について詳しくは、「カスタム直列化」を参照してください。

変更ストリームを開く

変更ストリームを開くには、メソッドまたはWatch() WatchAsync()メソッドを呼び出します。メソッドを呼び出すインスタンスによって、変更ストリームがリッスンするイベントの範囲が決まります。 Watch()WatchAsync()次のクラスでメソッドまたはメソッドを呼び出すことができます。

MongoClient: MongoDB 配置のすべての変更を監視
Database: データベース内のすべてのコレクションの変更を監視するには
Collection: コレクションの変更をモニターするには

次の例では、 restaurantsコレクションの変更ストリームを開き、変更が発生に応じて出力します。 SynchronousAsynchronous対応するコードを表示するには、タブまたはタブを選択します。

var database = client.GetDatabase("sample_restaurants");
var collection = database.GetCollection<Restaurant>("restaurants");
// Opens a change stream and prints the changes as they're received
using (var cursor = collection.Watch())
{
    foreach (var change in cursor.ToEnumerable())
    {
        Console.WriteLine("Received the following type of change: " + change.BackingDocument);
    }
}

var database = client.GetDatabase("sample_restaurants");
var collection = database.GetCollection<Restaurant>("restaurants");
// Opens a change streams and print the changes as they're received
using var cursor = await collection.WatchAsync();
await cursor.ForEachAsync(change =>
{
    Console.WriteLine("Received the following type of change: " + change.BackingDocument);
});

変更の監視を開始するには、アプリケーションを実行します。次に、別のアプリケーションまたは shell で、 restaurantsコレクションを変更します。 "name"の値が"Blarney Castle"であるドキュメントを更新すると、次の変更ストリーム出力が生成されます。

{ "_id" : { "_data" : "..." }, "operationType" : "update", "clusterTime" : Timestamp(...),
"wallTime" : ISODate("..."), "ns" : { "db" : "sample_restaurants", "coll" : "restaurants" },
"documentKey" : { "_id" : ObjectId("...") }, "updateDescription" : { "updatedFields" : { "cuisine" : "Irish" },
"removedFields" : [], "truncatedArrays" : [] } }

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

変更ストリーム出力を変更するには、パラメータをメソッドとメソッドに渡します。pipelineWatch()WatchAsync()このパラメーターを使用すると、指定された変更イベントのみを監視できます。 EmptyPipelineDefinitionクラスを使用し、関連する集計ステージメソッドを追加して、パイプラインを作成します。

pipelineパラメーターでは次の集計ステージを指定できます。

$addFields
$changeStreamSplitLargeEvent
$match
$project
$replaceRoot
$replaceWith
$redact
$set
$unset

Tip

PipelineDefinitionBuilderクラスを使用して集計パイプラインを構築する方法については、「ガイドによる操作の集計パイプラインステージ」を参照してください。

To learn more about modifying your change stream output, see the Modify Change Stream Output section in the MongoDB Server manual.

更新イベントの監視例

次の例では、 pipelineパラメータを使用して、アップデート操作のみを記録する変更ストリームを開きます。 SynchronousAsynchronous対応するコードを表示するには、タブまたはタブを選択します。

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
            .Match(change => change.OperationType == ChangeStreamOperationType.Update);
// Opens a change streams and print the changes as they're received
using (var cursor = collection.Watch(pipeline))
{
    foreach (var change in cursor.ToEnumerable())
    {
        Console.WriteLine("Received the following change: " + change);
    }
}

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
            .Match(change => change.OperationType == ChangeStreamOperationType.Update);
// Opens a change stream and prints the changes as they're received
using (var cursor = await collection.WatchAsync(pipeline))
{
    await cursor.ForEachAsync(change =>
    {
        Console.WriteLine("Received the following change: " + change);
    });
}

大規模な変更イベントの分裂例

アプリケーションが生成した変更イベントが16 MB を超えるサイズの場合、サーバーはBSONObjectTooLarge エラーを返します。このエラーを回避するには、$changeStreamSplitLargeEventパイプラインステージを使用してイベントを小さなフラグメントに分裂。 .NET/ C#ドライバー集計API には ChangeStreamSplitLargeEvent() メソッドが含まれており、このメソッドを使用して $changeStreamSplitLargeEvent ステージを変更ストリームパイプラインに追加できます。

この例では、 16 MB の制限を超える変更を監視し、変更イベントを分裂にドライバーに指示します。このコードは、各イベントの変更ドキュメントを出力し、ヘルパーメソッドを呼び出してイベントフラグメントを再アセンブルします。

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .ChangeStreamSplitLargeEvent();
using var cursor = collection.Watch(pipeline);
foreach (var completeEvent in GetNextChangeStreamEvent(cursor.ToEnumerable().GetEnumerator()))
{
    Console.WriteLine("Received the following change: " + completeEvent.BackingDocument);
}

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .ChangeStreamSplitLargeEvent();
using var cursor = await collection.WatchAsync(pipeline);
await foreach (var completeEvent in GetNextChangeStreamEventAsync(cursor))
{
    Console.WriteLine("Received the following change: " + completeEvent.BackingDocument);
}

注意

前述の例に示すように、変更イベントフラグメントを再アセンブルすることをお勧めしますが、この手順は任意です。同じロジックを使用して、分裂と完了した変更イベントの両方を監視できます。

上記の例では、GetNextChangeStreamEvent()、GetNextChangeStreamEventAsync()、MergeFragment() メソッドを使用して、変更イベントフラグメントを単一の変更ストリームドキュメントに再アセンブルします。次のコードは、これらのメソッドを定義します。

// Fetches the next complete change stream event
private static IEnumerable<ChangeStreamDocument<TDocument>> GetNextChangeStreamEvent<TDocument>(
IEnumerator<ChangeStreamDocument<TDocument>> changeStreamEnumerator)
{
    while (changeStreamEnumerator.MoveNext())
    {
        var changeStreamEvent = changeStreamEnumerator.Current;
        if (changeStreamEvent.SplitEvent != null)
        {
            var fragment = changeStreamEvent;
            while (fragment.SplitEvent.Fragment < fragment.SplitEvent.Of)
            {
                changeStreamEnumerator.MoveNext();
                fragment = changeStreamEnumerator.Current;
                MergeFragment(changeStreamEvent, fragment);
            }
        }
        yield return changeStreamEvent;
    }
}
// Merges a fragment into the base event
private static void MergeFragment<TDocument>(
    ChangeStreamDocument<TDocument> changeStreamEvent,
    ChangeStreamDocument<TDocument> fragment)
{
    foreach (var element in fragment.BackingDocument)
    {
        if (element.Name != "_id" && element.Name != "splitEvent")
        {
            changeStreamEvent.BackingDocument[element.Name] = element.Value; 
        }
    }
}

// Fetches the next complete change stream event
private static async IAsyncEnumerable<ChangeStreamDocument<TDocument>> GetNextChangeStreamEventAsync<TDocument>(
    IAsyncCursor<ChangeStreamDocument<TDocument>> changeStreamCursor)
{
    var changeStreamEnumerator = GetNextChangeStreamEventFragmentAsync(changeStreamCursor).GetAsyncEnumerator();
    while (await changeStreamEnumerator.MoveNextAsync())
    {
        var changeStreamEvent = changeStreamEnumerator.Current;
        if (changeStreamEvent.SplitEvent != null)
        {
            var fragment = changeStreamEvent;
            while (fragment.SplitEvent.Fragment < fragment.SplitEvent.Of)
            {
                await changeStreamEnumerator.MoveNextAsync();
                fragment = changeStreamEnumerator.Current;
                MergeFragment(changeStreamEvent, fragment);
            }
        }
        yield return changeStreamEvent;
    }
}
private static async IAsyncEnumerable<ChangeStreamDocument<TDocument>> GetNextChangeStreamEventFragmentAsync<TDocument>(
    IAsyncCursor<ChangeStreamDocument<TDocument>> changeStreamCursor)
{
    while (await changeStreamCursor.MoveNextAsync())
    {
        foreach (var changeStreamEvent in changeStreamCursor.Current)
        {
            yield return changeStreamEvent;
        }
    }
}
// Merges a fragment into the base event
private static void MergeFragment<TDocument>(
    ChangeStreamDocument<TDocument> changeStreamEvent,
    ChangeStreamDocument<TDocument> fragment)
{
    foreach (var element in fragment.BackingDocument)
    {
        if (element.Name != "_id" && element.Name != "splitEvent")
        {
            changeStreamEvent.BackingDocument[element.Name] = element.Value;
        }
    }
}

Tip

大規模な変更イベントの分割の詳細については、 MongoDB Serverマニュアルの $changeStreamSplitLargeEvent を参照してください。

`Watch()` の動作を変更する

Watch()メソッドとWatchAsync()メソッドは、操作を構成するために使用できるオプションを表す任意のパラメーターを受け入れます。オプションを指定しない場合、ドライバーは操作をカスタマイズしません。

次の表では、 Watch()とWatchAsync()の動作をカスタマイズするために設定できるオプションについて説明します。

オプション	説明
`FullDocument`	Specifies whether to show the full document after the change, rather than showing only the changes made to the document. To learn more about this option, see Include Pre-Images and Post-Images.
`FullDocumentBeforeChange`	Specifies whether to show the full document as it was before the change, rather than showing only the changes made to the document. To learn more about this option, see Include Pre-Images and Post-Images.
`ResumeAfter`	`Watch()` または `WatchAsync()` に、再開トークンで指定された操作の後に変更の返回を再開するように指示します。各変更ストリームイベントドキュメントには、`_id` フィールドとして再開トークンが含まれます。変更後に再開する操作を表す変更イベントドキュメントの `_id` フィールド全体を渡します。 `ResumeAfter` は `StartAfter` および `StartAtOperationTime` と排他的です。
`StartAfter`	`Watch()` または `WatchAsync()` に指示して、再開トークンで指定された操作の後に新しい変更ストリームを開始させます。無効化イベント後に通知を再開できるようにします。各変更ストリームのイベントドキュメントには、`_id` フィールドとして再開トークンが含まれます。変更後に再開する操作を表す変更イベントドキュメントの `_id` フィールド全体を渡します。 `StartAfter` は `ResumeAfter` および `StartAtOperationTime` と排他的です。
`StartAtOperationTime`	`Watch()` または `WatchAsync()` に、指定されたタイムスタンプ以降に発生したイベントのみを返すよう指示します。 `StartAtOperationTime` は `ResumeAfter` および `StartAfter` と相互排他的です。
`MaxAwaitTime`	空のバッチするを返す前に、新しいデータ変更が変更ストリームカーソルに報告されるまでサーバーが待機する最大時間をミリ秒単位で指定します。デフォルトは 1000 ミリ秒です。
`ShowExpandedEvents`	MongoDB Server v 6.0以降、変更ストリームは、 `createIndexes`イベントや`dropIndexes`イベントなどのデータ定義言語（DDL）イベントの変更通知をサポートします。展開されたイベントを変更ストリームに含めるには、変更ストリームカーソルを作成し、このパラメータを`True`に設定します。
`batchSize`	変更ストリームが各バッチで返すことができるドキュメントの最大数を指定します。これは `Watch()` または `WatchAsync()` に適用されます。`batchSize` オプションが設定されていない場合、ウォッチ関数の初期バッチサイズは `101` ドキュメントで、各後続バッチの最大サイズは 16 メビバイト (MiB) です。このオプションは、16 MiB より小さい制限を強制することはできますが、それを超える制限を設定することはできません。`batchSize` を 16 MiB より大きいバッチが生成される制限に設定した場合、このオプションは効果がありません。`Watch()` または `WatchAsync()` はデフォルトのバッチサイズを使用します。
`Collation`	Specifies the collation to use for the change stream cursor. See the Collation section of this page for more information.
`Comment`	操作にコメントを付けます。

照合

操作の照合を構成するには、照合クラスのインスタンスを作成します。

次の表では、Collation コンストラクターが受け入れるパラメーターを説明しています。また、各設定の値を読み取るために使用できる対応するクラスプロパティも一覧表示されます。

Parameter	説明	クラスプロパティ
`locale`	Specifies the International Components for Unicode (ICU) locale. For a list of supported locales, see Collation Locales and Default Parameters in the MongoDB Server Manual. If you want to use simple binary comparison, use the `Collation.Simple` static property to return a `Collation` object with the `locale` set to `"simple"`. Data Type: `string`	`Locale`
`caseLevel`	(任意) 大文字と小文字の比較を含めるかどうかを指定します。この引数が `true` の場合、ドライバーの動作は `strength` 引数の値によって異なります。 - `strength` が `CollationStrength.Primary` の場合、ドライバーは基本文字と大文字と小文字を比較します。 - `strength` が `CollationStrength.Secondary` の場合、ドライバーは基本文字、分音符号、その他の二次的な違い、および大文字と小文字を比較します。 - `strength` がその他の値の場合、この引数は無視されます。この引数が `false` の場合、ドライバーは強度レベル `Primary` または `Secondary` での大文字と小文字の比較を含みません。データ型: `boolean` デフォルト: `false`	`CaseLevel`
`caseFirst`	(Optional) Specifies the sort order of case differences during tertiary level comparisons. Data Type: CollationCaseFirst Default: `CollationCaseFirst.Off`	`CaseFirst`
`strength`	(Optional) Specifies the level of comparison to perform, as defined in the ICU documentation. Data Type: CollationStrength Default: `CollationStrength.Tertiary`	`Strength`
`numericOrdering`	（任意）ドライバーが数字の string を数値として比較するかどうかを指定します。この引数が `true`102の場合、ドライバーは数字の10 210 `false`string10 2を数値として比較します。例は、"" という文字列を比較する場合、および1 "``2 `` の場合、ドライバーは値をおよびとして処理し、が大きいことを検出します。この引数がまたは除外されている場合、ドライバーは数字の string を string として比較します。例は、"" という文字列を比較する場合、および "q" の場合、ドライバーは一度に 1 文字を 1 文字ずつ比較します。のため "" はが10 "q" より小さい場合、ドライバーは "q" を見つけます。は、"" より小さくなければなりません。詳細については、2 MongoDB Serverマニュアルの「照合制限」を参照してください。データ型:`boolean` デフォルト:`false`	`NumericOrdering`
`alternate`	(Optional) Specifies whether the driver considers whitespace and punctuation as base characters for purposes of comparison. Data Type: CollationAlternate Default: `CollationAlternate.NonIgnorable` (spaces and punctuation are considered base characters)	`Alternate`
`maxVariable`	(Optional) Specifies which characters the driver considers ignorable when the `alternate` argument is `CollationAlternate.Shifted`. Data Type: CollationMaxVariable Default: `CollationMaxVariable.Punctuation` (the driver ignores punctuation and spaces)	`MaxVariable`
`normalization`	(Optional) Specifies whether the driver normalizes text as needed. Most text doesn't require normalization. For more information about normalization, see the ICU documentation. Data Type: `boolean` Default: `false`	`Normalization`
`backwards`	(任意) 発音区別符号を含む string を、string の後ろから前にソートするかどうかを指定します。データ型: `boolean` デフォルト: `false`	`Backwards`

照合の詳細については、 MongoDB Serverマニュアルの照合ページを参照してください。

変更前と変更後のイメージを含めます

重要

配置で MongoDB v 6.0以降が使用されている場合にのみ、コレクションで変更前と変更後のイメージを有効にできます。

デフォルトでは、コレクションに対して操作を実行すると、対応する変更イベントにはその操作によって変更されたフィールドのデルタのみが含まれます。変更前または変更後の完全なドキュメントを表示するには、 ChangeStreamOptionsオブジェクトを作成し、 FullDocumentBeforeChangeまたはFullDocumentオプションを指定します。次に、 ChangeStreamOptionsオブジェクトをWatch()またはWatchAsync()メソッドに渡します。

変更前のイメージは、変更前のドキュメントの完全なバージョンです。変更ストリームイベントに変更前のイメージを含めるには、 FullDocumentBeforeChangeオプションを次のいずれかの値に設定します。

ChangeStreamFullDocumentBeforeChangeOption.WhenAvailable: 変更イベントには、変更前のイメージが利用可能な場合にのみ、変更イベント用の変更されたドキュメントの変更前のイメージが含まれます。
ChangeStreamFullDocumentBeforeChangeOption.Required: 変更イベントには、変更イベント用に変更されたドキュメントの変更前のイメージが含まれます。変更前のイメージが利用できない場合、ドライバーはエラーを発生させます。

変更後のイメージとは、変更後のドキュメントの完全なバージョンです。変更ストリームイベントに変更後のイメージを含めるには、 FullDocumentオプションを次のいずれかの値に設定します。

ChangeStreamFullDocumentOption.UpdateLookup: 変更イベントには、変更後一定時間の変更されたドキュメント全体のコピーが含まれます。
ChangeStreamFullDocumentOption.WhenAvailable: 変更イベントには、変更後のイメージが利用可能な場合にのみ、変更イベント用の変更されたドキュメントの変更後のイメージが含まれます。
ChangeStreamFullDocumentOption.Required: 変更イベントには、変更イベントの変更されたドキュメントの変更後のイメージが含まれます。変更後のイメージが利用できない場合、ドライバーはエラーを発生させます。

次の例では、コレクションの変更ストリームを開き、 FullDocumentオプションを指定して更新されたドキュメントの変更後のイメージを含めます。 SynchronousAsynchronous対応するコードを表示するには、タブまたはタブを選択します。

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .Match(change => change.OperationType == ChangeStreamOperationType.Update);
var options = new ChangeStreamOptions
{
    FullDocument = ChangeStreamFullDocumentOption.UpdateLookup,
};
using (var cursor = collection.Watch(pipeline, options))
{
    foreach (var change in cursor.ToEnumerable())
    {
        Console.WriteLine(change.FullDocument.ToBsonDocument());
    }
}

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .Match(change => change.OperationType == ChangeStreamOperationType.Update);
var options = new ChangeStreamOptions
{
    FullDocument = ChangeStreamFullDocumentOption.UpdateLookup,
};
using var cursor = await collection.WatchAsync(pipeline, options);
await cursor.ForEachAsync(change =>
{
    Console.WriteLine(change.FullDocument.ToBsonDocument());
});

上記のコード例を実行し、 "name"値が"Blarney Castle"であるドキュメントを更新すると、次の変更ストリーム出力が生成されます。

{ "_id" : ObjectId("..."), "name" : "Blarney Castle", "restaurant_id" : "40366356",
"cuisine" : "Traditional Irish", "address" : { "building" : "202-24", "coord" : [-73.925044200000002, 40.5595462],
"street" : "Rockaway Point Boulevard", "zipcode" : "11697" }, "borough" : "Queens", "grades" : [...] }

変更前と変更後のイメージの詳細については、Change Streams MongoDB Serverマニュアルの「とドキュメントの変更前イメージおよび変更後イメージ」を参照してください。

詳細情報

Change Streams変更ストリームの詳細については、MongoDB Server マニュアルの「ストリーム」を参照してください。

API ドキュメント

このガイドで説明したメソッドや型の詳細については、次の API ドキュメントを参照してください。

戻る

モニタリング

認証