Ändra strömmar i Azure Cosmos DB:s API för MongoDB

GÄLLER FÖR: MongoDB

Stöd för ändringsflöde i Azure Cosmos DB:s API för MongoDB är tillgängligt med hjälp av API:et för ändringsströmmar. Genom att använda API:et för ändringsströmmar kan dina program hämta ändringarna i samlingen eller objekten i en enda shard. Senare kan du vidta ytterligare åtgärder baserat på resultaten. Ändringar av objekten i samlingen samlas in i ordning efter ändringstiden och sorteringsordningen garanteras per shardnyckel.

Exempel

I följande exempel visas hur du hämtar ändringsströmmar på alla objekt i samlingen. Det här exemplet skapar en markör för att titta på objekt när de infogas, uppdateras eller ersätts. Fasen $match , $project fasen och fullDocument alternativet krävs för att hämta ändringsströmmarna. Det går för närvarande inte att titta efter borttagningsåtgärder med hjälp av ändringsströmmar. Som en 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 "borttaget" till true och ange en TTL för objektet. Eftersom uppdateringen "borttagen" till true ä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 collection = new MongoClient("<connection-string>")
    .GetDatabase("<database-name>")
    .GetCollection<BsonDocument>("<collection-name>");

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 
            }
        }"
    );

ChangeStreamOptions options = new ()
{
    FullDocument = ChangeStreamFullDocumentOption.UpdateLookup
};

using IChangeStreamCursor<BsonDocument> enumerator = collection.Watch(
    pipeline, 
    options
);

Console.WriteLine("Waiting for changes...");
while (enumerator.MoveNext())
{
    IEnumerable<BsonDocument> changes = enumerator.Current;
    foreach(BsonDocument change in changes)
    {
        Console.WriteLine(change);
    }  
}

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 i en enda shard

I följande exempel visas hur du hämtar ändringar i objekten i en enda shard. I det här exemplet hämtas ändringar av objekt som har en shardnyckel som är lika med "a" och shardnyckelvärdet lika med "1". Det är möjligt att olika klienter 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 ändringsströmmar

När du använder ändringsströmmar i stor skala är det bäst att fördela 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 updateDescription stöds ännu inte i utdatadokumentet.
• Operationstyperna insert, updateoch replace stöds för närvarande. Borttagningsåtgärden eller andra händelser stöds dock inte ännu.

På grund av dessa begränsningar krävs $match-fasen, $project-fasen och fullDocument-alternativen enligt föregående exempel.

Till skillnad från ändringsflödet i Azure Cosmos DB:s API för NoSQL finns det inte ett separat ändringsflödesprocessorbibliotek för att använda ändringsströmmar eller ett behov av en lånecontainer. Det finns för närvarande inte stöd för 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 ändringsströmmen på samlingen som inte finns eller om samlingen tas bort returneras ett NamespaceNotFound fel. Eftersom egenskapen operationType inte kan returneras i utdatadokumentet returneras felet i stället för operationType Invalidate felet NamespaceNotFound .

Nästa steg

• Använd time to live för att automatiskt förfalla data i Azure Cosmos DB:s API för MongoDB
• Indexering i Azure Cosmos DB:s API för MongoDB