Zmienianie strumieni w interfejsie API usługi Azure Cosmos DB dla bazy danych MongoDB

Artykuł
06/15/2023

DOTYCZY: Mongodb

Obsługa zestawienia zmian w interfejsie API usługi Azure Cosmos DB dla bazy danych MongoDB jest dostępna przy użyciu interfejsu API strumieni zmian. Za pomocą interfejsu API zmian strumieni aplikacje mogą pobierać zmiany wprowadzone w kolekcji lub do elementów w jednym fragmentie. Później możesz podjąć dalsze działania na podstawie wyników. Zmiany elementów w kolekcji są przechwytywane w kolejności ich czasu modyfikacji, a kolejność sortowania jest gwarantowana dla klucza fragmentu.

Uwaga

Aby użyć strumieni zmian, utwórz konto interfejsu API usługi Azure Cosmos DB dla bazy danych MongoDB z serwerem w wersji 3.6 lub nowszej. Jeśli uruchomisz przykłady strumienia zmian dla starszej wersji, może zostać wyświetlona nierozpoznana nazwa etapu potoku: $changeStream błąd.

Przykłady

W poniższym przykładzie pokazano, jak pobrać strumienie zmian dla wszystkich elementów w kolekcji. W tym przykładzie tworzony jest kursor do obserwowania elementów podczas wstawiania, aktualizowania lub zastępowania. Etap $match , $project etap i fullDocument opcja są wymagane do pobrania strumieni zmian. Obserwowanie operacji usuwania przy użyciu strumieni zmian nie jest obecnie obsługiwane. Aby obejść ten problem, można dodać znacznik nietrwały do elementów, które są usuwane. Można na przykład dodać atrybut w elemencie o nazwie "deleted". Jeśli chcesz usunąć element, możesz ustawić wartość "usunięto" na i ustawić true czas wygaśnięcia dla elementu. Ponieważ aktualizacja elementu "deleted" w celu true jest aktualizacją, ta zmiana będzie widoczna w strumieniu zmian.

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

W poniższym przykładzie pokazano, jak używać funkcji zmiany strumienia w języku Java, aby zapoznać się z pełnym przykładem, zobacz to repozytorium GitHub. W tym przykładzie pokazano również, jak za pomocą resumeAfter metody wyszukiwać wszystkie zmiany z ostatniego odczytu.

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

Zmiany w ramach pojedynczego fragmentu

W poniższym przykładzie pokazano, jak uzyskać zmiany elementów w ramach pojedynczego fragmentu. Ten przykład pobiera zmiany elementów, które mają klucz fragmentu równy "a", a wartość klucza fragmentu równa "1". Istnieje możliwość równoległego odczytywania zmian przez różnych klientów z różnych fragmentów.

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

Skalowanie strumieni zmian

W przypadku korzystania ze strumieni zmian na dużą skalę najlepiej równomiernie rozłożyć obciążenie. Użyj polecenia niestandardowego GetChangeStreamTokens , aby rozłożyć obciążenie między fizyczne fragmenty/partycje.

Bieżące ograniczenia

Podczas korzystania ze strumieni zmian obowiązują następujące ograniczenia:

Właściwości operationType i updateDescription nie są jeszcze obsługiwane w dokumencie wyjściowym.
insertTypy operacji , updatei replace są obecnie obsługiwane. Jednak operacja usuwania lub inne zdarzenia nie są jeszcze obsługiwane.

Ze względu na te ograniczenia wymagany jest etap $match, etap $project i opcje fullDocument, jak pokazano w poprzednich przykładach.

W przeciwieństwie do zestawienia zmian w interfejsie API usługi Azure Cosmos DB dla NoSQL nie ma oddzielnej biblioteki procesora zestawienia zmian do korzystania ze strumieni zmian lub potrzeby kontenera dzierżaw. Obecnie nie ma obsługi wyzwalaczy Azure Functions przetwarzania strumieni zmian.

Obsługa błędów

Podczas korzystania ze strumieni zmian obsługiwane są następujące kody błędów i komunikaty:

Kod błędu HTTP 16500 — gdy strumień zmian jest ograniczany, zwraca pustą stronę.
NamespaceNotFound (OperationType Invalidate) — jeśli uruchomisz strumień zmiany w kolekcji, która nie istnieje lub jeśli kolekcja zostanie porzucona, NamespaceNotFound zostanie zwrócony błąd. operationType Ponieważ nie można zwrócić właściwości w dokumencie wyjściowym, zamiast błędu operationType InvalidateNamespaceNotFound zwracany jest błąd.