Fluxos de alteração na API do Azure Cosmos DB para MongoDB

Artigo
06/15/2023

APLICA-SE AO: MongoDB

O suporte do feed de alterações na API do Azure Cosmos DB para MongoDB está disponível usando a API de fluxos de alteração. Usando a API de fluxos de alteração, seus aplicativos podem obter as alterações feitas na coleção ou nos itens em um único fragmento. Posteriormente, você pode executar ações adicionais com base nos resultados. As alterações nos itens na coleção são capturadas na ordem de seu tempo de modificação e a ordem de classificação é garantida por chave de fragmentação.

Observação

Para usar os fluxos de alteração, crie a API do Azure Cosmos DB para a conta do MongoDB com uma versão de servidor 3.6 ou superior. Se você executar os exemplos de fluxo de alterações em uma versão anterior, poderá ver o erro Nome da fase de pipeline não reconhecido: $changeStream.

Exemplos

O exemplo a seguir mostra como obter fluxos de alteração em todos os itens na coleção. Este exemplo cria um cursor para observar itens quando eles são inseridos, atualizados ou substituídos. O fase $match, a fase $project e a opção fullDocument são necessários para obter os fluxos de alteração. Não há suporte para a observação de operações de exclusão usando fluxos de alteração no momento. Como alternativa, você pode adicionar um marcador flexível nos itens que estão sendo excluídos. Por exemplo, você pode adicionar um atributo no item chamado "excluído". Quando quiser excluir o item, você pode definir "excluído" para true e definir um TTL no item. Como a atualização de "deleted" para true é uma atualização, essa alteração será visível no fluxo de alterações.

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

O exemplo a seguir mostra como usar a funcionalidade de fluxo de alterações em Java. Para ver o exemplo completo, confira este repositório GitHub. Este exemplo também mostra como usar o método resumeAfter para buscar todas as alterações da última leitura.

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

Alterações em um único fragmento

O exemplo a seguir mostra como obter alterações nos itens em um único fragmento. Este exemplo obtém as alterações de itens que têm a chave de fragmento igual a "a" e o valor da chave de fragmento igual a "1". É possível ter diferentes clientes lendo alterações de fragmentos diferentes em paralelo.

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

Dimensionamento de fluxos de alteração

Ao usar fluxos de alteração em escala, é melhor difundir a carga uniformemente. Utilize o comando personalizado GetChangeStreamTokens para difundir a carga entre fragmentos/partições físicas.

Limitações atuais

As seguintes limitações são aplicáveis ao usar fluxos de alteração:

As propriedades operationType e updateDescription ainda não têm suporte no documento de saída.
Atualmente, há suporte para os tipos de operações insert, update e replace. No entanto, ainda não há suporte para a operação de exclusão ou outros eventos.

Devido a essas limitações, as opções de fase $match, fase $project e fullDocument são necessárias, conforme mostrado nos exemplos anteriores.

Ao contrário do feed de alterações na API do Azure Cosmos DB para NoSQL, não há uma Biblioteca de Processador do Feed de Alterações separada para consumir fluxos de alteração ou a necessidade de um contêiner de concessões. Atualmente, não há suporte para gatilhos do Azure Functions para processamento de fluxos de alteração.

Tratamento de erros

Há suporte para os seguintes códigos de erro e mensagens ao usar os fluxo de alterações:

Código de erro HTTP 16500 – quando o fluxo de alterações é limitado, ele retorna uma página vazia.
NamespaceNotFound (OperationType Invalidate) – se você executar o fluxo de alterações na coleção que não existe ou se a coleção for solta, um erro NamespaceNotFound será retornado. Como a propriedade operationType não pode ser retornada no documento de saída, em vez do erro operationType Invalidate, o erro NamespaceNotFound será retornado.

Compartilhar via