Plantillas de flujos de cambios de Spanner a BigQuery

La plantilla de flujos de cambios de Spanner a BigQuery es una canalización de transmisión que transmite los registros de cambios de datos de Spanner y los escribe en tablas de BigQuery a través de Dataflow Runner v2.

Todas las columnas observadas por el flujo de cambios se incluyen en cada fila de la tabla de BigQuery, sin importar si una transacción de Spanner las modifica. Las columnas que no se observan no se incluyen en la fila de BigQuery. Cualquier cambio de Spanner que sea menor que la marca de agua de Dataflow se aplica de forma correcta a las tablas de BigQuery o se almacena en la cola de mensajes no entregados para reintentar aplicarlo. Las filas de BigQuery se insertan de forma desordenada en comparación con el orden original de las marcas de tiempo de confirmación de Spanner.

Si las tablas de BigQuery necesarias no existen, la canalización las crea. De lo contrario, se usan las tablas de BigQuery existentes. El esquema de las tablas de BigQuery existentes debe contener las columnas con seguimiento correspondientes de las tablas de Spanner y cualquier columna de metadatos adicionales que la opción ignoreFields no ignore de forma explícita. Consulta la descripción de los campos de metadatos en la siguiente lista. Cada fila nueva de BigQuery incluye todas las columnas observadas por el flujo de cambios de su fila correspondiente en tu tabla de Spanner en la marca de tiempo del registro de cambios.

Los siguientes campos de metadatos se agregan a las tablas de BigQuery. Para obtener más detalles sobre estos campos, consulta Registros de cambios de datos en la sección “Cambia particiones, registros y consultas”.

  • _metadata_spanner_mod_type: el tipo de modificación (inserción, actualización o eliminación) de la transacción de Spanner. Se extrae del registro de cambios de los datos de flujos de cambios.
  • _metadata_spanner_table_name: el nombre de la tabla de Spanner. Este campo no es el nombre de la tabla de metadatos del conector.
  • _metadata_spanner_commit_timestamp: la marca de tiempo de confirmación de Spanner, que es el momento en que se confirma un cambio. Este valor se extrae del registro de cambios de los datos de flujos de cambios.
  • _metadata_spanner_server_transaction_id: Una cadena única a nivel global que representa la transacción de Spanner en la que se confirmó el cambio. Usa este valor solo en el contexto del procesamiento de registros de transmisión de cambios. No está correlacionado con el ID de transacción en la API de Spanner. Este valor se extrae del registro de cambios de los datos de flujos de cambios.
  • _metadata_spanner_record_sequence: el número de secuencia para el registro dentro de la transacción de Spanner. Se garantiza que los números de secuencia son únicos y aumentan monótonamente, pero no necesariamente contiguos, dentro de una transacción. Este valor se extrae del registro de cambios de los datos de flujos de cambios.
  • _metadata_spanner_is_last_record_in_transaction_in_partition: indica si el registro es el último de una transacción de Spanner en la partición actual. Este valor se extrae del registro de cambios de los datos de flujos de cambios.
  • _metadata_spanner_number_of_records_in_transaction: la cantidad de registros de cambios de datos que forman parte de la transacción de Spanner en todas las particiones de transmisión de cambios. Este valor se extrae del registro de cambios de los datos de flujos de cambios.
  • _metadata_spanner_number_of_partitions_in_transaction: La cantidad de particiones que muestran registros de cambios de datos para la transacción de Spanner. Este valor se extrae del registro de cambios de los datos de flujos de cambios.
  • _metadata_big_query_commit_timestamp: la marca de tiempo de confirmación del momento en que se inserta la fila en BigQuery. Si useStorageWriteApi es true, la canalización no crea esta columna de forma automática en la tabla de registros de cambios. En ese caso, debes agregar esta columna de forma manual en la tabla de registro de cambios si es necesario.

Cuando uses esta plantilla, ten en cuenta los siguientes detalles:

  • Puedes usar esta plantilla para propagar columnas nuevas en tablas existentes o tablas nuevas de Spanner a BigQuery. Para obtener más información, consulta Controla la adición de tablas o columnas de seguimiento.
  • Para los tipos de captura de valor OLD_AND_NEW_VALUES y NEW_VALUES, cuando el registro de cambios de datos contiene un cambio UPDATE, la plantilla debe realizar una lectura inactiva en Spanner en la marca de tiempo de confirmación del registro de cambios de datos para recuperar las columnas que sí se observaron, pero no se modificaron. Asegúrate de configurar la base de datos “'version_retention_period” de forma adecuada para la lectura inactiva. Para el tipo de captura del valor NEW_ROW, la plantilla es más eficiente, porque el registro de cambios de datos captura la fila nueva completa, incluidas las columnas que no se actualizan en las solicitudes UPDATE, y la plantilla no necesita realizar una lectura inactiva.
  • Para minimizar la latencia de la red y los costos de transporte de la red, ejecuta el trabajo de Dataflow desde la misma región que tu instancia de Spanner o tus tablas de BigQuery. Si usas fuentes, receptores, ubicaciones de archivos de etapa de pruebas o ubicaciones de archivos temporales que se encuentran fuera de la región del trabajo, es posible que los datos se envíen a través de diferentes regiones. Para obtener más información, consulta Regiones de Dataflow.
  • Esta plantilla admite todos los tipos de datos válidos de Spanner. Si el tipo de BigQuery es más preciso que el tipo de Spanner, podría producirse una pérdida precisión durante la transformación. Específicamente:
    • Para el tipo JSON de Spanner, el orden de los miembros de un objeto se ordena de forma lexicográfica, pero no existe esa garantía para el tipo JSON de BigQuery.
    • Spanner admite el tipo de TIMESTAMP de nanosegundos, pero BigQuery solo admite el tipo de TIMESTAMP de microsegundos.
  • Esta plantilla no admite el uso de la API de BigQuery Storage Write en modo del tipo “exactamente una vez”.

Obtén más información sobre los flujos de cambios, cómo compilar canalizaciones de Dataflow de flujos de cambio y prácticas recomendadas.

Requisitos de la canalización

  • La instancia de Spanner debe existir antes de ejecutar la canalización.
  • La base de datos de Spanner debe existir antes de ejecutar la canalización.
  • La instancia de metadatos de Spanner debe existir antes de ejecutar la canalización.
  • La base de datos de metadatos de Spanner debe existir antes de ejecutar la canalización.
  • El flujo de cambios de Spanner debe existir antes de ejecutar la canalización.
  • El conjunto de datos de BigQuery debe existir antes de ejecutar la canalización.

Controla la adición de tablas o columnas de seguimiento

En esta sección, se describen las prácticas recomendadas para agregar tablas y columnas de seguimiento de Spanner mientras se ejecuta la canalización.

  • Antes de agregar una columna nueva a un alcance de flujo de cambios de Spanner, primero agrégala a la tabla de registro de cambios de BigQuery. La columna agregada debe tener un tipo de datos coincidente y ser NULLABLE. Espera al menos 10 minutos antes de continuar con la creación de la columna o tabla nueva en Spanner. Si escribes en la columna nueva sin esperar, es posible que se genere un registro sin procesar con un código de error no válido en el directorio de la cola de mensajes no entregados.
  • Para agregar una tabla nueva, primero agrégala a la base de datos de Spanner. La tabla se crea automáticamente en BigQuery cuando la canalización recibe un registro para la tabla nueva.
  • Después de agregar las columnas o tablas nuevas a la base de datos de Spanner, asegúrate de alterar tu flujo de cambios para hacer un seguimiento de las columnas o tablas nuevas que deseas si aún no se les hace un seguimiento implícito.
  • La plantilla no descarta tablas ni columnas de BigQuery. Si se quita una columna de la tabla de Spanner, los valores nulos se propagan a las columnas del registro de cambios de BigQuery para los registros generados después de que se quiten las columnas de la tabla de Spanner, a menos que quites la columna de BigQuery de forma manual.
  • La plantilla no admite actualizaciones de tipo de columna. Aunque Spanner admite cambiar una columna STRING a una BYTES o una columna BYTES a una STRING, no puedes modificar el tipo de datos de una columna existente ni usar el mismo nombre de columna con diferentes tipos de datos en BigQuery. Si sueltas y vuelves a crear una columna con el mismo nombre, pero con un tipo diferente en Spanner, es posible que los datos se escriban en la columna de BigQuery existente, pero el tipo no cambiará.
  • Esta plantilla no admite actualizaciones del modo de columna. Las columnas de metadatos replicadas en BigQuery se configuran en el modo REQUIRED. Todas las demás columnas replicadas en BigQuery se establecen en NULLABLE, independientemente de si se definen como NOT NULL en la tabla de Spanner. No puedes actualizar las columnas NULLABLE al modo REQUIRED en BigQuery.
  • No se admite cambiar el tipo de captura de valor de un flujo de cambios para ejecutar canalizaciones.

Parámetros de la plantilla

Parámetros obligatorios

  • spannerInstanceId: La instancia de Spanner desde la que se leerán los flujos de cambios.
  • spannerDatabase: La base de datos de Spanner desde la que se leerán los flujos de cambios.
  • spannerMetadataInstanceId: La instancia de Spanner que se usará para la tabla de metadatos del conector de flujos de cambios.
  • spannerMetadataDatabase: La base de datos de Spanner que se usará para la tabla de metadatos del conector de flujos de cambios.
  • spannerChangeStreamName: El nombre del flujo de cambios de Spanner desde el que se leerá.
  • bigQueryDataset: el conjunto de datos de BigQuery de salida de los flujos de cambios.

Parámetros opcionales

  • spannerProjectId: El proyecto desde el que se leerán los flujos de cambio. Este valor es también el proyecto en el que se crea la tabla de metadatos del conector de flujos de cambios. El valor predeterminado para este parámetro es el proyecto en el que se ejecuta la canalización de Dataflow.
  • spannerDatabaseRole: El rol de base de datos de Spanner que se usará cuando se ejecute la plantilla. Este parámetro solo es necesario cuando el principal de IAM que ejecuta la plantilla es un usuario de control de acceso detallado. El rol de la base de datos debe tener el privilegio SELECCIONAR en la transmisión de cambios y el privilegio EJECUTAR en la función de lectura de la transmisión de cambios. A fin de obtener más información, consulta Control de acceso detallado para los flujos de cambios (https://cloud.google.com/spanner/docs/fgac-change-streams).
  • spannerMetadataTableName: el nombre de la tabla de metadatos del conector de flujos de cambios de Cloud Spanner que se usará. Si no se proporciona, una tabla de metadatos de conectores de flujos de cambios de Spanner se crea de forma automática durante el flujo de la canalización. Debes proporcionar este parámetro cuando actualices una canalización existente. De lo contrario, no proporciones este parámetro.
  • rpcPriority: La prioridad de solicitud para llamadas de Spanner. El valor debe ser uno de estos valores: HIGH, MEDIUM o LOW. El valor predeterminado es HIGH.
  • spannerHost: El extremo de Cloud Spanner al que se llamará en la plantilla. Solo se usa para pruebas. (Ejemplo: https://batch-spanner.googleapis.com).
  • startTimestamp: La fecha y hora de inicio (https://datatracker.ietf.org/doc/html/rfc3339), inclusiva, que se usará para leer los flujos de cambios. Ex-2021-10-12T07:20:50.52Z. El valor predeterminado es la marca de tiempo del inicio de la canalización, es decir, la hora actual.
  • endTimestamp: La fecha y hora de finalización (https://datatracker.ietf.org/doc/html/rfc3339), inclusiva, que se usará para leer los flujos de cambios.Ex-2021- 10-12T07:20:50.52Z. El valor predeterminado es un tiempo infinito en el futuro.
  • bigQueryProjectId: el proyecto de BigQuery. El valor predeterminado es el proyecto para el trabajo de Dataflow.
  • bigQueryChangelogTableNameTemplate: la plantilla para el nombre de la tabla de BigQuery que contiene el registro de cambios. El valor predeterminado es: {_metadata_spanner_table_name}_changelog.
  • deadLetterQueueDirectory: es la ruta para almacenar los registros no procesados. La ruta predeterminada es un directorio en la ubicación temporal del trabajo de Dataflow. Por lo general, el valor predeterminado es suficiente.
  • dlqRetryMinutes: la cantidad de minutos entre reintentos de la cola de mensajes no entregados. El valor predeterminado es 10.
  • ignoreFields: Opcional: una lista de campos separados por comas (distingue mayúsculas de minúsculas) que se deben ignorar. Estos campos pueden ser de tablas observadas o campos de metadatos que agregan la canalización. Los campos ignorados no se insertan en BigQuery. Cuando ignoras el campo _metadata_spanner_table_name, también se ignora el parámetro bigQueryChangelogTableNameTemplate. La configuración predeterminada es vacía.
  • disableDlqRetries: indica si se deben inhabilitar o no los reintentos para la DLQ. La configuración predeterminada es "false".
  • useStorageWriteApi: Si es verdadero, la canalización usa la API de BigQuery Storage Write (https://cloud.google.com/bigquery/docs/write-api). El valor predeterminado es false. Para obtener más información, consulta Usa la API de Storage Write (https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api).
  • useStorageWriteApiAtLeastOnce: Cuando usas la API de Storage Write, se especifica la semántica de escritura. Para usar una semántica de al menos una vez (https://beam.apache.org/documentation/io/built-in/google-bigquery/#at-least-once-semantics), configura este parámetro en true. Para usar una semántica de una y solo una vez, configura el parámetro en false. Este parámetro se aplica solo cuando useStorageWriteApi es true. El valor predeterminado es false.
  • numStorageWriteApiStreams: Cuando usas la API de Storage Write, se especifica la cantidad de transmisiones de escritura. Si useStorageWriteApi es true y useStorageWriteApiAtLeastOnce es false, debes configurar este parámetro. La configuración predeterminada es 0.
  • storageWriteApiTriggeringFrequencySec: cuando se usa la API de Storage Write, se especifica la frecuencia de activación en segundos. Si useStorageWriteApi es true y useStorageWriteApiAtLeastOnce es false, debes configurar este parámetro.

Ejecuta la plantilla

Console

  1. Ve a la página Crear un trabajo a partir de una plantilla de Dataflow.
    2. Ir a Crear un trabajo a partir de una plantilla
  2. En el campo Nombre del trabajo, ingresa un nombre de trabajo único.
  3. Opcional: Para Extremo regional, selecciona un valor del menú desplegable. La región predeterminada es us-central1.

    Para obtener una lista de regiones en las que puedes ejecutar un trabajo de Dataflow, consulta Ubicaciones de Dataflow.

  4. En el menú desplegable Plantilla de Dataflow, selecciona the Cloud Spanner change streams to BigQuery template.
  5. En los campos de parámetros proporcionados, ingresa los valores de tus parámetros.
  6. Haga clic en Ejecutar trabajo.

gcloud

En tu shell o terminal, ejecuta la plantilla:

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Spanner_Change_Streams_to_BigQuery \
    --region REGION_NAME \
    --parameters \
spannerInstanceId=SPANNER_INSTANCE_ID,\
spannerDatabase=SPANNER_DATABASE,\
spannerMetadataInstanceId=SPANNER_METADATA_INSTANCE_ID,\
spannerMetadataDatabase=SPANNER_METADATA_DATABASE,\
spannerChangeStreamName=SPANNER_CHANGE_STREAM,\
bigQueryDataset=BIGQUERY_DATASET

Reemplaza lo siguiente:

  • JOB_NAME: Es el nombre del trabajo que elijas
  • VERSION: Es la versión de la plantilla que deseas usar.

    Puedes usar los siguientes valores:

    • latest para usar la última versión de la plantilla, que está disponible en la carpeta superior non-dated en el bucket gs://dataflow-templates-REGION_NAME/latest/
    • el nombre de la versión, como 2023-09-12-00_RC00, para usar una versión específica de la plantilla, que se puede encontrar anidada en la carpeta superior con fecha correspondiente en el bucket gs://dataflow-templates-REGION_NAME/
  • REGION_NAME: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
  • SPANNER_INSTANCE_ID: ID de la instancia de Spanner
  • SPANNER_DATABASE: base de datos de Spanner
  • SPANNER_METADATA_INSTANCE_ID: ID de la instancia de metadatos de Spanner
  • SPANNER_METADATA_DATABASE: base de datos de metadatos de Spanner
  • SPANNER_CHANGE_STREAM: flujo de cambios de Spanner
  • BIGQUERY_DATASET: El conjunto de datos de BigQuery de salida de los flujos de cambios.

API

Para ejecutar la plantilla con la API de REST, envía una solicitud POST HTTP. Para obtener más información de la API y sus permisos de autorización, consulta projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "spannerInstanceId": "SPANNER_INSTANCE_ID",
          "spannerDatabase": "SPANNER_DATABASE",
          "spannerMetadataInstanceId": "SPANNER_METADATA_INSTANCE_ID",
          "spannerMetadataDatabase": "SPANNER_METADATA_DATABASE",
          "spannerChangeStreamName": "SPANNER_CHANGE_STREAM",
          "bigQueryDataset": "BIGQUERY_DATASET"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Spanner_Change_Streams_to_BigQuery",
   }
}

Reemplaza lo siguiente:

  • PROJECT_ID: El ID del proyecto de Google Cloud en el que deseas ejecutar el trabajo de Dataflow.
  • JOB_NAME: Es el nombre del trabajo que elijas
  • VERSION: Es la versión de la plantilla que deseas usar.

    Puedes usar los siguientes valores:

    • latest para usar la última versión de la plantilla, que está disponible en la carpeta superior non-dated en el bucket gs://dataflow-templates-REGION_NAME/latest/
    • el nombre de la versión, como 2023-09-12-00_RC00, para usar una versión específica de la plantilla, que se puede encontrar anidada en la carpeta superior con fecha correspondiente en el bucket gs://dataflow-templates-REGION_NAME/
  • LOCATION: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
  • SPANNER_INSTANCE_ID: ID de la instancia de Spanner
  • SPANNER_DATABASE: base de datos de Spanner
  • SPANNER_METADATA_INSTANCE_ID: ID de la instancia de metadatos de Spanner
  • SPANNER_METADATA_DATABASE: base de datos de metadatos de Spanner
  • SPANNER_CHANGE_STREAM: flujo de cambios de Spanner
  • BIGQUERY_DATASET: El conjunto de datos de BigQuery de salida de los flujos de cambios.

¿Qué sigue?