MongoDB 用仮想コアベースの Azure Cosmos DB での Stream の変更

変更ストリームは、データベースからアプリケーションへ送られるデータベース変更のリアルタイム ストリームです。 この機能を使用すると、データベースの変更をサブスクライブしてリアクティブ アプリケーションを構築できるため、変更を検出するための継続的ポーリングが不要になります。

注

Azure Cosmos DB for MongoDB (仮想コア) では現在、単一シャード クラスターでの Change Stream がサポートされています。 近い将来、マルチシャード クラスターに対する Change Stream サポートの提供に積極的に取り組んでいます。

Azure Cosmos DB for MongoDB (仮想コア) では、変更ストリームに 400 MB の制限が課されます。 この制限を増やすサポート リクエストを送信してください。

変更ストリームを有効にする

この機能は、Azure CLI または ARM テンプレートを使用して有効または無効にすることができます。 ポータルのサポートは近日中に追加される予定です。

CLI を使用して仮想コア クラスターで変更ストリームを有効にする手順

1. Azure CLI にログイン

az login

2. クラスターの機能フラグの現在の設定を取得します。 これにより、新しい機能を追加するときに既存のフラグを保持できます。

az resource show --ids "/subscriptions/<sub id>/resourceGroups/<resource group name>/providers/Microsoft.DocumentDB/mongoClusters/<resource name of your Cosmos DB for MongoDB vCore cluster>" --api-version 2024-10-01-preview

3. PATCH 要求を送信して機能を有効にします。

az resource patch --ids "/subscriptions/<subscription_id>/resourceGroups/<resource_group_name>/providers/Microsoft.DocumentDB/mongoClusters/<vCore_cluster_name>" --api-version 2024-10-01-preview --properties "{\"previewFeatures\": [ \"ChangeStreams\"]}"

4. 結果を確認します。
   • 応答ペイロードに "previewFeatures": ["ChangeStreams"]が含まれていることを確認します。
   • "このクラスターでストリームの変更がサポートされていません" というエラーが発生した場合は、サポート要求を作成してください。

変更ストリームの構成

このコード例では、exampleCollection コレクションで変更ストリームを開始し、変更を継続的に監視します。 変更が検出されると、変更イベントが取得され、JSON 形式で出力されます。

JavaScript

// Open a change stream
const changeStream = db.exampleCollection.watch();

// Listen for changes
while (changeStream.hasNext()) 
    {
        const change = changeStream.next();
        printjson(change);
    }

ジャワ

package com.example;
 
import com.mongodb.MongoClientSettings;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
 
import org.bson.BsonDocument;
import org.bson.Document;
import org.bson.json.JsonWriterSettings;
import org.json.JSONObject;
 
public class ChangeStreamExample {
    public static void main(String[] args) {
        MongoClient mongoClient = MongoClients.create("mongodb://<username>:<pwd>@<clustername>.global.mongocluster.cosmos.azure.com/?tls=true");
        MongoDatabase database = mongoClient.getDatabase("test");
        MongoCollection<Document> collection = database.getCollection("test");
 
        // Open a change stream
        MongoCursor<ChangeStreamDocument<Document>> cursor = collection.watch().iterator();
 
        // Listen for changes
        try {
            while (cursor.hasNext()) {
                ChangeStreamDocument<Document> change = cursor.next();
 
                // Create a JSONObject from the fields of the ChangeStreamDocument
                JSONObject json = new JSONObject();
 
                // Add fields to the JSON object
                json.put("operationType", change.getOperationType().getValue());
                json.put("namespace", change.getNamespace().getFullName());
 
                if (change.getDocumentKey() != null) {
                    json.put("documentKey", change.getDocumentKey().toJson().toString());
                }
 
                if (change.getResumeToken() != null) {
                    json.put("resumeToken", change.getResumeToken().toJson());
                }
 
                if (change.getFullDocument() != null) {
                    json.put("fullDocument", change.getFullDocument().toJson());
                }
 
                if (change.getWallTime() != null) {
                    json.put("wallTime", change.getWallTime().toString());
                }
 
                // Pretty-print the JSON object with 4-space indent
                System.out.println(json.toString(4)); 
            }
        } 
        catch (Exception e) {
            System.err.println("Error processing change stream: " + e);
        } 
        finally {
            cursor.close();
            mongoClient.close();
        }
    }
}

重要

変更ストリームは、カーソルを開く際に resumeAfter する再開トークンを指定することで再開できます。 ただし、トークンに関連付けられている操作を見つけるのに十分な履歴が必要です。 _idフィールドの変更ストリームで観察されたドキュメントは、再開可能なトークンを表します。

cursor = db.exampleCollection.watch(resume_after=resume_token)

Change Stream を使用したデータベース変更の監視

この例を使用して、変更ストリームの出力を理解しましょう。

[挿入]

この変更ストリーム イベントで、新しいレコードが inserted データベース内の exampleCollection コレクションに cs されたことがわかります。イベントの詳細には、新しく追加されたドキュメントのすべての内容が含まれます。

{
  "_id": { "_data": "AeARBpQ/AAAA" }, // "resume_token"
  "operationType": "insert",
  "fullDocument": {
    "_id": { "$oid": "66e6f63e6f49ecaabf794958" },
    "employee_id": "17986",
    "name": "John",
    "position": "Software Engineer",
    "department": "IT",
    "rating": 4
  },
  "ns": { "db": "cs", "coll": "exampleCollection" },
  "documentKey": { "_id": { "$oid": "66e6f63e6f49ecaabf794958" } }
}

更新

この更新イベントでは、John の position & rating が変更されます。 変更ストリームには、ドキュメントの更新後の状態を持つupdate コレクション内の exampleCollection が反映されます。

{
  "_id": { "_data": "AWACAKM/AAAA" },
  "operationType": "update",
  "fullDocument": {
    "_id": { "$oid": "66e6f63e6f49ecaabf794958" },
    "employee_id": "17986",
    "name": "John",
    "position": "SSE",
    "department": "IT",
    "rating": 5
  },
  "ns": {
    "db": "cs", "coll": "exampleCollection"
  },
  "documentKey": {
    "_id": { "$oid": "66e6f63e6f49ecaabf794958" }
  }
}

削除

変更ストリーム イベントは、ドキュメントがdeleted データベースのexampleCollection コレクションからの cs であることを示します。 このイベントは、削除されたドキュメントの一意識別子を取得します。

{
  "_id": { "_data": "ASgBAJs/AAAA" },
  "operationType": "delete",
  "ns": { "db": "cs", "coll": "exampleCollection" },
  "documentKey": { "_id": { "$oid": "66e6f63e6f49ecaabf794958" } }
}

Change Stream でデータをパーソナライズする

構成中に 1 つ以上のパイプライン ステージの配列を指定して、変更ストリーム出力をカスタマイズします。 サポートされるオペレーターには以下が含まれます。

• $addFields
• $match
• $project
• $set
• $unset

制限事項

• マルチシャード クラスターでの変更ストリーム イベントは、まだサポートされていません。
• 変更ストリーム カーソルは、現在の状態でのフェールオーバー イベントの後に再初期化する必要があります。
• Update イベントは、まだ更新の説明をサポートしていません。
• pre-image はサポートされていないオプションです。
• 別のステージの入れ子になったパイプラインとしての $changestream は、まだサポートされていません。

関連コンテンツ

• 変更ストリームを Kafka と統合する (debezium コネクタを使用)。