Ändra dataströmmar i Azure Cosmos DB API för MongoDB

GÄLLER för: Azure Cosmos DB-API för MongoDB

Stöd för ändringsflöde i Azure Cosmos DB API för MongoDB är tillgängligt med hjälp av API:et för ändringsströmmar. Med hjälp av API:et för ändringsströmmar kan dina program hämta de ändringar som gjorts i samlingen eller till objekten i en enda shard. Senare kan du vidta ytterligare åtgärder baserat på resultatet. Ändringar av objekten i samlingen fångas i den ordning de ändras och sorteringsordningen garanteras per shardnyckel.

Exempel

I följande exempel visas hur du hämtar ändringsströmmar för alla objekt i samlingen. I det här exemplet skapas en markör för att titta på objekt när de infogas, uppdateras eller ersätts. Fas, $match fas och alternativ krävs för att hämta $project fullDocument ändringsströmmarna. Det finns för närvarande inte stöd för borttagningsåtgärder som använder ändringsströmmar. Som en tillfällig lösning kan du lägga till en mjuk markör för de objekt som tas bort. Du kan till exempel lägga till ett attribut i objektet med namnet "deleted". När du vill ta bort objektet kan du ange "borttagna" till och true ange ett TTL-värde för objektet. Eftersom uppdatering av "borttagna" true till är en uppdatering visas den här ändringen i ändringsströmmen.

var cursor = db.coll.watch(
    [
        { $match: { "operationType": { $in: ["insert", "update", "replace"] } } },
        { $project: { "_id": 1, "fullDocument": 1, "ns": 1, "documentKey": 1 } }
    ],
    { fullDocument: "updateLookup" });

while (!cursor.isExhausted()) {
    if (cursor.hasNext()) {
        printjson(cursor.next());
    }
}

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<BsonDocument>>()
    .Match(change => change.OperationType == ChangeStreamOperationType.Insert || change.OperationType == ChangeStreamOperationType.Update || change.OperationType == ChangeStreamOperationType.Replace)
    .AppendStage<ChangeStreamDocument<BsonDocument>, ChangeStreamDocument<BsonDocument>, BsonDocument>(
    "{ $project: { '_id': 1, 'fullDocument': 1, 'ns': 1, 'documentKey': 1 }}");

var options = new ChangeStreamOptions{
        FullDocument = ChangeStreamFullDocumentOption.UpdateLookup
    };

var enumerator = coll.Watch(pipeline, options).ToEnumerable().GetEnumerator();

while (enumerator.MoveNext()){
        Console.WriteLine(enumerator.Current);
    }

enumerator.Dispose();

Bson match = Aggregates.match(Filters.in("operationType", asList("update", "replace", "insert")));

// Pick the field you are most interested in
Bson project = Aggregates.project(fields(include("_id", "ns", "documentKey", "fullDocument")));

// This variable is for second example
BsonDocument resumeToken = null;

// Now time to build the pipeline
List<Bson> pipeline = Arrays.asList(match, project);

//#1 Simple example to seek changes

// Create cursor with update_lookup
MongoChangeStreamCursor<ChangeStreamDocument<org.bson.Document>> cursor = collection.watch(pipeline)
        .fullDocument(FullDocument.UPDATE_LOOKUP).cursor();

Document document = new Document("name", "doc-in-step-1-" + Math.random());
collection.insertOne(document);

while (cursor.hasNext()) {
    // There you go, we got the change document.
    ChangeStreamDocument<Document> csDoc = cursor.next();

    // Let is pick the token which will help us resuming
    // You can save this token in any persistent storage and retrieve it later
    resumeToken = csDoc.getResumeToken();
    //Printing the token
    System.out.println(resumeToken);
    
    //Printing the document.
    System.out.println(csDoc.getFullDocument());
    //This break is intentional but in real project feel free to remove it.
    break;
}

cursor.close();

Ändringar inom en enda shard

I följande exempel visas hur du hämtar ändringar av objekten i en enda shard. I det här exemplet hämtar ändringar av objekt som har en shardnyckel som är lika med "a" och värdet för shardnyckeln är lika med "1". Det går att ha olika klienter som läser ändringar från olika shards parallellt.

var cursor = db.coll.watch(
    [
        { 
            $match: { 
                $and: [
                    { "fullDocument.a": 1 }, 
                    { "operationType": { $in: ["insert", "update", "replace"] } }
                ]
            }
        },
        { $project: { "_id": 1, "fullDocument": 1, "ns": 1, "documentKey": 1} }
    ],
    { fullDocument: "updateLookup" });

Skala om ändringsströmmar

När du använder ändringsströmmar i stor skala är det bäst att sprida belastningen jämnt. Använd det anpassade kommandot GetChangeStreamTokens för att sprida belastningen över fysiska shards/partitioner.

Aktuella begränsningar

Följande begränsningar gäller när du använder ändringsströmmar:

• Egenskaperna operationType och stöds inte ännu i updateDescription utdatadokumentet.
• Drifttyperna insert , och stöds för update replace närvarande. Borttagningsåtgärden eller andra händelser stöds dock inte ännu.

På grund av dessa begränsningar krävs $match, $project och fullDocument som visas i föregående exempel.

Till skillnad från ändringsflödet i Azure Cosmos DB:s SQL API finns det inte ett separat bibliotek för ändringsflödesprocessorer för att använda ändringsströmmar eller ett behov av en lånecontainer. Det finns för närvarande inte stöd Azure Functions utlösare för att bearbeta ändringsströmmar.

Felhantering

Följande felkoder och meddelanden stöds när du använder ändringsströmmar:

• HTTP-felkod 16500 – När ändringsströmmen begränsas returneras en tom sida.

• NamespaceNotFound (OperationType Invalidate) – Om du kör en ändringsström i samlingen som inte finns eller om samlingen tas bort returneras NamespaceNotFound ett fel. Eftersom egenskapen operationType inte kan returneras i utdatadokumentet returneras felet i operationType Invalidate stället för NamespaceNotFound felet.

Nästa steg

• Använd Time to Live för att automatiskt förfalla data Azure Cosmos DB i API:et för MongoDB
• Indexering i Azure Cosmos DB API för MongoDB