Mengubah Stream di Azure DocumentDB

Aliran perubahan adalah aliran perubahan database real time yang mengalir dari database Anda ke aplikasi Anda. Fitur ini memungkinkan Anda membangun aplikasi reaktif dengan berlangganan perubahan database, menghilangkan kebutuhan polling berkelanjutan untuk mendeteksi perubahan.

Nota

Azure DocumentDB saat ini mendukung Change Stream dalam kluster shard tunggal. Kami secara aktif berupaya memberikan dukungan Change Stream untuk kluster multi-shard dalam waktu dekat.

Azure DocumentDB memberlakukan batas log yang dikelola sistem 400 MB untuk Change Streams. Pelanggan harus memproses "Change Streams" terus-menerus untuk menghindari kejadian perubahan historis yang hilang. Change Streams menggunakan log yang dikelola sistem dengan batas ukuran maksimum 400 MB. Jika token resume mengacu pada peristiwa perubahan yang telah diluncurkan dari log, Aliran Perubahan tidak dapat dilanjutkan dari titik tersebut dan harus dimulai ulang. Kami merencanakan pratinjau untuk dukungan diperluas dalam waktu dekat.

Mengaktifkan aliran perubahan

Anda dapat mengaktifkan atau menonaktifkan fitur ini menggunakan Azure CLI atau templat ARM. Dukungan portal akan segera ditambahkan.

Langkah-langkah untuk mengaktifkan aliran perubahan pada kluster melalui CLI

1. Masuk ke Azure CLI

az login

2. Ambil pengaturan saat ini untuk bendera fitur pada kluster Anda. Ini memastikan Anda mempertahankan bendera yang ada saat menambahkan fitur baru.

az resource show --ids "/subscriptions/<sub id>/resourceGroups/<resource group name>/providers/Microsoft.DocumentDB/mongoClusters/<resource name of your Azure DocumentDB cluster>" --api-version 2024-10-01-preview

3. Kirim permintaan PATCH untuk mengaktifkan fitur.

az resource patch --ids "/subscriptions/<subscription_id>/resourceGroups/<resource_group_name>/providers/Microsoft.DocumentDB/mongoClusters/<cluster-name>" --api-version 2024-10-01-preview --properties "{\"previewFeatures\": [ \"ChangeStreams\"]}"

4. Verifikasi hasil:
   • Pastikan payload respons mencakup "previewFeatures": ["ChangeStreams"].
   • Jika Anda mengalami kesalahan "change streams tidak didukung pada kluster ini", silakan buat permintaan dukungan.

Mengonfigurasi aliran perubahan

Contoh kode ini memulai aliran perubahan pada exampleCollection koleksi, terus memantau perubahan apa pun. Ketika suatu perubahan terdeteksi, sistem mengambil data peristiwa perubahan tersebut dan mencetaknya dalam format JSON.

JavaScript

// Open a change stream
const changeStream = db.exampleCollection.watch();

// Listen for changes
while (changeStream.hasNext()) 
    {
        const change = changeStream.next();
        printjson(change);
    }

Java

package com.example;
 
import com.mongodb.MongoClientSettings;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
 
import org.bson.BsonDocument;
import org.bson.Document;
import org.bson.json.JsonWriterSettings;
import org.json.JSONObject;
 
public class ChangeStreamExample {
    public static void main(String[] args) {
        MongoClient mongoClient = MongoClients.create("mongodb://<username>:<pwd>@<clustername>.global.mongocluster.cosmos.azure.com/?tls=true");
        MongoDatabase database = mongoClient.getDatabase("test");
        MongoCollection<Document> collection = database.getCollection("test");
 
        // Open a change stream
        MongoCursor<ChangeStreamDocument<Document>> cursor = collection.watch().iterator();
 
        // Listen for changes
        try {
            while (cursor.hasNext()) {
                ChangeStreamDocument<Document> change = cursor.next();
 
                // Create a JSONObject from the fields of the ChangeStreamDocument
                JSONObject json = new JSONObject();
 
                // Add fields to the JSON object
                json.put("operationType", change.getOperationType().getValue());
                json.put("namespace", change.getNamespace().getFullName());
 
                if (change.getDocumentKey() != null) {
                    json.put("documentKey", change.getDocumentKey().toJson().toString());
                }
 
                if (change.getResumeToken() != null) {
                    json.put("resumeToken", change.getResumeToken().toJson());
                }
 
                if (change.getFullDocument() != null) {
                    json.put("fullDocument", change.getFullDocument().toJson());
                }
 
                if (change.getWallTime() != null) {
                    json.put("wallTime", change.getWallTime().toString());
                }
 
                // Pretty-print the JSON object with 4-space indent
                System.out.println(json.toString(4)); 
            }
        } 
        catch (Exception e) {
            System.err.println("Error processing change stream: " + e);
        } 
        finally {
            cursor.close();
            mongoClient.close();
        }
    }
}

Important

Aliran perubahan dapat dilanjutkan dengan menentukan token resume ke resumeAfter saat membuka kursor. Diharapkan ada cukup data riwayat untuk menemukan operasi yang terkait dengan token. Dokumen yang diamati dalam aliran perubahan di _id bidang mewakili token yang dapat dilanjutkan.

cursor = db.exampleCollection.watch(resume_after=resume_token)

Memantau perubahan pada database dengan Change Stream

Mari kita pahami output aliran perubahan melalui contoh.

Sisipkan

Dalam peristiwa perubahan aliran ini, kita melihat bahwa rekaman baru ditambahkan inserted ke dalam koleksi exampleCollection di dalam basis data cs, dan detail peristiwa menyertakan konten lengkap dokumen yang baru ditambahkan.

{
  "_id": { "_data": "AeARBpQ/AAAA" }, // "resume_token"
  "operationType": "insert",
  "fullDocument": {
    "_id": { "$oid": "66e6f63e6f49ecaabf794958" },
    "employee_id": "17986",
    "name": "John",
    "position": "Software Engineer",
    "department": "IT",
    "rating": 4
  },
  "ns": { "db": "cs", "coll": "exampleCollection" },
  "documentKey": { "_id": { "$oid": "66e6f63e6f49ecaabf794958" } }
}

Update

Dalam peristiwa pembaruan ini, position & rating untuk John dimodifikasi. Aliran perubahan mencerminkan suatu update dalam koleksi exampleCollection dengan status setelah pembaruan dokumen.

{
  "_id": { "_data": "AWACAKM/AAAA" },
  "operationType": "update",
  "fullDocument": {
    "_id": { "$oid": "66e6f63e6f49ecaabf794958" },
    "employee_id": "17986",
    "name": "John",
    "position": "SSE",
    "department": "IT",
    "rating": 5
  },
  "ns": {
    "db": "cs", "coll": "exampleCollection"
  },
  "documentKey": {
    "_id": { "$oid": "66e6f63e6f49ecaabf794958" }
  }
}

Menghapus

Peristiwa aliran perubahan menunjukkan bahwa dokumen berasal deleted dari exampleCollection koleksi dalam cs database. Kejadian ini menangkap pengidentifikasi unik dokumen yang dihapus.

{
  "_id": { "_data": "ASgBAJs/AAAA" },
  "operationType": "delete",
  "ns": { "db": "cs", "coll": "exampleCollection" },
  "documentKey": { "_id": { "$oid": "66e6f63e6f49ecaabf794958" } }
}

Personalisasikan data dalam Aliran Perubahan

Sesuaikan output aliran perubahan Anda dengan menentukan array dari satu atau beberapa tahap alur selama konfigurasi. Operator yang didukung mencakup yang berikut ini.

• $addFields
• $match
• $project
• $set
• $unset

Keterbatasan

• Peristiwa perubahan aliran pada klaster multi-shard belum didukung.
• Kursor aliran perubahan perlu diinisialisasi ulang setelah peristiwa failover dalam keadaan saat ini.
• Update peristiwa belum mendukung deskripsi pembaruan.
• pre-image adalah opsi yang tidak didukung.
• $changestream sebagai alur berlapis dari tahap lain belum didukung.

Konten terkait

• Integrasikan Change Streams dengan Kafka (menggunakan konektor Debezium).