La plantilla de Datastream a BigQuery es una canalización de transmisión que lee datos de Datastream y los replica en BigQuery. La plantilla lee los datos de Cloud Storage mediante notificaciones de Pub/Sub y los replica en una tabla de etapa de pruebas de BigQuery particionada por tiempo. Después de la replicación, la plantilla ejecuta una MERGE en BigQuery para actualizar todos los cambios de captura de datos modificados (CDC) en una réplica de la tabla de origen.
La plantilla controla la creación y la actualización de las tablas de BigQuery que administra la replicación. Cuando se requiere un lenguaje de definición de datos (DDL), una devolución de llamada a Datastream extrae el esquema de la tabla de origen y lo traduce a los tipos de datos de BigQuery. Las operaciones admitidas incluyen las siguientes:
- Las tablas nuevas se crean a medida que se insertan los datos.
- Se agregan columnas nuevas a las tablas de BigQuery con valores iniciales nulos.
- Las columnas descartadas se ignoran en BigQuery y los valores futuros son nulos.
- Las columnas cuyos nombres se han cambiado se agregan a BigQuery como columnas nuevas.
- Los cambios de tipo no se propagan a BigQuery.
Se recomienda ejecutar esta canalización con el modo de transmisión al menos una vez, ya que la plantilla realiza la anulación de duplicación cuando combina datos de una tabla de BigQuery temporal con la tabla principal de BigQuery. Este paso en la canalización significa que no hay beneficios adicionales en usar el modo de transmisión “exactamente una vez”.
Requisitos de la canalización
- Una transmisión de Datastream que está lista para replicar los datos o ya los está replicando.
- Las notificaciones de Pub/Sub de Cloud Storage están habilitadas para los datos de Datastream.
- Se crean los conjuntos de datos de destino de BigQuery y se les otorgó acceso de administrador a la cuenta de servicio de Compute Engine.
- En la tabla de origen, se necesita una clave primaria para crear la réplica de destino.
- Una base de datos de origen de MySQL o de Oracle. Las bases de datos de PostgreSQL y SQL Server no son compatibles.
Parámetros de la plantilla
Parámetros obligatorios
- inputFilePattern: La ubicación del archivo para la salida de archivos de Datastream en Cloud Storage, en el formato
gs://<BUCKET_NAME>/<ROOT_PATH>/. - inputFileFormat: El formato de los archivos de salida que produce Datastream. Los valores permitidos son
avroyjson. La configuración predeterminada esavro. - gcsPubSubSubscription: Es la suscripción a Pub/Sub que usa Cloud Storage para notificar a Dataflow sobre los archivos nuevos disponibles para el procesamiento, en el formato
projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_NAME>. - outputStagingDatasetTemplate: Es el nombre del conjunto de datos que contiene las tablas de etapa de pruebas. Este parámetro admite plantillas, por ejemplo,
{_metadata_dataset}_logomy_dataset_log. Por lo general, este parámetro es un nombre de conjunto de datos. La configuración predeterminada es{_metadata_dataset}. - outputDatasetTemplate: Es el nombre del conjunto de datos que contiene las tablas de réplica. Este parámetro admite plantillas, por ejemplo,
{_metadata_dataset}omy_dataset. Por lo general, este parámetro es un nombre de conjunto de datos. La configuración predeterminada es{_metadata_dataset}. - deadLetterQueueDirectory: Es la ruta que usa Dataflow para escribir el resultado de la cola de mensajes no entregados. Esta ruta no debe ser la misma que la del archivo de salida de Datastream. La configuración predeterminada es
empty.
Parámetros opcionales
- streamName: El nombre o la plantilla del flujo que se consultará para obtener la información del esquema. El valor predeterminado es {_metadata_stream}. Por lo general, el valor predeterminado es suficiente.
- rfcStartDateTime: La fecha y hora de inicio que se usará para recuperar datos de Cloud Storage (https://tools.ietf.org/html/rfc3339). La configuración predeterminada es:
1970-01-01T00:00:00.00Z. - fileReadConcurrency: La cantidad de archivos de DataStream simultáneos que se leerán. La ruta predeterminada es
10. - outputProjectId: Es el ID del proyecto de Google Cloud que contiene los conjuntos de datos de BigQuery en los que se deben generar datos. El valor predeterminado para este parámetro es el proyecto en el que se ejecuta la canalización de Dataflow.
- outputStagingTableNameTemplate: Es la plantilla que se usará para nombrar las tablas de etapa de pruebas. Por ejemplo,
{_metadata_table}La configuración predeterminada es{_metadata_table}_log. - outputTableNameTemplate: Es la plantilla que se usará para el nombre de las tablas de réplica, por ejemplo,
{_metadata_table}. La configuración predeterminada es{_metadata_table}. - ignoreFields: Es la lista de campos separados por comas que se deben ignorar en BigQuery. La configuración predeterminada es
_metadata_stream,_metadata_schema,_metadata_table,_metadata_source,_metadata_tx_id,_metadata_dlq_reconsumed,_metadata_primary_keys,_metadata_error,_metadata_retry_count. Por ejemplo,_metadata_stream,_metadata_schema. - mergeFrequencyMinutes: Es la cantidad de minutos entre combinaciones para una tabla determinada. La configuración predeterminada es
5. - dlqRetryMinutes: Es la cantidad de minutos entre reintentos de la DLQ. La configuración predeterminada es
10. - dataStreamRootUrl: Es la URL raíz de la API de Datastream. La configuración predeterminada es https://datastream.googleapis.com/.
- applyMerge: Indica si se deben inhabilitar las consultas MERGE para la tarea. La configuración predeterminada es
true. - mergeConcurrency: Es la cantidad de consultas MERGE simultáneas de BigQuery. Solo es eficaz cuando applyMerge se establece como verdadero. La configuración predeterminada es
30. - partitionRetentionDays: Es la cantidad de días que se usarán para la retención de particiones cuando se ejecuten las combinaciones de BigQuery. La configuración predeterminada es
1. - useStorageWriteApiAtLeastOnce: Este parámetro solo se aplica si
Use BigQuery Storage Write APIestá habilitado. Si estrue, se usa una semántica de al menos una vez para la API de Storage Write. De lo contrario, se usa la semántica de “exactamente una vez”. La configuración predeterminada esfalse. - javascriptTextTransformGcsPath: El URI de Cloud Storage del archivo .js que define la función definida por el usuario (UDF) de JavaScript que se usará. Por ejemplo,
gs://my-bucket/my-udfs/my_file.js - javascriptTextTransformFunctionName: Es el nombre de la función definida por el usuario (UDF) de JavaScript que se usará. Por ejemplo, si el código de tu función de JavaScript es
myTransform(inJson) { /*...do stuff...*/ }, el nombre de la función esmyTransform. Para ver ejemplos de UDF de JavaScript, consulta Ejemplos de UDF (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples). - javascriptTextTransformReloadIntervalMinutes: Especifica la frecuencia en minutos con la que se debe volver a cargar la UDF. Si el valor es mayor que 0, Dataflow comprueba de forma periódica el archivo de UDF en Cloud Storage y vuelve a cargar la UDF si el archivo se modifica. Este parámetro te permite actualizar la UDF mientras se ejecuta la canalización, sin necesidad de reiniciar el trabajo. Si el valor es
0, se inhabilita la carga de UDF. El valor predeterminado es0. - pythonTextTransformGcsPath: El patrón de ruta de acceso de Cloud Storage para el código de Python que contiene las funciones definidas por el usuario. Por ejemplo,
gs://your-bucket/your-transforms/*.py - pythonRuntimeVersion: Es la versión del entorno de ejecución que se usará para esta UDF de Python.
- pythonTextTransformFunctionName: Es el nombre de la función a la que se llamará desde el archivo JavaScript. Usa solo letras, dígitos y guiones bajos. Por ejemplo,
transform_udf1 - runtimeRetries: Es la cantidad de veces que se reintentará un entorno de ejecución antes de que falle. La configuración predeterminada es 5.
- 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). - numStorageWriteApiStreams: Cuando usas la API de Storage Write, se especifica la cantidad de transmisiones de escritura. Si
useStorageWriteApiestrueyuseStorageWriteApiAtLeastOnceesfalse, 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
useStorageWriteApiestrueyuseStorageWriteApiAtLeastOnceesfalse, debes configurar este parámetro.
Función definida por el usuario
Para extender esta plantilla, puedes escribir una función definida por el usuario (UDF). La plantilla llama a la UDF para cada elemento de entrada. Las cargas útiles de elementos se serializan como cadenas JSON. Para obtener más información, consulta Crea funciones definidas por el usuario para plantillas de Dataflow.
Especificación de la función
La UDF tiene la siguiente especificación:
Ejecuta la plantilla
- Ve a la página Crear un trabajo a partir de una plantilla de Dataflow. Ir a Crear un trabajo a partir de una plantilla
- En el campo Nombre del trabajo, ingresa un nombre de trabajo único.
- 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.
- En el menú desplegable Plantilla de Dataflow, selecciona the Datastream to BigQuery template.
- En los campos de parámetros proporcionados, ingresa los valores de tus parámetros.
- Opcional: Para cambiar del procesamiento “exactamente una vez” al modo de transmisión al menos una vez, selecciona Al menos una vez.
- Haz clic en Ejecutar trabajo.
En tu shell o terminal, ejecuta la plantilla:
gcloud dataflow flex-template runJOB_NAME \ --project=PROJECT_ID \ --region=REGION_NAME \ --enable-streaming-engine \ --template-file-gcs-location=gs://dataflow-templates-REGION_NAME /VERSION /flex/Cloud_Datastream_to_BigQuery \ --parameters \ inputFilePattern=GCS_FILE_PATH ,\ gcsPubSubSubscription=GCS_SUBSCRIPTION_NAME ,\ outputStagingDatasetTemplate=BIGQUERY_DATASET ,\ outputDatasetTemplate=BIGQUERY_DATASET ,\ outputStagingTableNameTemplate=BIGQUERY_TABLE ,\ outputTableNameTemplate=BIGQUERY_TABLE _log
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 elijasREGION_NAME: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo,us-central1VERSION: the version of the template that you want to useYou can use the following values:
latestto use the latest version of the template, which is available in the non-dated parent folder in the bucket— gs://dataflow-templates-REGION_NAME/latest/- the version name, like
2023-09-12-00_RC00, to use a specific version of the template, which can be found nested in the respective dated parent folder in the bucket— gs://dataflow-templates-REGION_NAME/
GCS_FILE_PATH: es la ruta de acceso de Cloud Storage a los datos de Datastream. Por ejemplo:gs://bucket/path/to/data/GCS_SUBSCRIPTION_NAME: es la suscripción de Pub/Sub desde la que se leen los archivos modificados. Por ejemplo:projects/my-project-id/subscriptions/my-subscription-id.BIGQUERY_DATASET: es el nombre de tu conjunto de datos de BigQuery.BIGQUERY_TABLE: es la plantilla de tabla de BigQuery. Por ejemplo,{_metadata_schema}_{_metadata_table}_log
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": { "inputFilePattern": "GCS_FILE_PATH ", "gcsPubSubSubscription": "GCS_SUBSCRIPTION_NAME ", "outputStagingDatasetTemplate": "BIGQUERY_DATASET ", "outputDatasetTemplate": "BIGQUERY_DATASET ", "outputStagingTableNameTemplate": "BIGQUERY_TABLE ", "outputTableNameTemplate": "BIGQUERY_TABLE _log" }, "containerSpecGcsPath": "gs://dataflow-templates-LOCATION /VERSION /flex/Cloud_Datastream_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 elijasLOCATION: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo,us-central1VERSION: the version of the template that you want to useYou can use the following values:
latestto use the latest version of the template, which is available in the non-dated parent folder in the bucket— gs://dataflow-templates-REGION_NAME/latest/- the version name, like
2023-09-12-00_RC00, to use a specific version of the template, which can be found nested in the respective dated parent folder in the bucket— gs://dataflow-templates-REGION_NAME/
GCS_FILE_PATH: es la ruta de acceso de Cloud Storage a los datos de Datastream. Por ejemplo:gs://bucket/path/to/data/GCS_SUBSCRIPTION_NAME: es la suscripción de Pub/Sub desde la que se leen los archivos modificados. Por ejemplo:projects/my-project-id/subscriptions/my-subscription-id.BIGQUERY_DATASET: es el nombre de tu conjunto de datos de BigQuery.BIGQUERY_TABLE: es la plantilla de tabla de BigQuery. Por ejemplo,{_metadata_schema}_{_metadata_table}_log
Código fuente de la plantilla
¿Qué sigue?
- Obtén información para implementar Datastream y Dataflow para estadísticas.
- Obtén información sobre las plantillas de Dataflow.
- Consulta la lista de plantillas que proporciona Google.