Änderungsdatenströme in der API für MongoDB von Azure Cosmos-DB

Artikel
06/15/2023

GILT FÜR: MongoDB

Die Unterstützung von Änderungsfeeds in der API für MongoDB von Azure Cosmos DB ist über die API für Änderungsdatenströme verfügbar. Mithilfe der API für Änderungsdatenströme können Ihre Anwendungen die Änderungen an der Sammlung oder an den Elementen in einem einzelnen Shard abrufen. Später können Sie auf der Grundlage der Ergebnisse weitere Maßnahmen ergreifen. Änderungen an den Elementen in der Sammlung werden in der Reihenfolge ihres Änderungszeitpunkts erfasst und die Sortierreihenfolge ist für die einzelnen Shardschlüssel sichergestellt.

Hinweis

Um Änderungsdatenströme zu verwenden, erstellen Sie die Azure Cosmos DB-API für das MongoDB-Konto mit Serverversion 3.6 oder höher. Wenn Sie die Änderungsdatenstrom-Beispiele für eine frühere Version ausführen, wird möglicherweise die Fehlermeldung Nicht erkannter Name einer Pipelinestage: $changeStream angezeigt.

Beispiele

Das folgende Beispiel zeigt, wie Sie Änderungsdatenströme für alle Elemente der Sammlung abrufen können. In diesem Beispiel wird ein Cursor erstellt, um Elemente zu überwachen, wenn sie eingefügt, aktualisiert oder ersetzt werden. Die $match-Phase, $project-Phase und fullDocument-Option sind erforderlich, um die Änderungsdatenströme abzurufen. Die Überwachung auf Löschvorgänge mit Änderungsdatenströmen wird derzeit nicht unterstützt. Als Problemumgehung können Sie den zu löschenden Elementen eine schwache Markierung hinzufügen. Sie können beispielsweise ein Attribut mit der Bezeichnung Gelöscht in das Element einfügen. Wenn Sie das Element löschen möchten, können Sie Deleted auf true setzen und eine TTL für das Element festlegen. Da das Aktualisieren von „deleted“ in true ein Update ist, wird diese Änderung im Änderungsdatenstrom angezeigt.

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

Das folgende Beispiel zeigt, wie Sie die Funktionalität zur Änderung des Datenstroms in Java verwenden können. Das komplette Beispiel finden Sie in diesem GitHub-Repository. Dieses Beispiel zeigt auch, wie Sie die resumeAfter-Methode verwenden, um alle Änderungen vom letzten Lesevorgang zu suchen.

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

Änderungen innerhalb eines einzelnen Shards

Das folgende Beispiel zeigt, wie Sie Änderungen an den Elementen in einem einzelnen Shard abrufen können. In diesem Beispiel werden die Änderungen von Elementen abgerufen, deren ShardsSchlüssel gleich „a“ und der Shardschlüsselwert gleich „1“ ist. Es ist möglich, dass verschiedene Clients Änderungen von verschiedenen Shards parallel lesen.

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

Skalieren von Änderungsdatenströmen

Wenn Sie Änderungsdatenströme im großen Stil verwenden, ist es am besten, die Last gleichmäßig zu verteilen. Verwenden Sie den benutzerdefinierten Befehl „GetChangeStreamTokens“, um die Last auf physische Shards/Partitionen zu verteilen.

Aktuelle Einschränkungen

Es gelten die folgenden Einschränkungen, wenn Änderungsdatenströme verwendet werden:

Die Eigenschaften operationType und updateDescription werden im Ausgabedokument noch nicht unterstützt.
Die Vorgangstypen insert, update und replace werden derzeit unterstützt. Löschvorgänge oder andere Ereignisse werden jedoch noch nicht unterstützt.

Aufgrund dieser Einschränkungen sind die Phasen „$match“, „$project“ und die „fullDocument“-Optionen erforderlich, wie in den vorherigen Beispielen gezeigt.

Anders als beim Änderungsfeed in der API für NoSQL von Azure Cosmos DB gibt es keine separate Änderungsfeed-Prozessorbibliothek zur Nutzung von Änderungsdatenströmen, und auch keinen Bedarf für einen Leasecontainer. Es gibt derzeit keine Unterstützung für Azure Functions-Trigger zum Verarbeiten von Änderungsdatenströmen.

Fehlerbehandlung

Die folgenden Fehlercodes und Meldungen werden bei der Verwendung von Änderungsdatenströmen unterstützt:

HTTP-Fehlercode 16500: Wenn der Änderungsdatenstrom gedrosselt wird, wird eine leere Seite zurückgegeben.
NamespaceNotFound (OperationType Invalidate) : Wenn Sie den Änderungsdatenstrom für die nicht existierende Sammlung ausführen oder die Sammlung verworfen wird, wird ein NamespaceNotFound-Fehler zurückgegeben. Da die operationType-Eigenschaft nicht im Ausgabedokument zurückgegeben werden kann, wird anstelle des Fehlers operationType Invalidate der Fehler NamespaceNotFound zurückgegeben.