Lire des données en temps réel avec des flux de modifications

Les flux de modifications pour Firestore avec la compatibilité MongoDB permettent aux applications d'accéder aux modifications en temps réel (insertions, mises à jour et suppressions) apportées à une collection ou à une base de données entière. Un flux de modifications classe les mises à jour par heure de modification.

Les flux de modifications sont accessibles via les API compatibles avec MongoDB et les pilotes MongoDB traditionnels. L'implémentation des flux de modifications de Firestore avec compatibilité MongoDB peut gérer n'importe quel débit d'écritures et de lectures grâce à une implémentation unique du partitionnement automatique sur les écritures et du parallélisme de lecture. Cela vous permet de créer des charges de travail à haut débit. Vous pouvez également améliorer l'infrastructure de migration et de synchronisation des données entre Cloud Firestore et d'autres solutions de stockage.

En plus de la compatibilité avec les pilotes MongoDB, vous pouvez utiliser Cloud Firestore pour lire les flux de modifications en parallèle. Cela vous permet de créer des charges de travail de lecture parallèles à haut débit. Chaque flux représente une partition de résultats bien répartie.

Les flux de modifications sont compatibles avec les fonctionnalités suivantes :

  • Flux de modifications configurables avec un champ d'application de base de données ou de collection.
  • Durée de conservation d'un flux de modifications spécifiée lors de la création. La durée de conservation par défaut est de sept jours et la durée de conservation minimale est d'un jour. La durée de conservation doit être un multiple de 1 jour, jusqu'à un maximum de 7 jours. La durée de conservation ne peut pas être modifiée après la création. Pour modifier la période de conservation, vous devez supprimer et recréer le flux de modifications.
  • Événements de modification delete, insert, update et drop observables à l'aide de db.collection.watch() et db.watch().
  • updateDescription.updatedFields contient les différences de mise à jour.
  • Toutes les options fullDocument et fullDocumentBeforeChange.
    • Recherche de mises à jour dans le document complet.
    • Image du document avant qu'il ne soit remplacé, mis à jour ou supprimé.
    • Image du document après son remplacement ou sa mise à jour.
    • Pour les images avant et après datant de plus d'une heure, vous devez activer la récupération à un moment précis (PITR).
  • Toutes les options de reprise, y compris resumeAfter et startAfter.
  • Lorsque vous utilisez watch() pour observer les modifications, vous pouvez enchaîner les étapes d'agrégation telles que $addFields, $match, $project, $replaceRoot, $replaceWith, $set et $unset.

Configurer les flux de modifications

Pour créer, supprimer ou afficher les flux de modifications existants pour une base de données, utilisez la console Google Cloud.

Rôles et autorisations

Pour créer, supprimer et lister les flux de modifications, un principal doit disposer respectivement des autorisations IAM (Identity and Access Management) datastore.schemas.create, datastore.schemas.delete et datastore.schemas.list.

Le rôle Administrateur de l'index Datastore (roles/datastore.indexAdmin), par exemple, accorde ces autorisations.

Créer un flux de modifications

Avant de pouvoir ouvrir un curseur de flux de modifications correspondant, vous devez créer un flux de modifications. L'activation automatique des flux de modifications lors de la création d'une collection ou d'une base de données n'est pas prise en charge.

Pour créer un flux de modifications, utilisez la console Google Cloud.

  1. Dans la console Google Cloud, accédez à la page Base de données.

    Accéder à la page "Bases de données"

  2. Dans la liste, sélectionnez une base de données Firestore compatible avec MongoDB. Le panneau Firestore Studio s'ouvre.
  3. Dans le panneau Explorateur, recherchez le nœud Flux de modifications, cliquez sur Autres actions, puis sélectionnez Créer un flux de modifications.
  4. Saisissez un nom, un champ d'application et une période de conservation uniques pour le flux de modifications, puis cliquez sur Enregistrer.

Afficher les flux de modifications

Vous pouvez afficher des informations sur les flux de modifications dans la console Google Cloud.

  1. Dans la console Google Cloud, accédez à la page Base de données.

    Accéder à la page "Bases de données"

  2. Dans la liste, sélectionnez une base de données Firestore compatible avec MongoDB. Le panneau Firestore Studio s'ouvre.
  3. Dans le panneau Explorateur, recherchez le nœud Flux de modifications.
  4. Pour ouvrir ou fermer le nœud, cliquez sur Activer/Désactiver le nœud.

Supprimer un flux de modifications

Pour supprimer un flux de modifications, utilisez la console Google Cloud.

  1. Dans la console Google Cloud, accédez à la page Base de données.

    Accéder à la page "Bases de données"

  2. Dans la liste, sélectionnez une base de données Firestore compatible avec MongoDB. Le panneau Firestore Studio s'ouvre.
  3. Dans le panneau Explorateur, recherchez le nœud Flux de modifications.
  4. Pour ouvrir ou fermer le nœud, cliquez sur Activer/Désactiver le nœud.
  5. Dans l'explorateur, recherchez le flux de modifications que vous souhaitez supprimer.
  6. Cliquez sur Autres actions, puis sélectionnez Supprimer le flux de modifications.
  7. Dans la boîte de dialogue, saisissez le nom du flux de modifications pour confirmer la suppression, puis cliquez sur Supprimer.

Ouvrir ou reprendre un curseur de flux de modifications

Les exemples suivants montrent comment créer, reprendre et configurer un curseur de flux de modifications.

Avant de créer un curseur de flux de modifications, vous devez créer explicitement un flux de modifications pour la base de données ou la collection.

Créer un curseur de flux de modifications

Pour créer un curseur de flux de modifications, utilisez la méthode watch dans les pilotes MongoDB. Pour écouter toutes les modifications apportées à une base de données, créez un flux de modifications à portée de base de données et appelez la méthode watch sur l'objet db.

let cursor = db.watch()

Pour créer un curseur limité à une collection, vous devez d'abord créer un flux de modifications pour cette collection. Appelez ensuite la méthode watch sur la collection correspondante.

let cursor = db.my_collection.watch()

Maintenant que vous avez créé un curseur de flux de modifications, vous pouvez commencer le streaming. Par exemple, si vous insérez un document et appelez tryNext sur le curseur, la modification s'affiche dans le flux de modifications.

let doc = db.my_collection.insertOne({value: "hello world"})
console.log(cursor.tryNext())

Si vous mettez à jour et supprimez le document, ces modifications s'affichent dans le flux de modifications :

db.my_collection.updateOne({"_id": doc.insertedId}, {$set: {value: "hello world!"}})
db.my_collection.deleteOne({"_id": doc.insertedId}})

// Prints the update event
console.log(cursor.tryNext())

// Prints the delete event
console.log(cursor.tryNext())

Reprendre un flux de modifications

Pour reprendre un flux de modifications, utilisez les options resumeAfter ou startAfter. Pour déterminer à quel endroit du journal des modifications reprendre à partir de resumeAfter et startAfter, utilisez un jeton de reprise.

// Create a cursor and add one event to the change stream.
let cursor = db.my_collection.watch();
db.my_collection.insertOne({value: "hello world"});
let event = cursor.tryNext();

// Get the resume token from the event.
let resumeToken = event._id;

// Add a new event to the change stream.
db.my_collection.insertOne({value: "foobar"});

// Create a new cursor by using the resume token as a starting point.
let newCursor = db.my_collection.watch({resumeAfter: resumeToken})

// Log the change event containing the "foobar" value.
console.log(newCursor.tryNext())

Pour utiliser startAfter :

// Start after the resume token.
let startAfterCursor = db.my_collection.watch({startAfter: resumeToken})

Inclure des images avant et après dans les mises à jour et les supprimer

Si nécessaire, vous pouvez inclure des images avant et après des documents dans les événements de modification de mise à jour et de suppression. La disponibilité des images est soumise à la période de récupération à un moment précis (PITR). Pour lire les images de documents datant de plus d'une heure, vous devez activer la récupération PITR.

Les flux de modifications tirent parti de la période PITR pour fournir une vue du document avant et après l'événement de modification donné. Par défaut, les événements de mise à jour contiennent un champ updateDescription qui correspond au delta des champs modifiés par l'opération de mise à jour.

Pour inclure les images avant et après dans un événement de modification, vous devez spécifier les options fullDocumentBeforeChange et fullDocument dans la requête du flux de modifications.

let cursor = db.my_collection.watch({
  "fullDocument": "required",
  "fullDocumentBeforeChange": "required"
})

Si la requête tente de lire un document en dehors de la période de conservation PITR ou si la récupération à un moment précis n'est pas activée, la valeur required génère un message d'erreur côté serveur.

Au lieu de générer une erreur, vous pouvez utiliser la valeur whenAvailable pour renvoyer une valeur null si les images ne sont plus disponibles.

let cursor = db.my_collection.watch({
  "fullDocument": "whenAvailable",
  "fullDocumentBeforeChange": "whenAvailable"
})

Inclure l'image actuelle dans les mises à jour

Par défaut, les événements de mise à jour contiennent un champ updateDescription qui correspond au delta des champs modifiés par l'opération de mise à jour. Pour rechercher la version la plus récente de l'intégralité du document, utilisez la valeur updateLookup dans l'option fullDocument.

Cette fonctionnalité ne nécessite pas de récupération à un instant donné et effectue une recherche du document.

let cursor = db.my_collection.watch({
  "fullDocument": "updateLookup",
})

Lectures parallèles

Pour augmenter le débit, vous pouvez utiliser l'option firestoreWorkerConfig afin de répartir une requête de flux de modifications sur plusieurs nœuds de calcul. Chaque nœud de calcul est responsable de la diffusion des modifications pour un ensemble distinct de documents. Vous devez créer un curseur parallèle à l'aide d'une requête runCommand ou aggregate.

Par exemple, vous pouvez distribuer un flux de modifications sur trois nœuds de calcul comme suit :

let cursor1 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 0 }}
  }]);

let cursor2 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 1 }}
  }]);

let cursor3 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 2 }}
  }]);

Flux de modifications et sauvegardes

Ni la configuration ni les données du flux de modifications ne sont disponibles dans les opérations de restauration de sauvegarde. Si vous restaurez une base de données avec des flux de modifications, vous devez recréer ces flux dans la base de données de destination pour ouvrir des curseurs vers cette base de données.

Facturation

Différences de comportement

La section suivante décrit les différences entre les flux de modifications de Firestore compatible avec MongoDB et ceux de MongoDB.

updateDescription

updateDescription est un document dans un événement update qui décrit les champs qui ont été mis à jour ou supprimés par l'opération de mise à jour. Dans Cloud Firestore, les différences notables sont les suivantes :

  • Dans updateDescription, les champs truncatedArrays et disambiguatedPaths ne sont pas renseignés.
  • updateDescription.updatedFields représente une différence canonique entre les images avant et après d'un document avant et après l'application d'une mutation.

Prenons l'exemple de l'état initial suivant d'un document :

db.my_collection.insertOne({
  _id: 1,
  root: {
    array: [{a: 1}, {b: 2}, {c: 3}]
  }
})

Scénario 1 : modifier uniquement le premier élément du tableau.

Dans ce scénario, le comportement de Cloud Firestore correspond à celui de MongoDB.

db.my_collection.updateOne(
  {_id: 1},
  {'$set': {"root.array.0.a": 100}}
)

{
  updatedFields: {"root.array.0.a": 100},
  removedFields: []
}

Scénario 2 : écraser avec un tableau entier

Dans ce scénario, l'opération ne met à jour que le premier champ de tableau, mais écrase l'ensemble du tableau.

La comparaison de mise à jour Cloud Firestore ne fait pas la différence entre ces deux scénarios et renvoie le même updateDescription.updatedFields pour les deux :

db.my_collection.updateOne(
  {_id: 1},
  {'$set': {"root.array": [{a: 100}, {b: 2}, {c: 3}]}}
)

// In other implementations, updatedFields reflects the mutation itself
{
  updatedFields: {
    "root.array": [{a: 100}, {b: 2}, {c: 3}]
  },
  removedFields: []
}

// Firestore updatedFields is the diff between the before and after versions of the document
{
  updatedFields: {"root.array.0.a": 100},
  removedFields: []
}

Étape suivante