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

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. Korzystając z interfejsu API zmian strumieni, aplikacje mogą pobierać zmiany wprowadzone w kolekcji lub do elementów w jednym fragmentze. 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 na klucz fragmentu.

Uwaga

Aby użyć strumieni zmian, utwórz interfejs 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 we wszystkich elementach w kolekcji. W tym przykładzie tworzony jest kursor do obserwowania elementów podczas wstawiania, aktualizowania lub zastępowania. Aby $match pobrać strumienie zmian, wymagany jest etap, $project etap i fullDocument opcja. 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żesz 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 "usunięta" do true jest aktualizacją, ta zmiana będzie widoczna w strumieniu zmian.

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

W poniższym przykładzie pokazano, jak używać funkcji zmiany strumienia w języku Java, aby zapoznać się z kompletnym przykładem, zobacz to repozytorium GitHub. W tym przykładzie pokazano również, jak za pomocą resumeAfter metody wyszukać 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 w elementach w ramach pojedynczego fragmentu. Ten przykład pobiera zmiany elementów, które mają klucz fragmentu równy "a" i wartość klucza fragmentu równe "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 używania 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.
• Typy insertoperacji , updatei są replace obecnie obsługiwane. Jednak operacja usuwania lub inne zdarzenia nie są jeszcze obsługiwane.

Ze względu na te ograniczenia, etap $match, etap $project i opcje fullDocument są wymagane, 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 ani potrzeby kontenera dzierżaw. Obecnie wyzwalacze usługi Azure Functions nie obsługują 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ń zmiany 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 Invalidate zwracany NamespaceNotFound jest błąd.

Następne kroki

• Użyj czasu wygaśnięcia, aby automatycznie wygasać dane w interfejsie API usługi Azure Cosmos DB dla bazy danych MongoDB
• Indeksowanie w interfejsie API usługi Azure Cosmos DB dla bazy danych MongoDB