Ändra Stream på Azure DocumentDB

Ändringsströmmar är en realtidsström med databasändringar som flödar från databasen till ditt program. Med den här funktionen kan du skapa reaktiva program genom att prenumerera på databasändringar, vilket eliminerar behovet av kontinuerlig avsökning för att identifiera ändringar.

Anmärkning

Azure DocumentDB stöder för närvarande Change Stream i kluster med en enda shard. Vi arbetar aktivt med att ge Change Stream-stöd för kluster med flera shardar inom en snar framtid.

Azure DocumentDB tillämpar en systemhanterad logggräns på 400 MB för ändringsströmmar. Kunder bör bearbeta Ändringsströmmar kontinuerligt för att undvika att historiska ändringshändelser saknas. Förändringsströmmar använder systemhanterade loggar med en storleksgräns på 400 MB. Om en återstartstoken refererar till en ändringshändelse som har rullat bort från loggen kan ändringsströmmen inte återupptas från den punkten och behöver startas om. Vi planerar en förhandsversion för utökat stöd inom en snar framtid.

Aktivera ändringsströmmar

Du kan aktivera eller inaktivera den här funktionen med hjälp av Azure CLI eller en ARM-mall. Portalstöd kommer snart att läggas till.

Steg för att aktivera ändringsströmmar i klustret via CLI

1. Logga in på Azure CLI

az login

2. Hämta de aktuella inställningarna för funktionsflaggor i klustret. Detta säkerställer att du behåller alla befintliga flaggor när du lägger till den nya funktionen.

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. Skicka PATCH-begäran för att aktivera funktionen.

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. Kontrollera resultatet:
   • Se till att svarsnyttolasten innehåller "previewFeatures": ["ChangeStreams"].
   • Om du får felet "ändringsströmmar stöds inte i det här klustret" skapar du en supportbegäran.

Konfigurera ändringsströmmar

Den här exempelkoden initierar en ändringsström i exampleCollection samlingen och övervakar kontinuerligt eventuella ändringar. När en ändring identifieras hämtar den ändringshändelsen och skriver ut den i JSON-format.

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

Viktigt!

Ändringsströmmar kan återupptas genom att ange en återuppta-token till resumeAfter när markören öppnas. Det förväntas dock att det finns tillräckligt med historik för att hitta operationen som är associerad med token. Dokumentet som observerats i ändringsströmmen i _id fältet representerar den återanvändbara token.

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

Övervaka databasändringar med Change Stream

Nu ska vi förstå utdata från ändringsströmmen via exemplet.

Infoga

I denna förändringsströmhändelse ser vi att en ny post inserted tillades i exampleCollection-samlingen i cs-databasen, och händelseinformationen innehåller det fullständiga innehållet i det nyligen tillagda dokumentet.

{
  "_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

I den här uppdateringshändelsen ändras position och rating för John. Ändringsströmmen återspeglar en update i exampleCollection-samlingen med dokumentets status efter uppdateringen.

{
  "_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" }
  }
}

Ta bort

Ändringsströmshändelsen anger att ett dokument kom deleted från exampleCollection samlingen i cs databasen. Händelsen fångar den unika identifieraren för det borttagna dokumentet.

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

Anpassa data i Change Stream

Anpassa dina utdata för ändringsströmmen genom att ange en matris med en eller flera pipelinesteg under konfigurationen. Operatorer som stöds är följande.

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

Begränsningar

• Ändringsströmhändelser i kluster med flera fragment stöds ännu inte.
• Markörer för ändringsström måste återställas efter en felövergångshändelse i nuvarande tillstånd.
• Update händelsen har ännu inte stöd för uppdateringsbeskrivning.
• pre-image är ett alternativ som inte stöds.
• $changestream som en kapslad pipeline i en annan fas stöds ännu inte.

Relaterat innehåll

• Integrera ändringsströmmar med Kafka (med debezium-anslutare).