Lee datos en tiempo real con flujos de cambios

Los flujos de cambios para Firestore con compatibilidad con MongoDB permiten que las aplicaciones accedan a los cambios en tiempo real (inserciones, actualizaciones y eliminaciones) que se realizan en una colección o en una base de datos completa. Un flujo de cambios ordena las actualizaciones por hora de modificación.

Se puede acceder a los flujos de cambios a través de las APIs compatibles con MongoDB y los controladores de MongoDB tradicionales. La implementación de flujos de cambios de Firestore con compatibilidad con MongoDB puede controlar cualquier capacidad de procesamiento de escrituras y lecturas a través de una implementación única de partición automática en escrituras y paralelismo de lectura. Esto te permite compilar cargas de trabajo de alta capacidad de procesamiento. También puedes mejorar la infraestructura de migración y sincronización de datos entre Cloud Firestore y otras soluciones de almacenamiento.

Además de la compatibilidad con los controladores de MongoDB, puedes usar Cloud Firestore para leer flujos de cambios en paralelo. Esto te permite compilar cargas de trabajo de lectura paralelas y de alta capacidad de procesamiento. Cada transmisión representa una partición de resultados bien distribuida.

Los flujos de cambios admiten las siguientes funciones:

  • Flujos de cambios configurables con alcance de base de datos o colección
  • Es una duración de retención para un flujo de cambios que se especifica en el momento de la creación. La retención predeterminada es de 7 días y la mínima es de 1 día. La retención debe ser un múltiplo de 1 día, hasta un máximo de 7 días. La duración de la retención no se puede cambiar después de la creación. Para cambiar el período de retención, debes borrar y volver a crear el flujo de cambios.
  • Eventos de cambio de delete, insert, update y drop que se pueden observar con db.collection.watch() y db.watch()
  • updateDescription.updatedFields contiene diferencias de actualización.
  • Todas las opciones de fullDocument y fullDocumentBeforeChange.
    • Buscando actualizaciones en el documento completo.
    • Es la imagen previa del documento antes de que se reemplazara, actualizara o borrara.
    • Es la imagen posterior del documento después de que se reemplazó o actualizó.
    • Las imágenes anteriores y posteriores que tengan más de una hora requieren que se habilite la recuperación de un momento determinado (PITR).
  • Todas las opciones de reanudación, incluidas resumeAfter y startAfter.
  • Cuando usas watch() para observar cambios, puedes encadenar etapas de agregación, como $addFields, $match, $project, $replaceRoot, $replaceWith, $set y $unset.

Configura los flujos de cambios

Para crear, borrar o ver los flujos de cambios existentes de una base de datos, usa la consola de Google Cloud.

Funciones y permisos

Para crear, borrar y enumerar flujos de cambios, una principal requiere los permisos de Identity and Access Management (IAM) datastore.schemas.create, datastore.schemas.delete y datastore.schemas.list, respectivamente.

Por ejemplo, el rol Administrador de índices del almacén de datos (roles/datastore.indexAdmin) otorga estos permisos.

Crear flujo de cambios

Antes de abrir un cursor de flujo de cambios correspondiente, debes crear un flujo de cambios. No se admite la habilitación automática de flujos de cambios en el momento de crear la colección o la base de datos.

Para crear un flujo de cambios, usa la consola de Google Cloud.

  1. En la consola de Google Cloud, ve a la página Bases de datos.

    Ir a Bases de datos

  2. En la lista, selecciona una base de datos de Firestore compatible con MongoDB. Se abrirá el panel de Firestore Studio.
  3. En el panel Explorador, busca el nodo Flujos de cambios, haz clic en Más acciones y, luego, selecciona Crear flujo de cambios.
  4. Ingresa un nombre, un alcance y un período de retención únicos para el flujo de cambios y, luego, haz clic en Guardar.

Ver flujos de cambios

Puedes ver los detalles de los flujos de cambios en la consola de Google Cloud.

  1. En la consola de Google Cloud, ve a la página Bases de datos.

    Ir a Bases de datos

  2. En la lista, selecciona una base de datos de Firestore compatible con MongoDB. Se abrirá el panel de Firestore Studio.
  3. En el panel Explorador, busca el nodo Flujos de cambios.
  4. Para abrir o cerrar el nodo, haz clic en Activar o desactivar nodo.

Borra un flujo de cambios

Para borrar un flujo de cambios, usa la consola de Google Cloud.

  1. En la consola de Google Cloud, ve a la página Bases de datos.

    Ir a Bases de datos

  2. En la lista, selecciona una base de datos de Firestore compatible con MongoDB. Se abrirá el panel de Firestore Studio.
  3. En el panel Explorador, busca el nodo Flujos de cambios.
  4. Para abrir o cerrar el nodo, haz clic en Activar o desactivar nodo.
  5. En el Explorador, busca el flujo de cambios que deseas borrar.
  6. Haz clic en Más acciones y, luego, selecciona Borrar flujo de cambios.
  7. En el cuadro de diálogo, ingresa el nombre del flujo de cambios para confirmar la eliminación y, luego, haz clic en Borrar.

Abrir o reanudar un cursor de flujo de cambios

En los siguientes ejemplos, se muestra cómo crear, reanudar y configurar un cursor de flujo de cambios.

Antes de crear un cursor de flujo de cambios, debes crear un flujo de cambios de forma explícita para la base de datos o la colección.

Crea un cursor de flujo de cambios

Para crear un nuevo cursor de flujo de cambios, usa el método watch en los controladores de MongoDB. Para escuchar todos los cambios en una base de datos, crea un flujo de cambios con alcance de base de datos y llama al método watch en el objeto db.

let cursor = db.watch()

Para crear un cursor con alcance en una colección, primero debes crear un flujo de cambios para esa colección. Luego, llama al método watch en la colección correspondiente.

let cursor = db.my_collection.watch()

Ahora que creaste un cursor de flujo de cambios, puedes comenzar a transmitir. Por ejemplo, si insertas un documento y llamas a tryNext en el cursor, verás que el cambio aparece en el flujo de cambios.

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

Si actualizas y borras el documento, verás que esos cambios aparecen en el flujo de cambios:

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

Reanuda un flujo de cambios

Para reanudar un flujo de cambios, usa las opciones resumeAfter o startAfter. Para determinar desde dónde reanudar el registro de cambios de resumeAfter y startAfter, usa un token de reanudación.

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

Para usar startAfter, ingresa el siguiente comando:

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

Incluye imágenes previas y posteriores en las actualizaciones y eliminaciones

Si es necesario, puedes incluir imágenes anteriores y posteriores de los documentos en los eventos de cambio de actualización y eliminación. La disponibilidad de las imágenes está sujeta al período de recuperación de un momento determinado (PITR) y, para leer imágenes de documentos con más de una hora de antigüedad, debes habilitar la PITR.

Los flujos de cambios aprovechan la ventana de PITR para proporcionar una vista del documento antes y después del evento de cambio determinado. De forma predeterminada, los eventos de actualización contienen un campo updateDescription, que es el delta de los campos modificados por la operación de actualización.

Para incluir las imágenes anteriores y posteriores en un evento de cambio, debes especificar las opciones fullDocumentBeforeChange y fullDocument en la consulta del flujo de cambios.

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

Si la consulta intenta leer un documento fuera del período de retención de la PITR o si la PITR no está habilitada, el valor required arroja un mensaje de error del servidor.

Como alternativa a arrojar un error, puedes usar el valor whenAvailable para devolver un valor null si las imágenes ya no están disponibles.

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

Incluye la imagen actual en las actualizaciones

De forma predeterminada, los eventos de actualización contienen un campo updateDescription, que es el delta de los campos modificados por la operación de actualización. Para buscar la versión más reciente de todo el documento, usa el valor updateLookup en la opción fullDocument.

Esta función no requiere PITR y realiza una búsqueda del documento.

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

Lecturas paralelas

Para aumentar la capacidad de procesamiento, puedes usar la opción firestoreWorkerConfig para dividir una consulta de flujo de cambios entre varios trabajadores. Cada trabajador es responsable de entregar los cambios de un conjunto distinto de documentos. Debes crear un cursor paralelo a través de una consulta runCommand o aggregate.

Por ejemplo, puedes distribuir un flujo de cambios entre 3 trabajadores de la siguiente manera:

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

Flujos de cambios y copias de seguridad

Ni la configuración del flujo de cambios ni los datos del flujo de cambios están disponibles en las operaciones de restablecimiento de copias de seguridad. Si restableces una base de datos con flujos de cambios, debes volver a crear esos flujos de cambios en la base de datos de destino para abrir cursores en esa base de datos.

Facturación

Diferencias de comportamiento

En la siguiente sección, se describen las diferencias en los flujos de cambios entre Firestore con compatibilidad con MongoDB y MongoDB.

updateDescription

updateDescription es un documento en un evento update que describe los campos que se actualizaron o quitaron con la operación de actualización. En Cloud Firestore, las diferencias notables son las siguientes:

  • En updateDescription, los campos truncatedArrays y disambiguatedPaths no se propagan.
  • updateDescription.updatedFields representa una diferencia canónica entre las imágenes anteriores y posteriores de un documento antes y después de que se aplique una mutación.

Considera el siguiente estado inicial de un documento:

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

Situación 1: Mutar solo el primer elemento del array

En esta situación, el comportamiento de Cloud Firestore coincide con el de MongoDB.

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

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

Situación 2: Reemplazar con un array completo

En esta situación, la operación actualiza solo el primer campo del array, pero reemplaza todo el array.

La diferencia de actualización de Cloud Firestore no distingue entre estas dos situaciones y devuelve el mismo updateDescription.updatedFields para ambas:

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: []
}

¿Qué sigue?