Alterar fluxos na API do Azure Cosmos DB para MongoDB

APLICA-SE A: MongoDB

O suporte para feeds de alterações na API do Azure Cosmos DB para MongoDB está disponível através da API de fluxos de alteração. Ao utilizar a API de fluxos de alterações, as aplicações podem fazer as alterações à coleção ou aos itens numa única partição horizontal. Mais tarde, pode efetuar mais ações com base nos resultados. As alterações aos itens na coleção são capturadas pela ordem do tempo de modificação e a sequência de ordenação é garantida por chave de partição horizontal.

Nota

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

Exemplos

O exemplo seguinte mostra como obter fluxos de alterações em todos os itens da coleção. Este exemplo cria um cursor para ver itens quando são inseridos, atualizados ou substituídos. A $match fase, $project a fase e fullDocument a opção são necessárias para obter os fluxos de alteração. Atualmente, a monitorização de operações de eliminação com fluxos de alterações não é suportada. Como solução, pode adicionar um marcador suave aos itens que estão a ser eliminados. Por exemplo, pode adicionar um atributo no item denominado "eliminado". Quando quiser eliminar o item, pode definir "eliminado" como true e definir um TTL no item. Uma vez que a atualização "eliminada" para true é uma atualização, esta 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 seguinte mostra como utilizar a funcionalidade de fluxo de alterações em Java, para obter o exemplo completo, veja este repositório do GitHub. Este exemplo também mostra como utilizar o resumeAfter método para procurar 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 numa única partição horizontal

O exemplo seguinte mostra como obter alterações aos itens numa única partição horizontal. Este exemplo obtém as alterações de itens que têm a chave de partição horizontal igual a "a" e o valor da chave de partição horizontal igual a "1". É possível que diferentes clientes leiam alterações de diferentes partições horizontais 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" });

Dimensionar fluxos de alteração

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

Limitações atuais

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

• As operationType propriedades e updateDescription ainda não são suportadas no documento de saída.
• Os inserttipos de operações , updatee replace são atualmente suportados. No entanto, a operação de eliminação ou outros eventos ainda não são suportados.

Devido a estas limitações, as opções $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 existe uma Biblioteca de Processadores de Feeds de Alterações separada para consumir fluxos de alterações ou uma necessidade de um contentor de concessões. Atualmente, não existe suporte para Funções do Azure acionadores para processar fluxos de alteração.

Processamento de erros

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

• Código de erro HTTP 16500 – quando o fluxo de alterações é limitado, devolve uma página vazia.

• NamespaceNotFound (OperationType Invalidate) – se executar o fluxo de alterações na coleção que não existe ou se a coleção for removida, será devolvido um NamespaceNotFound erro. Uma vez que a operationType propriedade não pode ser devolvida no documento de saída, em vez do operationType Invalidate erro, é devolvido o NamespaceNotFound erro.

Passos seguintes

• Utilizar 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