Flux de modification dans l’API pour MongoDB d’Azure Cosmos DB

S’APPLIQUE À : MongoDB

La prise en charge du flux de modification dans l’API pour MongoDB d’Azure Cosmos DB est disponible à l’aide de l’API de flux de modification. À l’aide de l’API de flux de modification, vos applications peuvent obtenir les modifications apportées à la collection ou aux éléments d’une seul partition. Par la suite, vous pouvez effectuer d’autres actions en fonction des résultats. Les modifications apportées aux éléments de la collection sont capturées dans l’ordre de leur modification, et l’ordre de tri est garanti par clé de partition.

Notes

Pour utiliser les flux de modifications, créez l’API d’Azure Cosmos DB pour le compte MongoDB avec la version 3.6 ou ultérieure du serveur. Si vous exécutez les exemples de flux de modifications sur une version antérieure, vous pouvez voir l’erreur Nom de phase de pipeline non reconnu : $changeStream.

Exemples

L’exemple suivant montre comment récupérer des flux de modification sur tous les éléments de la collection. Cet exemple crée un curseur pour surveiller les éléments lorsqu’ils sont insérés, mis à jour ou remplacés. L’étape $match, l’étape $project et l’option fullDocument sont nécessaires pour récupérer les flux de modification. La surveillance des opérations de suppression à l’aide de flux de modification n’est pas prise en charge actuellement. Pour contourner ce problème, vous pouvez ajouter un marqueur réversible sur les éléments en cours de suppression. Par exemple, vous pouvez ajouter un attribut à l’élément appelé « supprimé ». Lorsque vous voulez supprimer l’élément, vous pouvez définir « supprimé » sur true et définir une TTL sur l’élément. Étant donné que la mise à jour de « supprimé » sur true reste une mise à jour, cette modification sera visible dans le flux de modification.

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

L’exemple suivant montre comment utiliser la fonctionnalité de flux de modifications dans Java. Pour obtenir un exemple complet, consultez ce Dépôt GitHub. Cet exemple montre également comment utiliser la méthode resumeAfter pour rechercher toutes les modifications de la dernière lecture.

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

Modifications au sein d’une partition unique

L’exemple suivant montre comment récupérer les modifications apportées aux éléments au sein d’une partition unique. Cet exemple récupère les modifications des éléments dont la clé de partition est égale à « a » et la valeur de clé de partition égale à « 1 ». Il est possible que différents clients lisent en parallèle les modifications de différentes partitions.

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

Mise à l’échelle des flux de modification

Lorsque vous utilisez des flux de modification à grande échelle, il est préférable de répartir uniformément la charge. Utilisez la commande personnalisée GetChangeStreamTokens pour répartir la charge sur les partitions/étendues physiques.

Limites actuelles

Les limitations suivantes s’appliquent quand plusieurs flux de modification sont utilisés :

• Les propriétés operationType et updateDescription ne sont pas encore prises en charge dans le document de sortie.
• Les types d'opérations insert, update et replace sont actuellement pris en charge. Toutefois, l’opération de suppression ou d’autres événements ne sont pas encore pris en charge.

En raison de ces limitations, l’étape $match, l’étape $project et les options fullDocument sont requises, comme indiqué dans les exemples précédents.

Contrairement au flux de modification dans l’API pour NoSQL d’Azure Cosmos DB, il n’existe pas de bibliothèque de processeurs de flux de modification distincte pour consommer des flux de modification ou un conteneur de baux. Il n’existe actuellement aucune prise en charge pour les déclencheurs Azure Functions permettant de traiter les flux de modification.

Gestion des erreurs

Les codes d’erreur et messages suivants sont pris en charge lorsque des flux de modification sont utilisés :

• Code d’erreur HTTP 16500 : lorsque le flux de modification est limité, il retourne une page vide.

• NamespaceNotFound (OperationType Invalidate) : si vous exécutez le flux de modification sur une collection qui n’existe pas ou si la collection est supprimée, une erreur NamespaceNotFound est retournée. Étant donné que la propriété operationType ne peut pas être retournée dans le document de sortie, au lieu de l’erreur operationType Invalidate, l’erreur NamespaceNotFound est retournée.

Étapes suivantes

• Utiliser la durée de vie pour faire expirer les données automatiquement dans l’API pour MongoDB d’Azure Cosmos DB
• Indexation dans l’API pour MongoDB d’Azure Cosmos DB