Plantilla de cambios de Bigtable a BigQuery

La plantilla de cambios en Bigtable a BigQuery es un flujo de procesamiento en streaming que transmite registros de cambios de datos de Bigtable y los escribe en tablas de BigQuery mediante Dataflow.

Un flujo de cambios de Bigtable te permite suscribirte a mutaciones de datos por tabla. Cuando te suscribes a los flujos de cambios de una tabla, se aplican las siguientes restricciones:

  • Solo se devuelven las celdas modificadas y los descriptores de las operaciones de eliminación.
  • Solo se devuelve el nuevo valor de una celda modificada.

Cuando se escriben registros de cambios de datos en BigQuery, es posible que las filas se inserten en un orden diferente al orden original de las marcas de tiempo de confirmación de Bigtable.

Las filas de la tabla de registro de cambios que no se pueden escribir en BigQuery debido a un error persistente se colocan de forma permanente en un directorio de cola de mensajes fallidos (cola de mensajes no procesados) de Cloud Storage para que el usuario las revise o las procese más adelante.

Si la tabla de BigQuery necesaria no existe, la canalización la crea. De lo contrario, se usará una tabla de BigQuery ya disponible. El esquema de las tablas de BigQuery debe contener las columnas de la siguiente tabla.

Cada fila nueva de BigQuery incluye un registro de cambio de datos devuelto por el flujo de cambios de su fila correspondiente en tu tabla de Bigtable.

Esquema de la tabla de salida de BigQuery

Nombre de la columna Tipo Admite valores nulos Descripción
row_key STRING o BYTES No Clave de la fila modificada. Si la opción de canalización writeRowkeyAsBytes se define como true, el tipo de columna debe ser BYTES. De lo contrario, usa el tipo STRING.
mod_type STRING No El tipo de mutación de la fila. Usa uno de los siguientes valores: SET_CELL, DELETE_CELLS o DELETE_FAMILY.
column_family STRING No La familia de columnas afectada por la mutación de la fila.
column STRING Sí El calificador de columna afectado por la mutación de la fila. En el tipo de mutación DELETE_FAMILY, asigna el valor NULL.
commit_timestamp TIMESTAMP No La hora en la que Bigtable aplica la mutación.
big_query_commit_timestamp TIMESTAMP Sí Opcional: especifica la hora a la que BigQuery escribe la fila en una tabla de salida. El campo no se rellena si el nombre de la columna está presente en el valor de la opción de canalización bigQueryChangelogTableFieldsToIgnore.
timestamp TIMESTAMP o INT64 Sí Valor de marca de tiempo de la celda afectada por la mutación. Si la opción de canalización writeNumericTimestamps se define como true, el tipo de columna debe ser INT64. Si no, usa el tipo TIMESTAMP. En los tipos de mutación DELETE_CELLS y DELETE_FAMILY, se establece en NULL.
timestamp_from TIMESTAMP o INT64 Sí Describe el inicio inclusivo del intervalo de marca de tiempo de todas las celdas eliminadas por la mutación DELETE_CELLS. En el caso de otros tipos de mutaciones, asigna el valor NULL.
timestamp_to TIMESTAMP o INT64 Sí Describe el final exclusivo del intervalo de marca de tiempo de todas las celdas eliminadas por la mutación DELETE_CELLS. En el caso de otros tipos de mutaciones, asigna el valor NULL.
is_gc BOOL No Opcional: Si la mutación se activa mediante una política de recogida de elementos no utilizados, se debe definir como true. En todos los demás casos, se asigna el valor false. El campo no se rellena cuando el nombre de la columna está presente en el valor de la opción de canalización bigQueryChangelogTableFieldsToIgnore.
source_instance STRING No Opcional: describe el nombre de la instancia de Bigtable de la que procede la mutación. El campo no se rellena cuando el nombre de la columna está presente en el valor de la opción de canalización bigQueryChangelogTableFieldsToIgnore.
source_cluster STRING No Opcional: describe el nombre del clúster de Bigtable del que procede la mutación. El campo no se rellena cuando el nombre de la columna está presente en el valor de la opción de canalización bigQueryChangelogTableFieldsToIgnore.
source_table STRING No Opcional: describe el nombre de la tabla de Bigtable a la que se aplica la mutación. El valor de esta columna puede ser útil si varias tablas de Bigtable envían cambios a la misma tabla de BigQuery. El campo no se rellena cuando el nombre de la columna está presente en el valor de la opción de canalización bigQueryChangelogTableFieldsToIgnore.
tiebreaker INT64 No Opcional: Cuando dos mutaciones se registran al mismo tiempo en diferentes clústeres de Bigtable, se aplica a la tabla de origen la mutación con el valor tiebreaker más alto. Las mutaciones con valores de tiebreaker más bajos se descartan. El campo no se rellena cuando el nombre de la columna está presente en el valor de la opción de canalización bigQueryChangelogTableFieldsToIgnore.
value STRING o BYTES Sí El nuevo valor definido por la mutación. Si la opción de canalización writeValuesAsBytes se define como true, el tipo de columna debe ser BYTES. De lo contrario, usa el tipo STRING. El valor se asigna a las mutaciones de SET_CELL. En el caso de otros tipos de mutación, el valor se define como NULL.

Requisitos del flujo de procesamiento

  • La instancia de origen de Bigtable especificada.
  • La tabla de origen de Bigtable especificada. La tabla debe tener habilitados los flujos de cambios.
  • El perfil de aplicación de Bigtable especificado.
  • El conjunto de datos de destino de BigQuery especificado.

Parámetros de plantilla

Parámetros obligatorios

  • bigQueryDataset nombre del conjunto de datos de la tabla de BigQuery de destino.
  • bigtableChangeStreamAppProfile el ID del perfil de aplicación de Bigtable. El perfil de la aplicación debe usar el enrutamiento de un solo clúster y permitir transacciones de una sola fila.
  • bigtableReadInstanceId el ID de la instancia de Bigtable de origen.
  • bigtableReadTableId el ID de la tabla de Bigtable de origen.

Parámetros opcionales

  • writeRowkeyAsBytes indica si se deben escribir las claves de fila como BYTES de BigQuery. Si se define como true, las claves de las filas se escriben en la columna BYTES. De lo contrario, las claves de fila se escriben en la columna STRING. El valor predeterminado es false.
  • writeValuesAsBytes si se define como true, los valores se escriben en una columna de tipo BYTES; de lo contrario, se escriben en una columna de tipo STRING . El valor predeterminado es false.
  • writeNumericTimestamps indica si se debe escribir la marca de tiempo de Bigtable como INT64 de BigQuery. Si se asigna el valor true, los valores se escriben en la columna INT64. De lo contrario, los valores se escriben en la columna TIMESTAMP. Columnas afectadas: timestamp, timestamp_from y timestamp_to. El valor predeterminado es false. Si se define como true, la hora se mide en microsegundos desde la época de Unix (1 de enero de 1970 en UTC).
  • bigQueryProjectId ID del proyecto del conjunto de datos de BigQuery. El valor predeterminado es el proyecto de la tarea de Dataflow.
  • bigQueryChangelogTableName nombre de la tabla de BigQuery de destino. Si no se especifica, se usa el valor bigtableReadTableId + "_changelog". El valor predeterminado es una cadena vacía.
  • bigQueryChangelogTablePartitionGranularity especifica la granularidad de la partición de la tabla de registro de cambios. Si se define, la tabla se particiona. Usa uno de los siguientes valores admitidos: HOUR, DAY, MONTH o YEAR. De forma predeterminada, la tabla no tiene particiones.
  • bigQueryChangelogTablePartitionExpirationMs define el tiempo de vencimiento de la partición de la tabla de registro de cambios en milisegundos. Si se define como true, se eliminan las particiones que tengan una antigüedad superior al número de milisegundos especificado. De forma predeterminada, no se establece ninguna fecha de vencimiento.
  • bigQueryChangelogTableFieldsToIgnore lista separada por comas de las columnas del registro de cambios que, cuando se especifican, no se crean ni se rellenan. Usa uno de los siguientes valores admitidos: is_gc, source_instance, source_cluster, source_table, tiebreaker o big_query_commit_timestamp. De forma predeterminada, todas las columnas se rellenan.
  • dlqDirectory el directorio que se va a usar para la cola de mensajes fallidos. Los registros que no se pueden procesar se almacenan en este directorio. El valor predeterminado es un directorio de la ubicación temporal de la tarea de Dataflow. En la mayoría de los casos, puede usar la ruta predeterminada.
  • bigtableChangeStreamMetadataInstanceId el ID de instancia de metadatos de los flujos de cambios de Bigtable. El valor predeterminado es una cadena vacía.
  • bigtableChangeStreamMetadataTableTableId el ID de la tabla de metadatos del conector de flujos de cambios de Bigtable. Si no se proporciona, se crea automáticamente una tabla de metadatos del conector de flujos de cambios de Bigtable durante la ejecución de la canalización. El valor predeterminado es una cadena vacía.
  • bigtableChangeStreamCharset nombre del conjunto de caracteres de los flujos de cambios de Bigtable. El valor predeterminado es UTF-8.
  • bigtableChangeStreamStartTimestamp marca de tiempo inicial (https://tools.ietf.org/html/rfc3339) (inclusive) que se usará para leer los flujos de cambios. Por ejemplo, 2022-05-05T07:59:59Z. El valor predeterminado es la marca de tiempo de la hora de inicio de la canalización.
  • bigtableChangeStreamIgnoreColumnFamilies lista de cambios en los nombres de las familias de columnas separados por comas que se deben ignorar. El valor predeterminado es una cadena vacía.
  • bigtableChangeStreamIgnoreColumns lista de cambios de nombres de columnas separados por comas que se deben ignorar. Ejemplo: "cf1:col1,cf2:col2". El valor predeterminado es una cadena vacía.
  • bigtableChangeStreamName nombre único de la canalización del cliente. Te permite reanudar el procesamiento desde el punto en el que se detuvo una canalización que se estaba ejecutando. El valor predeterminado es un nombre generado automáticamente. Consulte los registros de trabajos de Dataflow para ver el valor utilizado.
  • bigtableChangeStreamResume si se define como true, una nueva canalización reanuda el procesamiento desde el punto en el que se detuvo una canalización que se estaba ejecutando anteriormente con el mismo valor de bigtableChangeStreamName. Si la canalización con el valor bigtableChangeStreamName proporcionado no se ha ejecutado nunca, no se iniciará ninguna canalización. Si se define como false, se inicia una nueva canalización. Si ya se ha ejecutado una canalización con el mismo valor de bigtableChangeStreamName para la fuente en cuestión, no se iniciará una nueva canalización. El valor predeterminado es false.
  • bigtableReadChangeStreamTimeoutMs tiempo de espera de las solicitudes ReadChangeStream de Bigtable en milisegundos.
  • bigtableReadProjectId el ID del proyecto de Bigtable. El valor predeterminado es el proyecto de la tarea de Dataflow.

Ejecutar la plantilla

Consola

  1. Ve a la página Crear tarea a partir de plantilla de Dataflow.
    2. Ir a Crear tarea a partir de plantilla
  2. En el campo Nombre de la tarea, introduce un nombre único.
  3. Opcional: En Endpoint regional, seleccione un valor en el menú desplegable. La región predeterminada es us-central1.

    Para ver una lista de las regiones en las que puedes ejecutar una tarea de Dataflow, consulta Ubicaciones de Dataflow.

  4. En el menú desplegable Plantilla de flujo de datos, seleccione the Bigtable change streams to BigQuery template.
  5. En los campos de parámetros proporcionados, introduzca los valores de los parámetros.
  6. Haz clic en Ejecutar trabajo.

gcloud

En tu shell o terminal, ejecuta la plantilla:

gcloud dataflow flex-template run JOB_NAME \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_BigQuery \
    --parameters \
bigtableReadInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableReadTableId=BIGTABLE_TABLE_ID,\
bigtableChangeStreamAppProfile=BIGTABLE_APPLICATION_PROFILE_ID,\
bigQueryDataset=BIGQUERY_DESTINATION_DATASET

Haz los cambios siguientes:

  • PROJECT_ID: el ID del proyecto Google Cloud en el que quieres ejecutar la tarea de Dataflow
  • JOB_NAME: un nombre de trabajo único que elijas
  • VERSION: la versión de la plantilla que quieres usar

    Puedes usar los siguientes valores:

    • latest para usar la última versión de la plantilla, que está disponible en la carpeta principal sin fecha del contenedor: 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 encuentra anidada en la carpeta principal correspondiente con la fecha en el bucket: gs://dataflow-templates-REGION_NAME/
  • REGION_NAME: la región en la que quieras desplegar tu trabajo de Dataflow. Por ejemplo, us-central1
  • BIGTABLE_INSTANCE_ID: el ID de tu instancia de Bigtable.
  • BIGTABLE_TABLE_ID: el ID de tu tabla de Bigtable.
  • BIGTABLE_APPLICATION_PROFILE_ID: el ID de tu perfil de aplicación de Bigtable.
  • BIGQUERY_DESTINATION_DATASET: nombre del conjunto de datos de destino de BigQuery

API

Para ejecutar la plantilla mediante la API REST, envía una solicitud HTTP POST. Para obtener más información sobre la API y sus ámbitos 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",
    "containerSpecGcsPath": "gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_BigQuery",
    "parameters": {
        "bigtableReadInstanceId": "BIGTABLE_INSTANCE_ID",
        "bigtableReadTableId": "BIGTABLE_TABLE_ID",
        "bigtableChangeStreamAppProfile": "BIGTABLE_APPLICATION_PROFILE_ID",
        "bigQueryDataset": "BIGQUERY_DESTINATION_DATASET"
    }
  }
}

Haz los cambios siguientes:

  • PROJECT_ID: el ID del proyecto Google Cloud en el que quieres ejecutar la tarea de Dataflow
  • JOB_NAME: un nombre de trabajo único que elijas
  • VERSION: la versión de la plantilla que quieres usar

    Puedes usar los siguientes valores:

    • latest para usar la última versión de la plantilla, que está disponible en la carpeta principal sin fecha del contenedor: 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 encuentra anidada en la carpeta principal correspondiente con la fecha en el bucket: gs://dataflow-templates-REGION_NAME/
  • LOCATION: la región en la que quieras desplegar tu trabajo de Dataflow. Por ejemplo, us-central1
  • BIGTABLE_INSTANCE_ID: el ID de tu instancia de Bigtable.
  • BIGTABLE_TABLE_ID: el ID de tu tabla de Bigtable.
  • BIGTABLE_APPLICATION_PROFILE_ID: el ID de tu perfil de aplicación de Bigtable.
  • BIGQUERY_DESTINATION_DATASET: nombre del conjunto de datos de destino de BigQuery

Siguientes pasos