Change Stream no Azure DocumentDB

Os fluxos de alterações são um fluxo em tempo real de alterações na base de dados que flui da sua base de dados para a sua aplicação. Esta funcionalidade permite-lhe construir aplicações reativas subscrevendo alterações na base de dados, eliminando a necessidade de sondagens contínuas para detetar alterações.

Observação

O Azure DocumentDB suporta atualmente o Change Stream em clusters single-shard. Estamos a trabalhar ativamente para fornecer suporte para Change Stream em clusters de múltiplos fragmentos num futuro próximo.

O Azure DocumentDB impõe um limite de registo gerido pelo sistema de 400 MB para Fluxos de Alteração. Os clientes devem processar continuamente os Fluxos de Mudança para evitar perder eventos históricos de alteração. Os Fluxos de Alteração utilizam registos geridos pelo sistema com um limite de tamanho de 400 MB. Se um token de retomar se referir a um evento de alteração que saiu do registo, o Fluxo de Alterações não pode ser retomado a partir desse ponto e deve ser reiniciado. Planeamos uma antevisão para suporte alargado num futuro próximo.

Ativar fluxos de alteração

Pode ativar ou desativar esta funcionalidade usando o Azure CLI ou um template ARM. O suporte ao portal será adicionado em breve.

Passos para ativar fluxos de alteração no cluster via CLI

1. Iniciar sessão na CLI do Azure

az login

2. Recupera as definições atuais dos sinalizadores de funcionalidades no teu cluster. Isto garante que mantém quaisquer flags existentes ao adicionar a nova funcionalidade.

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. Envie um pedido PATCH para ativar a funcionalidade.

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. Verificar resultado:
   • Certifique-se de que a carga útil de resposta inclui "previewFeatures": ["ChangeStreams"].
   • Se encontrar o erro "alterar fluxos não é suportado neste cluster", por favor crie um pedido de suporte.

Configuração dos fluxos de alteração

Este código de exemplo inicia um fluxo de alterações na exampleCollection coleção, monitorizando continuamente quaisquer alterações. Quando é detetada uma alteração, recupera o evento de alteração e imprime-o em formato 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();
        }
    }
}

Importante

Os fluxos de alteração são retomáveis ao especificar um token de retomada ao abrir o cursor com resumeAfter. No entanto, espera-se que haja histórico suficiente para localizar a operação associada ao token. O documento observado no fluxo de alterações no _id campo representa o token retomável.

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

Monitorizar alterações na base de dados com o Change Stream

Vamos compreender a saída do fluxo de alterações através do exemplo.

Inserir

Neste evento de fluxo de mudança, vemos que um novo registo foi inserido inserted na coleção exampleCollection dentro de cs base de dados, e os detalhes do evento incluem o conteúdo completo do documento recém-adicionado.

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

Neste evento de atualização, os position e rating para John são modificados. O fluxo de alterações reflete uma alteração na coleção update com o estado do documento pós-atualização exampleCollection.

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

Eliminar

O evento de fluxo de mudança indica que um documento foi deleted da exampleCollection coleção na cs base de dados. O evento capta o identificador único do documento removido.

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

Personalizar dados no Change Stream

Personalize a saída do seu fluxo de alterações especificando um array de uma ou mais etapas do pipeline durante a configuração. Os operadores suportados incluem os seguintes.

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

Limitações

• Os eventos de fluxo de mudança em clusters multi-shard ainda não são suportados.
• Os cursores de fluxo de mudança precisam ser reinicializados após um evento de comutação por falha no estado atual.
• Update ainda não suporta a descrição da atualização.
• pre-image é uma opção não suportada.
• $changestream como um pipeline aninhado dentro de outro estágio ainda não é suportado.

Conteúdo relacionado

• Integrar os fluxos de Mudança com o Kafka (usando conector debezium).