Stromen wijzigen in Azure Cosmos DB API voor MongoDB

VAN TOEPASSING OP: Azure Cosmos DB-API voor MongoDB

Ondersteuning voor wijzigingsfeeds in Azure Cosmos DB API voor MongoDB is beschikbaar met behulp van de API voor wijzigingsstromen. Met behulp van de API voor wijzigingsstreams kunnen uw toepassingen de wijzigingen in de verzameling of de items in één shard ophalen. Later kunt u verdere acties uitvoeren op basis van de resultaten. Wijzigingen in de items in de verzameling worden vastgelegd in de volgorde van de wijzigingstijd en de sorteer volgorde wordt gegarandeerd per shardsleutel.

Voorbeelden

In het volgende voorbeeld ziet u hoe u wijzigingsstromen kunt ophalen voor alle items in de verzameling. In dit voorbeeld wordt een cursor gemaakt om items te bekijken wanneer ze worden ingevoegd, bijgewerkt of vervangen. De $match fase, $project fase en optie zijn vereist om de fullDocument wijzigingsstromen op te halen. Het zoeken naar verwijderbewerkingen met behulp van wijzigingsstreams wordt momenteel niet ondersteund. Als tijdelijke oplossing kunt u een zachte markering toevoegen aan de items die worden verwijderd. U kunt bijvoorbeeld een kenmerk toevoegen aan het item met de naam 'verwijderd'. Wanneer u het item wilt verwijderen, kunt u 'verwijderd' instellen op en een true TTL instellen voor het item. Omdat het bijwerken van 'verwijderd' naar een update is, is deze true wijziging zichtbaar in de wijzigingsstroom.

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();

Wijzigingen in één shard

In het volgende voorbeeld ziet u hoe u wijzigingen kunt aanbrengen in de items in één shard. In dit voorbeeld worden de wijzigingen van items met een shardsleutel die gelijk is aan 'a' en de shardsleutelwaarde gelijk aan '1'. Het is mogelijk om verschillende clients wijzigingen van verschillende shards parallel te laten lezen.

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

Wijzigingsstromen schalen

Wanneer u wijzigingsstromen op schaal gebruikt, kunt u de belasting het beste gelijkmatig spreiden. Gebruik de aangepaste getChangeStreamTokens-opdracht om de belasting over fysieke shards/partities te verdelen.

Huidige beperkingen

De volgende beperkingen zijn van toepassing bij het gebruik van wijzigingsstromen:

• De operationType eigenschappen en worden nog niet ondersteund in het updateDescription uitvoerdocument.
• De insert update bewerkingentypen , en replace worden momenteel ondersteund. De verwijderbewerking of andere gebeurtenissen worden echter nog niet ondersteund.

Vanwege deze beperkingen zijn de $match, $project fase en volledigedocumentopties vereist, zoals wordt weergegeven in de vorige voorbeelden.

In tegenstelling tot de wijzigingsfeed in de SQL-API van Azure Cosmos DB is er geen afzonderlijke processorbibliotheek voor de wijzigingsfeed om wijzigingsstromen te gebruiken of is er geen leasecontainer nodig. Er is momenteel geen ondersteuning voor Azure Functions voor het verwerken van wijzigingsstromen.

Foutafhandeling

De volgende foutcodes en berichten worden ondersteund bij het gebruik van wijzigingsstromen:

• HTTP-foutcode 16500: wanneer de wijzigingsstroom wordt beperkt, wordt een lege pagina weergegeven.

• NamespaceNotFound (OperationType Invalidate) : als u een wijzigingsstroom op de verzameling die niet bestaat of als de verzameling is uitgevallen, wordt een NamespaceNotFound fout geretourneerd. Omdat de operationType eigenschap niet kan worden geretourneerd in het uitvoerdocument, wordt de fout geretourneerd in plaats van de operationType Invalidate NamespaceNotFound fout.

Volgende stappen

• Time to live gebruiken om gegevens automatisch te laten verlopen in Azure Cosmos DB API voor MongoDB
• Indexeren in Azure Cosmos DB API voor MongoDB