Změna datových proudů v rozhraní API služby Azure Cosmos DB pro MongoDB

PLATÍ PRO: MongoDB

Podpora kanálu změn v rozhraní API služby Azure Cosmos DB pro MongoDB je dostupná pomocí rozhraní API streamů změn. Pomocí rozhraní API streamů změn můžou vaše aplikace získat změny provedené v kolekci nebo v položkách v jednom horizontálním oddílu. Později můžete na základě výsledků provést další akce. Změny položek v kolekci se zaznamenávají v pořadí jejich času úprav a pořadí řazení je zaručeno na klíč horizontálního oddílu.

Poznámka:

Pokud chcete použít streamy změn, vytvořte účet rozhraní API služby Azure Cosmos DB pro MongoDB se serverem verze 3.6 nebo vyšší. Pokud spustíte příklady streamu změn u starší verze, může se zobrazit nerozpoznaný název fáze kanálu: $changeStream chyba.

Příklady

Následující příklad ukazuje, jak získat streamy změn u všech položek v kolekci. Tento příklad vytvoří kurzor pro sledování položek při jejich vložení, aktualizaci nebo nahrazení. Fáze $match , $project fáze a fullDocument možnost jsou potřeba k získání streamů změn. Sledování operací odstranění pomocí streamů změn se v současné době nepodporuje. Jako alternativní řešení můžete přidat obnovitelné značky k odstraněným položkám. Do položky s názvem "odstraněno" můžete například přidat atribut. Pokud chcete položku odstranit, můžete u položky nastavit hodnotu true TTL a nastavit na ní hodnotu TTL. Vzhledem k tomu, že aktualizace odstraněné na true aktualizaci je aktualizace, bude tato změna viditelná ve streamu změn.

JavaScript

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

C#

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

Java

Následující příklad ukazuje, jak použít funkci změn streamu v Javě, pro úplný příklad si prohlédnou toto úložiště GitHub. Tento příklad také ukazuje, jak použít metodu resumeAfter k vyhledání všech změn z posledního čtení.

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

Změny v rámci jednoho horizontálního oddílu

Následující příklad ukazuje, jak získat změny položek v rámci jednoho horizontálního oddílu. Tento příklad získá změny položek, které mají klíč horizontálního oddílu rovno "a" a hodnota klíče horizontálního oddílu se rovná "1". Je možné mít různé klienty, kteří čtou změny z různých horizontálních oddílů paralelně.

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

Škálování datových proudů změn

Při použití streamů změn ve velkém měřítku je nejlepší rovnoměrně rozložit zatížení. Pomocí vlastního příkazu GetChangeStreamTokens rozdělte zatížení mezi fyzické horizontální oddíly nebo oddíly.

Aktuální omezení

Při použití streamů změn platí následující omezení:

• Výstupní dokument zatím nepodporuje vlastnosti operationType a updateDescription vlastnosti.
• Typy insertoperací a replace , updatejsou v současné době podporovány. Operace odstranění nebo jiné události se ale zatím nepodporují.

Vzhledem k těmto omezením se vyžadují možnosti $match fáze, fáze $project a fullDocument, jak je znázorněno v předchozích příkladech.

Na rozdíl od kanálu změn v rozhraní API služby Azure Cosmos DB pro NoSQL neexistuje samostatná knihovna procesoru kanálu změn, která by spotřebovávat streamy změn nebo potřebovala kontejner zapůjčení. Triggery Azure Functions v současné době nepodporují zpracování streamů změn.

Zpracování chyb

Při použití streamů změn se podporují následující kódy chyb a zprávy:

• Kód chyby HTTP 16500 – Pokud je omezený datový proud změn, vrátí prázdnou stránku.

• NamespaceNotFound (OperationType Invalidate) – Pokud spustíte stream změn v kolekci, která neexistuje nebo pokud je kolekce vyřazena, NamespaceNotFound vrátí se chyba. operationType Vzhledem k tomu, že vlastnost nelze vrátit ve výstupním dokumentu místo operationType Invalidate chyby, vrátí se NamespaceNotFound chyba.

Další kroky

• Automatické vypršení platnosti dat v rozhraní API služby Azure Cosmos DB pro MongoDB pomocí funkce Time to Live
• Indexování v rozhraní API služby Azure Cosmos DB pro MongoDB