MongoDB için Azure Cosmos DB API'sinde akışları değiştirme

Uygulama hedefı: MongoDB için Azure Cosmos DB API

Azure Cosmos DB'nin MongoDB API'sinde değişiklik akışı desteği, değişiklik akışları API'si kullanılarak kullanılabilir. Uygulamalarınız değişiklik akışları API'sini kullanarak koleksiyonda veya tek bir parçadaki öğelerde yapılan değişiklikleri edinebilirsiniz. Daha sonra, sonuçlara göre daha fazla eylem gerçekleştirebilirsiniz. Koleksiyondaki öğelerde yapılan değişiklikler, değişiklik zamanlarına göre yakalanır ve parça anahtarı başına sıralama düzeni garanti altına alındı.

Örnekler

Aşağıdaki örnekte, koleksiyondaki tüm öğelerde değişiklik akışlarının nasıl elde edildikleri gösterir. Bu örnek, öğeleri eklendik, güncelleştirildiğinde veya değiştirildiğinde izlemek için bir imleç oluşturur. Değişiklik $match $project akışlarını almak fullDocument için aşama, aşama ve seçenek gereklidir. Değişiklik akışlarını kullanarak silme işlemlerini izlemek şu anda desteklenmiyor. Geçici bir çözüm olarak, silinen öğelere geçici bir işaret ebilirsiniz. Örneğin, öğeye "deleted" adlı bir öznitelik ebilirsiniz. Öğeyi silmek istediğiniz zaman "deleted" (silindi) öğesini ve true öğede bir TTL değeri ayarlayın. "Silinen" olarak true güncelleştirilen bir güncelleştirme olduğu için, bu değişiklik değişiklik akışında görünür.

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

var options = new ChangeStreamOptions{
        FullDocument = ChangeStreamFullDocumentOption.UpdateLookup
    };

var enumerator = coll.Watch(pipeline, options).ToEnumerable().GetEnumerator();

while (enumerator.MoveNext()){
        Console.WriteLine(enumerator.Current);
    }

enumerator.Dispose();

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

Tek parça içindeki değişiklikler

Aşağıdaki örnekte, tek bir parça içindeki öğelerde yapılan değişikliklerin nasıl elde etmek olduğu gösterir. Bu örnek, "a" değerine eşit parça anahtarına ve parça anahtarı değeri "1" değerine eşit olan öğelerin değişikliklerini alır. Farklı parçalardan değişiklikleri okuyan farklı istemcilerin paralel olarak sahip olmak mümkündür.

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

Değişiklik akışlarını ölçeklendirme

Değişiklik akışlarını büyük ölçekte kullanırken yükün doğru şekilde yayılması en iyisidir. Yükü fiziksel parçalara/bölümlere yaymak için GetChangeStreamTokens özel komutunu kullanın.

Geçerli sınırlamalar

Değişiklik akışları kullanırken aşağıdaki sınırlamalar geçerlidir:

• Ve operationType updateDescription özellikleri henüz çıkış belgesinde desteklenmiyor.
• insert, update ve replace işlem türleri şu anda de desteklemektedir. Ancak silme işlemi veya diğer olaylar henüz desteklenmiyor.

Bu sınırlamalar nedeniyle, $match, $project ve fullDocument seçenekleri önceki örneklerde gösterildiği gibi gereklidir.

Azure Cosmos DB'nin SQL API'sinde değişiklik akışından farklı olarak, değişiklik akışlarını tüketen ayrı bir Değişiklik Akışı İşlemci Kitaplığı veya kiralama kapsayıcısı ihtiyacı olmaz. Şu anda değişiklik akışlarını Azure İşlevleri tetikleyicileri desteklememektedir.

Hata işleme

Değişiklik akışları kullanırken aşağıdaki hata kodları ve iletiler de destek alır:

• HTTP hata kodu 16500 - Değişiklik akışı kısıtlandı, boş bir sayfa döndürür.

• NamespaceNotFound (OperationType Invalidate) - Mevcut olmayan koleksiyonda değişiklik akışı çalıştıracaksanız veya koleksiyon bırakılırsa bir NamespaceNotFound hata döndürülür. özelliği operationType çıkış belgesinde döndürülene kadar hata operationType Invalidate yerine hata NamespaceNotFound döndürülür.

Sonraki adımlar

• MongoDB için Azure Cosmos DB API'sinde verilerin süresinin dolması için yaşam süresi kullanın
• MongoDB için Azure Cosmos DB API'sinde dizin oluşturma