Alterar fluxos na API do Azure Cosmos DB para MongoDB

APLICA-SE A: MongoDB

O suporte de 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. Mais tarde, você pode tomar outras ações com base nos resultados. As alterações nos itens da coleção são capturadas na ordem de seu tempo de modificação e a ordem de classificação é garantida por chave de estilhaço.

Nota

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

Exemplos

O exemplo a seguir mostra como obter fluxos de alteração em todos os itens da coleção. Este exemplo cria um cursor para observar itens quando eles são inseridos, atualizados ou substituídos. O $match estágio, $project o estágio e fullDocument a opção são necessários para obter os fluxos de mudança. No momento, não há suporte para a observação de operações de exclusão usando fluxos de alteração. Como solução alternativa, você pode adicionar um marcador suave 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" como true e definir um TTL no item. Como a atualização "excluído" para true é uma atualização, essa alteração será visível no fluxo de alterações.

JavaScript

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

C#

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

Java

O exemplo a seguir mostra como usar a funcionalidade de fluxo de alterações em Java, para o exemplo completo, consulte este repositório GitHub. Este exemplo também mostra como usar o resumeAfter método 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 dentro de um único fragmento

O exemplo a seguir mostra como obter alterações nos itens dentro de 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 mudanças de diferentes fragmentos 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 mudança em escala, é melhor distribuir uniformemente a carga. Utilize o comando personalizado GetChangeStreamTokens para distribuir 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 operationType propriedades e updateDescription ainda não são suportadas no documento de saída.
• Os inserttipos , update, e replace operações são suportados atualmente. No entanto, a operação de exclusão ou outros eventos ainda não são suportados.

Devido a essas limitações, as opções $match estágio, $project estágio 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 do Processador de Feed de Alterações separada para consumir fluxos de alteração ou uma necessidade de um contêiner de concessão. Atualmente, não há suporte para gatilhos do Azure Functions para processar fluxos de alterações.

Processamento de erros

Os seguintes códigos de erro e mensagens são suportados ao usar fluxos de alteração:

• Código de erro HTTP 16500 - Quando o fluxo de alteração é 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 descartada, um NamespaceNotFound erro será retornado. Como a operationType propriedade não pode ser retornada no documento de saída, em vez do operationType Invalidate erro, o NamespaceNotFound erro é retornado.

Próximos passos

• Use o tempo de vida para expirar dados automaticamente na API do Azure Cosmos DB para MongoDB
• Indexação na API do Azure Cosmos DB para MongoDB