Stream perubahan di API Azure Cosmos DB untuk MongoDB

Artikel
06/15/2023

BERLAKU UNTUK: MongoDB

Dukunganumpan perubahan di API Azure Cosmos DB untuk MongoDB tersedia dengan menggunakan API stream perubahan. Dengan menggunakan API stream perubahan, aplikasi Anda bisa mendapatkan perubahan yang dilakukan pada koleksi atau item dalam satu pecahan. Nantinya Anda dapat mengambil tindakan lebih lanjut berdasarkan hasilnya. Perubahan pada item dalam koleksi ditangkap dalam urutan waktu modifikasinya dan urutan sortir dijamin per kunci pecahan.

Catatan

Untuk menggunakan stream perubahan, buat API Azure Cosmos DB untuk akun MongoDB dengan versi server 3.6 atau yang lebih tinggi. Jika Anda menjalankan contoh stream perubahan terhadap versi yang lebih lama, Anda mungkin melihat kesalahan Nama tahap pipeline yang tidak dikenal: $changeStream.

Contoh

Contoh berikut menunjukkan cara mendapatkan stream perubahan pada semua item dalam koleksi. Contoh ini membuat kursor untuk mengamati item saat disisipkan, diperbarui, atau diganti. $matchTahap, $project tahap, fullDocument dan opsi diperlukan untuk mendapatkan stream perubahan. Mengamati operasi hapus menggunakan stream perubahan saat ini tidak didukung. Sebagai solusinya, Anda bisa menambahkan penanda lembut pada item yang sedang dihapus. Misalnya, Anda dapat menambahkan atribut dalam item bernama "dihapus." Bila Anda ingin menghapus item, Anda bisa menyetel "dihapus" ke true dan mengatur TTL pada item. Karena memperbarui "dihapus" true adalah pembaruan, perubahan ini akan terlihat di stream perubahan.

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

Contoh berikut menunjukkan cara menggunakan fungsi stream perubahan di Java, untuk contoh lengkapnya, lihat repo GitHub ini. Contoh ini juga menunjukkan cara menggunakan resumeAfter metode untuk mencari semua perubahan dari bacaan terakhir.

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

Perubahan dalam satu pecahan

Contoh berikut menunjukkan cara mendapatkan perubahan pada item dalam satu pecahan. Contoh ini mendapatkan perubahan item yang memiliki kunci pecahan sama dengan "a" dan nilai kunci pecahan sama dengan "1". Klien yang berbeda dapat membaca perubahan dari pecahan yang berbeda secara paralel.

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

Penskalaan aliran perubahan

Saat menggunakan aliran perubahan pada skala, sebaiknya sebarkan beban secara merata. Gunakan GetChangeStreamTokens custom command untuk menyebarkan beban ke seluruh partisi/pecahan fisik.

Batasan saat ini

Batasan berikut ini berlaku saat menggunakan stream perubahan:

Properti operationType dan updateDescription belum didukung dalam dokumen output.
Jenis operasi insert, update, dan replace saat ini didukung. Namun, operasi hapus atau peristiwa lain belum didukung.

Karena keterbatasan ini, opsi tahap $match, tahap $project, dan fullDocument diperlukan seperti yang ditunjukkan dalam contoh sebelumnya.

Tidak seperti umpan perubahan di API Azure Cosmos DB untuk NoSQL, tidak ada Pustaka Prosesor Umpan Perubahan terpisah untuk menggunakan aliran perubahan atau kebutuhan akan kontainer sewa. Saat ini tidak ada dukungan untuk pemicu Azure Functions untuk memproses stream perubahan.

Penanganan kesalahan

Kode kesalahan dan pesan berikut ini didukung saat menggunakan stream perubahan:

Kode kesalahan HTTP 16500 - Saat stream perubahan dibatasi, akan ditampilkan halaman kosong.
NamespaceNotFound (OperationType Invalidate) - Jika Anda menjalankan stream perubahan pada koleksi yang tidak ada atau jika koleksi dihilangkan, maka NamespaceNotFound kesalahan akan ditampilkan. Karena operationType properti tidak dapat ditampilkan dalam dokumen output, bukan operationType Invalidate kesalahan, NamespaceNotFound kesalahan akan ditampilkan.