Crea y administra flujos de cambios

En esta página, se describe cómo crear, modificar y ver transmisiones de cambios de Spanner para bases de datos con dialecto de GoogleSQL y bases de datos con dialecto de PostgreSQL. Para obtener más información sobre los flujos de cambios, consulta Acerca de los flujos de cambios.

Dado que los flujos de cambios son objetos de esquema, se crean y gestionan a través del mismo lenguaje DDL.actualizaciones de esquema Se utiliza para cualquier otro tipo de trabajo de definición de bases de datos, como la creación de tablas o la adición de índices.

Spanner comienza una operación de larga duración después de que envías una declaración en DDL que cambia el esquema, incluidas las que se usan para crear, alterar o borrar flujos de cambios. Una vez finalizada esta operación de larga duración, un flujo de cambios nuevo o modificado comienza a supervisar las columnas o tablas especificadas por su nueva configuración.

Crear transmisión de cambios

Para crear un flujo de cambios, debe proporcionar su nombre y los objetos del esquema que supervisa: ya sea la base de datos completa o una lista de tablas y columnas específicas. Opcionalmente, puede configurar un flujo de cambios con cualquiera de las siguientes opciones:

• Especifica el período de retención de datos para anular el período de retención predeterminado de siete días.
• Especifique el tipo de captura de valor para anular el tipo de captura de valor predeterminado OLD_AND_NEW_VALUES.
• Aplica un filtro de eliminaciones basadas en el TTL para filtrar las eliminaciones basadas en el TTL de tus flujos de cambios.
• Aplique un filtro de modificaciones de tabla para excluir todas las modificaciones de tabla INSERT, UPDATE o DELETE.

• Habilite una exclusión de registros a nivel de transacción para excluir ciertas transacciones de sus flujos de cambios.

CREATE CHANGE STREAM CHANGE_STREAM_NAME
  [FOR column_or_table_watching_definition[, ... ] ]
  [
    OPTIONS (
      retention_period = timespan,
      value_capture_type = type,
      exclude_ttl_deletes = boolean,
      exclude_insert = boolean,
      exclude_update = boolean,
      exclude_delete = boolean,
      allow_txn_exclusion = boolean
    )
  ]

CREATE CHANGE STREAM CHANGE_STREAM_NAME
  [FOR column_or_table_watching_definition[, ... ] ]
  [
    WITH (
      retention_period = timespan,
      value_capture_type = type,
      exclude_ttl_deletes = boolean,
      exclude_insert = boolean,
      exclude_update = boolean,
      exclude_delete = boolean,
      allow_txn_exclusion = boolean
    )
  ]

Una nueva secuencia de cambios comienza a supervisar los objetos de esquema que se le han asignado tan pronto como finaliza la operación de larga duración que la creó.

Los siguientes ejemplos ilustran la creación de flujos de cambios con diversas configuraciones.

Ver una base de datos completa

Para crear un flujo de cambios que supervise cada cambio de datos realizado en las tablas de una base de datos, utilice la palabra clave ALL:

CREATE CHANGE STREAM EverythingStream
FOR ALL;

La configuración ALL incluye implícitamente todas las tablas y columnas de datos futuras de la base de datos, tan pronto como se creen. No incluye vistas, tablas de esquemas de información ni otros objetos, además de las tablas de datos normales.

Consulte tablas específicas

Para limitar el alcance de un flujo de cambios a tablas específicas, en lugar de a toda la base de datos, especifique una lista de una o más tablas:

CREATE CHANGE STREAM SingerAlbumStream
FOR Singers, Albums;

Spanner actualiza automáticamente los flujos de cambios que supervisan tablas completas para reflejar cualquier cambio de esquema que afecte a esas tablas, como columnas añadidas o eliminadas.

Consulte columnas específicas

Utilice la sintaxis table(column_1[, column_2, ...]) para observar los cambios en una o más columnas específicas que no sean claves dentro de las tablas que usted nombre:

CREATE CHANGE STREAM NamesAndTitles
FOR Singers(FirstName, LastName), Albums(Title);

No puedes especificar columnas de clave primaria aquí porque cada flujo de cambios always rastrea las claves primarias de cada tabla que monitorea. Esto permite que cada registro de cambio de datos identifique la fila modificada mediante su clave primaria.

Observa tablas y columnas en una sola secuencia.

Puedes combinar la sintaxis de observación de tablas y la de observación de columnas de los dos ejemplos anteriores en un único flujo de cambios:

CREATE CHANGE STREAM NamesAndAlbums
FOR Singers(FirstName, LastName), Albums;

Especifica un período de retención más largo

Para especificar un período de retención de datos del flujo de cambios más largo que el predeterminado de siete días, establece retention_period en un período de hasta treinta días, expresado en horas (h) o días (d).

Dos ejemplos:

CREATE CHANGE STREAM LongerDataRetention
FOR ALL
OPTIONS ( retention_period = '36h' );

CREATE CHANGE STREAM MaximumDataRetention
FOR ALL
OPTIONS ( retention_period = '7d' );

CREATE CHANGE STREAM LongerDataRetention
FOR ALL
WITH ( retention_period = '36h' );

CREATE CHANGE STREAM MaximumDataRetention
FOR ALL
WITH ( retention_period = '7d' );

Especifique un tipo de captura de valor diferente.

Para especificar un tipo de captura de valor de transmisión de cambios que no sea OLD_AND_NEW_VALUES, establece value_capture_type en NEW_VALUES o NEW_ROW, como se muestra en los siguientes ejemplos:

CREATE CHANGE STREAM NewRowChangeStream
FOR ALL
OPTIONS ( value_capture_type = 'NEW_ROW' );

CREATE CHANGE STREAM NewValuesChangeStream
FOR ALL
OPTIONS ( value_capture_type = 'NEW_VALUES' );

CREATE CHANGE STREAM NewRowChangeStream
FOR ALL
WITH ( value_capture_type = 'NEW_ROW' );

CREATE CHANGE STREAM NewValuesChangeStream
FOR ALL
WITH ( value_capture_type = 'NEW_VALUES' );

Eliminar filtros basados ​​en TTL

Puedes filtrar las eliminaciones basadas en TTL de del ámbito de tu flujo de cambios usando el filtro exclude_ttl_deletes.

Para obtener más información sobre cómo funciona este filtro, consulta Filtro de borrado basado en el tiempo de actividad.

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
OPTIONS (exclude_ttl_deletes = true)

CREATE CHANGE STREAM NewFilterChangeStream FOR ALL
OPTIONS (exclude_ttl_deletes = true)

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
WITH (exclude_ttl_deletes = true)

CREATE CHANGE STREAM NewFilterChangeStream FOR ALL
WITH (exclude_ttl_deletes = true)

Para agregar o quitar el filtro de eliminaciones basadas en TTL de un flujo de cambios existente, consulta Cómo modificar el filtro de eliminaciones basadas en TTL. Puede confirmar sus filtros de flujo de cambios viendo las definiciones del flujo de cambios como DDL.

Filtrar por tipo de modificación de la tabla

Filtra una o más de estas modificaciones de la tabla del alcance de tu flujo de cambios con las siguientes opciones de filtro disponibles:

• exclude_insert: Excluye todas las modificaciones de la tabla INSERT.
• exclude_update: excluir todas las modificaciones de la tabla UPDATE
• exclude_delete: Excluye todas las modificaciones de la tabla DELETE.

Para obtener más información sobre cómo funcionan estos filtros, consulta Filtros de tipo de modificación de la tabla.

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
OPTIONS (MOD_TYPE_FILTER_NAME = true)

CREATE CHANGE STREAM NewFilterChangeStream FOR ALL
OPTIONS (exclude_insert = true, exclude_update = true)

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
WITH (MOD_TYPE_FILTER_NAME = true)

CREATE CHANGE STREAM NewFilterChangeStream FOR ALL
WITH (exclude_insert = true, exclude_update = true)

Para agregar o quitar un filtro de tipo de modificación de tabla de un flujo de cambios existente, consulta Cómo modificar el filtro por tipo de modificación de tabla. Puede confirmar qué filtros de tipo de modificación de tabla existen para su flujo de cambios viendo las definiciones del flujo de cambios como DDL.

Habilitar la exclusión de registros a nivel de transacción

Puede habilitar sus flujos de cambios para excluir registros de transacciones de escritura específicas configurando la opción allow_txn_exclusion al crear un flujo de cambios, o modificando un flujo de cambios existente .

Para obtener más información sobre cómo funciona esta opción, consulta Exclusión de registros a nivel de la transacción.

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
OPTIONS (allow_txn_exclusion = true)

CREATE CHANGE STREAM NewChangeStream FOR ALL
OPTIONS (allow_txn_exclusion = true)

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
WITH (allow_txn_exclusion = true)

CREATE CHANGE STREAM NewChangeStream FOR ALL
WITH (allow_txn_exclusion = true)

Para habilitar o inhabilitar la exclusión de registros a nivel de la transacción de un flujo de cambios existente, consulta Cómo modificar la exclusión de registros a nivel de la transacción. Para comprobar la configuración de esta opción, consulte ver definiciones de flujos de cambios como DDL.

Cómo especificar una transacción de escritura que se excluirá de los flujos de cambios

Para especificar que se excluya una transacción de escritura de los flujos de cambios, debes establecer el parámetro exclude_txn_from_change_streams en true. Los siguientes ejemplos de código muestran cómo se puede especificar que una transacción de escritura se excluya de los flujos de cambios utilizando la biblioteca cliente.

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/spanner"
)

// readWriteTxnExcludedFromChangeStreams executes the insert and update DMLs on
// Singers table excluded from tracking change streams with ddl option
// allow_txn_exclusion = true.
func readWriteTxnExcludedFromChangeStreams(w io.Writer, db string) error {
	// db = `projects/<project>/instances/<instance-id>/database/<database-id>`
	ctx := context.Background()
	client, err := spanner.NewClient(ctx, db)
	if err != nil {
		return fmt.Errorf("readWriteTxnExcludedFromChangeStreams.NewClient: %w", err)
	}
	defer client.Close()

	_, err = client.ReadWriteTransactionWithOptions(ctx, func(ctx context.Context, txn *spanner.ReadWriteTransaction) error {
		stmt := spanner.Statement{
			SQL: `INSERT Singers (SingerId, FirstName, LastName)
					VALUES (111, 'Virginia', 'Watson')`,
		}
		_, err := txn.Update(ctx, stmt)
		if err != nil {
			return fmt.Errorf("readWriteTxnExcludedFromChangeStreams.Update: %w", err)
		}
		fmt.Fprintln(w, "New singer inserted.")
		stmt = spanner.Statement{
			SQL: `UPDATE Singers SET FirstName = 'Hi' WHERE SingerId = 111`,
		}
		_, err = txn.Update(ctx, stmt)
		if err != nil {
			return fmt.Errorf("readWriteTxnExcludedFromChangeStreams.Update: %w", err)
		}
		fmt.Fprint(w, "Singer first name updated.")
		return nil
	}, spanner.TransactionOptions{ExcludeTxnFromChangeStreams: true})
	if err != nil {
		return err
	}
	return nil
}

static void readWriteTxnExcludedFromChangeStreams(DatabaseClient client) {
  // Exclude the transaction from allowed tracking change streams with alloww_txn_exclusion=true.
  // This exclusion will be applied to all the individual operations inside this transaction.
  client
      .readWriteTransaction(Options.excludeTxnFromChangeStreams())
      .run(
          transaction -> {
            transaction.executeUpdate(
                Statement.of(
                    "INSERT Singers (SingerId, FirstName, LastName)\n"
                        + "VALUES (1341, 'Virginia', 'Watson')"));
            System.out.println("New singer inserted.");

            transaction.executeUpdate(
                Statement.of("UPDATE Singers SET FirstName = 'Hi' WHERE SingerId = 111"));
            System.out.println("Singer first name updated.");

            return null;
          });
}

Modifica un flujo de cambios

Para modificar la configuración de un flujo de cambios, usa una declaración DDL ALTER CHANGE STREAM. Utiliza una sintaxis similar a CREATE CHANGE STREAM. Puedes cambiar las columnas que monitoriza un flujo de datos o la duración de su período de retención de datos. También puedes suspender su observación por completo y conservar sus registros de cambios de datos.

Modificar lo que observa una transmisión de cambios

Este ejemplo agrega toda la tabla Songs al flujo de cambios NamesAndAlbums configurado anteriormente:

ALTER CHANGE STREAM NamesAndAlbums
SET FOR Singers(FirstName, LastName), Albums, Songs;

Spanner reemplaza el comportamiento del flujo de cambios especificado con la nueva configuración una vez finalizada la operación de larga duración que actualiza la definición del flujo de cambios dentro del esquema de la base de datos.

Cómo modificar el período de retención de datos de un flujo de cambios

Para modificar el período durante el cual un flujo de cambios conserva sus registros internos, establece retention_period en una instrucción DDL de ALTER CHANGE STREAM.

En este ejemplo, se ajusta el período de retención de datos al flujo de cambios NamesAndAlbums creado anteriormente:

ALTER CHANGE STREAM NamesAndAlbums
SET OPTIONS ( retention_period = '36h' );

ALTER CHANGE STREAM NamesAndAlbums
SET ( retention_period = '36h' );

Modifica el tipo de captura de valor de un flujo de cambios

Para modificar el tipo de captura de valores de un flujo de cambios, establece la cláusula value_capture_type en una declaración DDL ALTER CHANGE STREAM.

Este ejemplo ajusta el tipo de captura de valor a NEW_VALUES.

ALTER CHANGE STREAM NamesAndAlbums
SET OPTIONS ( value_capture_type = 'NEW_VALUES' );

ALTER CHANGE STREAM NamesAndAlbums
SET ( value_capture_type = 'NEW_VALUES' );

Cómo modificar el filtro de eliminaciones basadas en el TTL

Para modificar el filtro TTL-based deletes para un flujo de cambios, establezca el filtro exclude_ttl_deletes en una instrucción DDL ALTER CHANGE STREAM. Puedes usarlo para agregar o quitar el filtro de tus flujos de cambios existentes.

Para obtener más información sobre cómo funcionan estos filtros, consulta Filtro de borrado basado en el tiempo de actividad.

Agrega el filtro de borrados basados en TTL a un flujo de cambios existente

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET OPTIONS (exclude_ttl_deletes = true)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET OPTIONS (exclude_ttl_deletes = true)

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET (exclude_ttl_deletes = true)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET (exclude_ttl_deletes = true)

Quita el filtro de borrados basados en TTL de un flujo de cambios existente

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET OPTIONS (exclude_ttl_deletes = false)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET OPTIONS (exclude_ttl_deletes = false)

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET (exclude_ttl_deletes = false)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET (exclude_ttl_deletes = false)

Modifica el filtro por tipo de modificación de la tabla

Para modificar los filtros de tipo de modificación de la tabla para un flujo de cambios, establece el tipo de filtro en una instrucción DDL ALTER CHANGE STREAM. Puedes usar este método para agregar un filtro nuevo a tu flujo de cambios o quitar uno existente.

Agrega un filtro de tipo de modificación de tabla a un flujo de cambios existente

ALTER CHANGE STREAM CHANGE_STREAM_NAME
SET OPTIONS (MOD_TYPE_FILTER_NAME = true)

ALTER CHANGE STREAM NewFilterChangeStream
SET OPTIONS (exclude_delete = true)

ALTER CHANGE STREAM CHANGE_STREAM_NAME
SET (MOD_TYPE_FILTER_NAME = true)

ALTER CHANGE STREAM NewFilterChangeStream
SET (exclude_delete = true)

Quita un filtro de tipo de modificación de tabla de un flujo de cambios existente

ALTER CHANGE STREAM CHANGE_STREAM_NAME
SET OPTIONS (MOD_TYPE_FILTER_NAME = false)

ALTER CHANGE STREAM NewFilterChangeStream
SET OPTIONS (exclude_delete = false)

ALTER CHANGE STREAM CHANGE_STREAM_NAME
SET (MOD_TYPE_FILTER_NAME = false)

ALTER CHANGE STREAM NewFilterChangeStream
SET (exclude_delete = false)

Se modificó el flujo de cambios para permitir la exclusión de registros a nivel de la transacción

Puedes modificar tu flujo de cambios para permitir que excluya registros de transacciones de escritura específicas. Para ello, establezca la opción allow_txn_exclusion en true en una instrucción DDL ALTER CHANGE STREAM. Si no estableces esta opción o si la estableces en false, el flujo de cambios supervisa todas las transacciones de escritura.

Para obtener más información sobre cómo funciona esta opción, consulte Exclusión de registros a nivel de transacción.

Habilita la exclusión de registros a nivel de la transacción para un flujo de cambios existente

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET OPTIONS (allow_txn_exclusion = true)

ALTER CHANGE STREAM NewAllowedChangeStream FOR ALL
SET OPTIONS (allow_txn_exclusion = true)

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET (allow_txn_exclusion = true)

ALTER CHANGE STREAM NewAllowedChangeStream FOR ALL
SET (allow_txn_exclusion = true)

Deshabilitar la exclusión de registros a nivel de transacción para un flujo de cambios existente

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET OPTIONS (allow_txn_exclusion = false)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET OPTIONS (allow_txn_exclusion = false)

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET (allow_txn_exclusion = false)

ALTER CHANGE STREAM NewAllowedChangeStream FOR ALL
SET (allow_txn_exclusion = false)

Suspender una secuencia de cambios

Si desea que un flujo de cambios detenga su actividad pero conserve sus registros internos, al menos durante el período de retención de datos, puede modificarlo para que no supervise nada.

Para ello, emita una instrucción DDL ALTER CHANGE STREAM que reemplace la definición del flujo de cambios con la frase especial DROP FOR ALL. Por ejemplo:

ALTER CHANGE STREAM MyStream DROP FOR ALL;

El flujo de datos sigue existiendo en la base de datos, pero no monitoriza ningún objeto y no genera más registros de cambios de datos. Los registros de cambios existentes permanecen intactos, sujetos a la política de retención de datos del flujo.

Para reanudar una transmisión suspendida, emita otra instrucción ALTER CHANGE STREAM con su configuración anterior.

Borra un flujo de cambios

Para eliminar permanentemente un flujo de cambios, emita una instrucción DROP CHANGE STREAM que incluya el nombre del flujo:

DROP CHANGE STREAM NamesAndAlbums;

Spanner detiene inmediatamente el flujo de datos, lo elimina del esquema de la base de datos y borra los registros de cambios de datos.

Cómo enumerar y ver flujos de cambios

La consola de Google Cloud proporciona una interfaz web para enumerar y revisar las definiciones de flujos de cambios de una base de datos. También puede visualizar la estructura de los flujos de cambios como sus sentencias DDL equivalentes, o consultando el esquema de información de la base de datos.

Visualice los flujos de cambios con Google Cloud console

Para ver una lista de los flujos de cambios de una base de datos y revisar sus definiciones:

1. Visite la página de instancias de Spanner de la consolaGoogle Cloud .

   Abrir la página de instancias

2. Navegue hasta la instancia y la base de datos adecuadas.

3. En el menú de navegación, haz clic en flujos de cambios.

Se muestra una lista de todos los flujos de cambios de esa base de datos y se resume la configuración de cada uno. Si haces clic en el nombre de un flujo, se revelarán más detalles sobre las tablas y las columnas que supervisa.

Cómo ver las definiciones de los flujos de cambios como DDL

Al visualizar el esquema de una base de datos como DDL, se incluyen descripciones de todos sus flujos de cambios, donde aparecen como sentencias CREATE CHANGE STREAM.

• Para hacerlo desde la consola, haz clic en el vínculo Mostrar DDL equivalente en la página de la base de datos en la consola de Google Cloud .

• Para hacerlo desde la línea de comandos, usa el comando ddl describe de Google Cloud CLI.

Cómo consultar el esquema de información sobre los flujos de cambios

Puedes consultar directamente el esquema de información de una base de datos sobre sus flujos de cambios. Las siguientes tablas contienen los metadatos que definen los nombres de los flujos de cambios, las tablas y columnas que supervisan y sus períodos de retención:

• information_schema.change_streams

• information_schema.change_stream_tables

• information_schema.change_stream_columns

• information_schema.change_stream_options

Prácticas recomendadas para la gestión de flujos de cambios

A continuación, se incluyen algunas prácticas recomendadas para configurar y administrar flujos de cambios.

Considere una base de datos de metadatos independiente.

Los flujos de cambios usan una base de datos de metadatos para mantener el estado interno. La base de datos de metadatos puede ser la misma que la base de datos que contiene los flujos de cambios o puede ser diferente. Te recomendamos que crees una base de datos independiente para el almacenamiento de metadatos.

El conector de flujos de cambios de Spanner necesita permisos de lectura y escritura para la base de datos de metadatos. No es necesario preparar esta base de datos con un esquema; el conector se encarga de ello.

Usar una base de datos de metadatos independiente elimina las complejidades que podrían surgir si se permite que el conector escriba directamente en la base de datos de tu aplicación:

• Al separar la base de datos de metadatos de la base de datos de producción con el flujo de cambios, el conector solo necesita permisos de lectura para la base de datos de producción.

• Al restringir el tráfico del conector a una base de datos de metadatos independiente, las escrituras realizadas por el propio conector no se incluyen en los flujos de cambios de producción. Esto es especialmente relevante para los flujos de cambios que supervisan toda la base de datos.

Si no se usa una base de datos separada para almacenar los metadatos, recomendamos supervisar el impacto en la CPU del conector de flujos de cambios en sus instancias.

Realiza comparativas de los nuevos flujos de cambios y cambia su tamaño si es necesario

Antes de agregar nuevos flujos de cambios a su instancia de producción, considere realizar pruebas de rendimiento con una carga de trabajo realista en una instancia de prueba con los flujos de cambios habilitados. Esto te permite determinar si necesitas agregar nodos a tu instancia para aumentar sus capacidades de procesamiento y almacenamiento.

Ejecute estas pruebas hasta que las métricas de CPU y almacenamiento se estabilicen. De manera óptima, el uso de CPU de la instancia debe permanecer por debajo de los máximos recomendados, y su uso de almacenamiento no debe exceder el límite de almacenamiento de la instancia.

Utilice diferentes regiones para equilibrar la carga.

Cuando utilice flujos de cambios en una configuración de instancia multirregión , considere ejecutar sus canalizaciones de procesamiento en una región diferente a la región líder predeterminada. Esto ayuda a distribuir la carga de transmisión entre las réplicas que no son líderes. Sin embargo, si necesita priorizar la menor latencia de transmisión posible sobre el equilibrio de carga, ejecute la carga de transmisión en la región líder.

¿Qué sigue?

• Crea canalizaciones de flujos de cambios con Dataflow.

• Utilice templates para conectar rápidamente un flujo de cambios a BigQuery, Cloud Storage o Pub/Sub.