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

Les flux de modifications pour Firestore compatible avec 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 ordonne 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 compatible avec 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 le parallélisme de lecture. Vous pouvez ainsi 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. Vous pouvez ainsi créer des charges de travail de lecture parallèles à haut débit. Chaque flux représente une partition de résultats bien distribuée.

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

  • Flux de modifications configurables avec une portée 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 un jour, jusqu'à un maximum de sept 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 du document complet pour les mises à jour.
    • Pré-image du document avant son remplacement, sa mise à jour ou sa suppression.
    • Post-image du document après son remplacement ou sa mise à jour.
    • Les pré-images et les post-images datant de plus d'une heure nécessitent l'activation de 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 des é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 nécessite respectivement les 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 du flux de modifications lors de la création d'une collection ou d'une base de données n'est pas disponible.

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

  1. Dans la console Google Cloud, accédez à la page Bases 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 de flux de modifications, une portée et une période de conservation uniques, 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 Bases 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 Basculer 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 Bases 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 Basculer le nœud.
  5. Dans l'Explorateur, recherchez le flux de modifications que vous souhaitez supprimer.
  6. Cliquez sur Autres actions et 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 explicitement créer 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 à portée de collection, vous devez d'abord créer un flux de modifications pour cette collection. Ensuite, appelez 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 à partir de quel point du journal des modifications reprendre 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 les pré-images et les post-images dans les mises à jour et les suppressions

Si nécessaire, vous pouvez inclure des pré-images et des post-images de 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 des 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 à la différence entre les champs modifiés par l'opération de mise à jour.

Pour inclure les pré-images et les post-images dans un événement de modification, vous devez spécifier les options fullDocumentBeforeChange et fullDocument dans la requête de 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 PITR 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 à la différence entre les champs modifiés par l'opération de mise à jour. Pour rechercher la version la plus récente du document complet, utilisez la valeur updateLookup dans l'option fullDocument.

Cette fonctionnalité ne nécessite pas la récupération PITR 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 diviser une requête de flux de modifications entre 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 via 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 du flux de modifications 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 de modifications 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 pré-images et les post-images d'un document avant et après l'application d'une mutation.

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

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

Scénario 1 : ne muter que 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 le tableau entier.

La différence de mise à jour Cloud Firestore ne fait pas la distinction 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