Změna streamů 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 k dispozici pomocí rozhraní API pro streamy změn. Pomocí rozhraní API pro datové proudy 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í podle času jejich úpravy a pořadí řazení je zaručeno pro každý 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 pro starší verzi, může se zobrazit chyba Nerozpoznaný název fáze kanálu: $changeStream .

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 se vyžadují k získání datových proudů změn. Sledování operací odstranění pomocí datových proudů změn se v současné době nepodporuje. Jako alternativní řešení můžete k odstraněným položkám přidat symbolovou značku. Do položky můžete například přidat atribut s názvem "odstraněno". Pokud chcete položku odstranit, můžete nastavit hodnotu Odstraněno na true a hodnotu TTL pro položku. Vzhledem k tomu, že aktualizace "odstraněno" na true 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žívat funkci streamu změn v Javě. Úplný příklad najdete v tomto úložišti GitHub. Tento příklad také ukazuje, jak použít metodu resumeAfter k vyhledání všech změn od 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 roven "a" a hodnota klíče horizontálního oddílu rovna "1". Je možné, že různí klienti budou číst 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í datových proudů změn ve velkém měřítku je nejlepší rovnoměrně rozložit zatížení. Pomocí vlastního příkazu GetChangeStreamTokens rozložte zatížení napříč fyzickými horizontálními oddíly nebo oddíly.

Aktuální omezení

Při používání datových proudů změn platí následující omezení:

• Vlastnosti operationType a updateDescription zatím nejsou ve výstupním dokumentu podporované.
• V insertsoučasné době se podporují typy operací , updatea replace . Operace odstranění ani 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 využívala streamy změn nebo je potřeba kontejner zapůjčení. V současné době se nepodporuje Azure Functions triggery pro zpracování datových proudů změn.

Zpracování chyb

Při použití datových proudů změn jsou podporovány následující kódy chyb a zprávy:

• Kód chyby HTTP 16500 – Když dojde k omezení streamu změn, vrátí prázdnou stránku.

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

Další kroky

• Použití funkce Time to Live k automatickému vypršení platnosti dat v rozhraní API služby Azure Cosmos DB pro MongoDB
• Indexování v rozhraní API služby Azure Cosmos DB pro MongoDB