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”.

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

  • Esta plantilla no propaga los cambios de esquema de Spanner a BigQuery. Debido a que realizar un cambio de esquema en Spanner probablemente dañe la canalización, es posible que debas volver a crear la canalización después del cambio de esquema.
  • 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.

Parámetros de la plantilla

Parámetro Descripción
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.
spannerDatabaseRole Opcional: 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 SELECT en la transmisión de cambios y el privilegio EXECUTE en la función de lectura de la transmisión de cambios. Si deseas obtener más información, consulta Control de acceso detallado para 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.
spannerProjectId Opcional: 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.
spannerMetadataTableName Opcional: El nombre de la tabla de metadatos del conector de flujos de cambios de Spanner que se usará. Si no se proporciona, una tabla de metadatos de conectores de transmisión 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 Opcional: La prioridad de solicitud para llamadas de Spanner. El valor debe ser uno de los siguientes: HIGH, MEDIUM o LOW. El valor predeterminado es HIGH.
startTimestamp Opcional: El valor de DateTime de inicio, inclusive, que se usará para leer los flujos de cambios. Ej.: 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 Opcional: El valor de DateTime de finalización, inclusive, que se usará para leer los flujos de cambios. Ej.: 2021-10-12T07:20:50.52Z. El valor predeterminado es un tiempo infinito en el futuro.
bigQueryProjectId Opcional: el proyecto de BigQuery. El valor predeterminado es el proyecto para el trabajo de Dataflow.
bigQueryChangelogTableNameTemplate Opcional: la plantilla para el nombre de las tablas de registros de cambios de BigQuery. La configuración predeterminada es _metadata_spanner_table_name_changelog.
deadLetterQueueDirectory Opcional: La ruta para almacenar los registros no procesados que no se pudieron procesar. La ruta predeterminada es un directorio en la ubicación temporal del trabajo de Dataflow. Por lo general, el valor predeterminado es suficiente.
dlqRetryMinutes Opcional: 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 agrega 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.
useStorageWriteApi Opcional: Si es true, la canalización usa la API de BigQuery Storage Write. El valor predeterminado es false. Para obtener más información, consulta Usa la API de BigQuery Storage Write.
useStorageWriteApiAtLeastOnce Opcional: Cuando usas la API de Storage Write, se especifica la semántica de escritura (opcional). Para usar una semántica de al menos una vez, establece 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 Opcional: 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.
storageWriteApiTriggeringFrequencySec Opcional: 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

Consola

  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 HTTP POST. 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?