Plantilla de conectividad a bases de datos de Java (JDBC) a BigQuery

La plantilla de JDBC a BigQuery es una canalización por lotes que copia los datos de una tabla de base de datos relacional a una tabla de BigQuery existente. Esta canalización utiliza JDBC para conectarse a las bases de datos relacionales. Usa esta plantilla para copiar datos desde cualquier base de datos relacional con controladores de JDBC disponibles en BigQuery.

Para obtener una capa adicional de protección, también puedes pasar una clave de Cloud KMS junto con un nombre de usuario, contraseña y parámetros de string de conexión codificados en Base64 encriptados con la clave de Cloud KMS. Para obtener detalles adicionales sobre la encriptación de tus parámetros de nombre de usuario, contraseña y string de conexión, consulta el extremo de encriptación de la API de Cloud KMS.

Requisitos de la canalización

Los controladores de JDBC de la base de datos relacional deben estar disponibles.
La tabla de BigQuery debe existir antes de ejecutar la canalización.
La tabla de BigQuery debe tener un esquema compatible.
La base de datos relacional debe ser accesible desde la subred en la que se ejecuta Dataflow.

Parámetros de la plantilla

Parámetros obligatorios

driverJars: Lista separada por comas de los archivos JAR del controlador. (Ejemplo: gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar).
driverClassName: El nombre de la clase del controlador de JDBC. (Ejemplo: com.mysql.jdbc.Driver).
connectionURL: la cadena de URL de conexión de JDBC. Por ejemplo, jdbc:mysql://some-host:3306/sampledb Se puede pasar como una string codificada en Base64 y, luego, encriptada con una clave de Cloud KMS. Ten en cuenta la diferencia entre una cadena de conexión de base de datos de Oracle que no es de RAC (jdbc:oracle:thin:@some-host:<port>:<sid>) y una cadena de conexión de base de datos de Oracle RAC (jdbc:oracle:thin:@//some-host[:<port>]/<service_name>). (Ejemplo: jdbc:mysql://some-host:3306/sampledb).
outputTable: La ubicación de la tabla de BigQuery en la que se escribirá el resultado. El nombre debe tener el formato <project>:<dataset>.<table_name>. El esquema de la tabla debe coincidir con los objetos de entrada. (Por ejemplo,
bigQueryLoadingTemporaryDirectory: el directorio temporal para el proceso de carga de BigQuery (Ejemplo: gs://your-bucket/your-files/temp_dir).

Parámetros opcionales

connectionProperties: La cadena de propiedades para usar en la conexión de JDBC. El formato de la string debe ser [propertyName=property;]*. (Ejemplo: unicode=true;characterEncoding=UTF-8).
username: El nombre de usuario para usar en la conexión de JDBC. Se puede pasar como una string codificada en Base64 encriptada con una clave de Cloud KMS.
password: La contraseña para usar en la conexión de JDBC (opcional). Se puede pasar como una string codificada en Base64 encriptada con una clave de Cloud KMS.
query: La consulta que se ejecutará en la fuente para extraer los datos. Se debe especificar la consulta O ambas tablas Y PartitionColumn. (Ejemplo: selecciona * de sampledb.sample_table).
KMSEncryptionKey: La clave de encriptación de Cloud KMS para desencriptar el nombre de usuario, la contraseña y la cadena de conexión. Si se pasa la clave de Cloud KMS, el nombre de usuario, la contraseña y la string de conexión deben pasarse encriptados. (Ejemplo: projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key).
useColumnAlias: Si está habilitado (configurado como verdadero), la canalización considerará el alias de columna (“AS”) en lugar del nombre de la columna para asignar las filas a BigQuery. La configuración predeterminada es "false".
isTruncate: Si está habilitado (configurado como verdadero), la canalización se truncará antes de cargar los datos en BigQuery. La configuración predeterminada es "false", que se usa solo para agregar datos.
partitionColumn: Si se proporciona este parámetro (junto con table), en JdbcIO se lee la tabla en paralelo mediante la ejecución de varias instancias de la consulta en la misma tabla (subconsulta) mediante rangos. Por el momento, solo se admiten columnas de partición largas. Se debe especificar la consulta O ambas tablas Y PartitionColumn.
table: Tabla desde la que se leerá el uso de particiones. Se debe especificar la consulta O ambas tablas Y PartitionColumn. Este parámetro también acepta una subconsulta entre paréntesis (Por ejemplo: [selecciona ID, nombre de Persona] como subconsulta).
numPartitions: La cantidad de particiones. Esto, junto con el límite inferior y superior, forman las particiones para las expresiones de cláusula WHERE generadas que se usan a fin de dividir la columna de partición de manera uniforme. Cuando la entrada es menor que 1, se establece el número en 1.
lowerBound: Límite inferior usado en el esquema de partición. Si no se proporciona, Beam lo infiere de manera automática (para los tipos compatibles).
upperBound: Límite superior usado en el esquema de partición. Si no se proporciona, Beam lo infiere de manera automática (para los tipos compatibles).
fetchSize: la cantidad de filas que se recuperarán de la base de datos a la vez. No se usa para lecturas particionadas. La configuración predeterminada es 50,000.
createDisposition: CreateDisposition de BigQuery. Por ejemplo, CREATE_IF_NEEDED, CREATE_NEVER. La configuración predeterminada es: CREATE_NEVER.
bigQuerySchemaPath: La ruta de Cloud Storage para el esquema JSON de BigQuery. Si createDisposition se establece como CREATE_IF_NEEDED, se debe especificar este parámetro. (Ejemplo: gs://your-bucket/your-schema.json).
disabledAlgorithms: Algoritmos separados por comas que se deben inhabilitar Si este valor se establece como none, no se inhabilita ningún algoritmo. Ten cuidado porque los algoritmos inhabilitados de forma predeterminada tienen vulnerabilidades o problemas de rendimiento. (Ejemplo: SSLv3, RC4).
extraFilesToStage: Rutas de Cloud Storage separadas por comas o Secrets de Secret Manager para los archivos que se deben almacenar en etapa intermedia en el trabajador. Estos archivos se guardarán en el directorio /extra_files de cada trabajador (ejemplo: gs://your-bucket/file.txt,projects/project-id/secrets/secret-id/versions/version-id).
useStorageWriteApi: Si está habilitada (configurada como verdadera), la canalización usará la API de Storage Write cuando escriba los datos en BigQuery (consulta https://cloud.google.com/blog/products/data-analytics/streaming-data-into-bigquery-using-storage-write-api). La configuración predeterminada es "false".
useStorageWriteApiAtLeastOnce: Este parámetro solo se aplica si “Usar la API de BigQuery Storage Write” está habilitada. Si se habilita, se usará la semántica de “al menos una vez” para la API de Storage Write; de lo contrario, se usará la semántica de “exactamente una vez”. La configuración predeterminada es "false".

Ejecuta la plantilla

Consola

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 JDBC to BigQuery template.
En los campos de parámetros proporcionados, ingresa los valores de tus parámetros.
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/Jdbc_to_BigQuery_Flex \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --parameters \
       driverJars=DRIVER_JARS,\
       driverClassName=DRIVER_CLASS_NAME,\
       connectionURL=CONNECTION_URL,\
       outputTable=OUTPUT_TABLE,\
       bigQueryLoadingTemporaryDirectory=BIG_QUERY_LOADING_TEMPORARY_DIRECTORY,\

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/
Precaución: La versión más reciente de las plantillas podría actualizarse con cambios rotundos. Los entornos de producción deben usar plantillas que se conserven en la carpeta superior con la fecha más reciente para evitar que estos cambios rotundos afecten los flujos de trabajo de producción.
REGION_NAME: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
DRIVER_JARS: Son las rutas de Cloud Storage separadas por comas de los controladores JDBC.
DRIVER_CLASS_NAME: El nombre de la clase del controlador de JDBC.
CONNECTION_URL: La string de la URL de la conexión de JDBC.
OUTPUT_TABLE: Es la tabla de salida de BigQuery.
BIG_QUERY_LOADING_TEMPORARY_DIRECTORY: Directorio temporal para el proceso de carga de BigQuery.

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
{
   "launchParameter": {
     "jobName": "JOB_NAME",
     "parameters": {
       "driverJars": "DRIVER_JARS",
       "driverClassName": "DRIVER_CLASS_NAME",
       "connectionURL": "CONNECTION_URL",
       "outputTable": "OUTPUT_TABLE",
       "bigQueryLoadingTemporaryDirectory": "BIG_QUERY_LOADING_TEMPORARY_DIRECTORY",
     },
     "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Jdbc_to_BigQuery_Flex",
     "environment": { "maxWorkers": "10" }
  }
}

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/
Precaución: La versión más reciente de las plantillas podría actualizarse con cambios rotundos. Los entornos de producción deben usar plantillas que se conserven en la carpeta superior con la fecha más reciente para evitar que estos cambios rotundos afecten los flujos de trabajo de producción.
LOCATION: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
DRIVER_JARS: Son las rutas de Cloud Storage separadas por comas de los controladores JDBC.
DRIVER_CLASS_NAME: El nombre de la clase del controlador de JDBC.
CONNECTION_URL: La string de la URL de la conexión de JDBC.
OUTPUT_TABLE: Es la tabla de salida de BigQuery.
BIG_QUERY_LOADING_TEMPORARY_DIRECTORY: Directorio temporal para el proceso de carga de BigQuery.

Código fuente de la plantilla

Java

/*
 * Copyright (C) 2018 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.templates;

import static com.google.cloud.teleport.v2.utils.GCSUtils.getGcsFileAsString;
import static com.google.cloud.teleport.v2.utils.KMSUtils.maybeDecrypt;

import com.google.api.services.bigquery.model.TableRow;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.options.JdbcToBigQueryOptions;
import com.google.cloud.teleport.v2.utils.BigQueryIOUtils;
import com.google.cloud.teleport.v2.utils.JdbcConverters;
import com.google.common.annotations.VisibleForTesting;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.TableRowJsonCoder;
import org.apache.beam.sdk.io.jdbc.JdbcIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider;
import org.apache.beam.sdk.values.PCollection;

/**
 * A template that copies data from a relational database using JDBC to an existing BigQuery table.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/jdbc-to-googlecloud/README_Jdbc_to_BigQuery_Flex.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Jdbc_to_BigQuery_Flex",
    category = TemplateCategory.BATCH,
    displayName = "JDBC to BigQuery with BigQuery Storage API support",
    description = {
      "The JDBC to BigQuery template is a batch pipeline that copies data from a relational database table into an existing BigQuery table. "
          + "This pipeline uses JDBC to connect to the relational database. You can use this template to copy data from any relational database with available JDBC drivers into BigQuery.",
      "For an extra layer of protection, you can also pass in a Cloud KMS key along with a Base64-encoded username, password, and connection string parameters encrypted with the Cloud KMS key. "
          + "See the <a href=\"https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt\">Cloud KMS API encryption endpoint</a> for additional details on encrypting your username, password, and connection string parameters."
    },
    optionsClass = JdbcToBigQueryOptions.class,
    flexContainerName = "jdbc-to-bigquery",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/jdbc-to-bigquery",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The JDBC drivers for the relational database must be available.",
      "The BigQuery table must exist before pipeline execution.",
      "The BigQuery table must have a compatible schema.",
      "The relational database must be accessible from the subnet where Dataflow runs."
    })
public class JdbcToBigQuery {

  /**
   * Main entry point for executing the pipeline. This will run the pipeline asynchronously. If
   * blocking execution is required, use the {@link JdbcToBigQuery#run} method to start the pipeline
   * and invoke {@code result.waitUntilFinish()} on the {@link PipelineResult}.
   *
   * @param args The command-line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    // Parse the user options passed from the command-line
    JdbcToBigQueryOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(JdbcToBigQueryOptions.class);

    run(options, writeToBQTransform(options));
  }

  /**
   * Create the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @param writeToBQ the transform that outputs {@link TableRow}s to BigQuery.
   * @return The result of the pipeline execution.
   */
  @VisibleForTesting
  static PipelineResult run(JdbcToBigQueryOptions options, Write<TableRow> writeToBQ) {
    // Validate BQ STORAGE_WRITE_API options
    BigQueryIOUtils.validateBQStorageApiOptionsBatch(options);

    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    /*
     * Steps: 1) Read records via JDBC and convert to TableRow via RowMapper
     *        2) Append TableRow to BigQuery via BigQueryIO
     */
    JdbcIO.DataSourceConfiguration dataSourceConfiguration =
        JdbcIO.DataSourceConfiguration.create(
                StaticValueProvider.of(options.getDriverClassName()),
                maybeDecrypt(options.getConnectionURL(), options.getKMSEncryptionKey()))
            .withUsername(maybeDecrypt(options.getUsername(), options.getKMSEncryptionKey()))
            .withPassword(maybeDecrypt(options.getPassword(), options.getKMSEncryptionKey()));

    if (options.getDriverJars() != null) {
      dataSourceConfiguration = dataSourceConfiguration.withDriverJars(options.getDriverJars());
    }

    if (options.getConnectionProperties() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withConnectionProperties(options.getConnectionProperties());
    }

    /*
     * Step 1: Read records via JDBC and convert to TableRow
     *         via {@link org.apache.beam.sdk.io.jdbc.JdbcIO.RowMapper}
     */
    PCollection<TableRow> rows;
    if (options.getPartitionColumn() != null && options.getTable() != null) {
      // Read with Partitions
      // TODO(pranavbhandari): Support readWithPartitions for other data types.
      JdbcIO.ReadWithPartitions<TableRow, Long> readIO =
          JdbcIO.<TableRow>readWithPartitions()
              .withDataSourceConfiguration(dataSourceConfiguration)
              .withTable(options.getTable())
              .withPartitionColumn(options.getPartitionColumn())
              .withRowMapper(JdbcConverters.getResultSetToTableRow(options.getUseColumnAlias()));
      if (options.getNumPartitions() != null) {
        readIO = readIO.withNumPartitions(options.getNumPartitions());
      }
      if (options.getLowerBound() != null && options.getUpperBound() != null) {
        readIO =
            readIO.withLowerBound(options.getLowerBound()).withUpperBound(options.getUpperBound());
      }

      if (options.getFetchSize() != null && options.getFetchSize() > 0) {
        readIO = readIO.withFetchSize(options.getFetchSize());
      }

      rows = pipeline.apply("Read from JDBC with Partitions", readIO);
    } else {
      if (options.getQuery() == null) {
        throw new IllegalArgumentException(
            "Either 'query' or both 'table' AND 'PartitionColumn' must be specified to read from JDBC");
      }
      JdbcIO.Read<TableRow> readIO =
          JdbcIO.<TableRow>read()
              .withDataSourceConfiguration(dataSourceConfiguration)
              .withQuery(options.getQuery())
              .withCoder(TableRowJsonCoder.of())
              .withRowMapper(JdbcConverters.getResultSetToTableRow(options.getUseColumnAlias()));

      if (options.getFetchSize() != null && options.getFetchSize() > 0) {
        readIO = readIO.withFetchSize(options.getFetchSize());
      }

      rows = pipeline.apply("Read from JdbcIO", readIO);
    }

    /*
     * Step 2: Append TableRow to an existing BigQuery table
     */
    rows.apply("Write to BigQuery", writeToBQ);

    // Execute the pipeline and return the result.
    return pipeline.run();
  }

  /**
   * Create the {@link Write} transform that outputs the collection to BigQuery as per input option.
   */
  @VisibleForTesting
  static Write<TableRow> writeToBQTransform(JdbcToBigQueryOptions options) {
    // Needed for loading GCS filesystem before Pipeline.Create call
    FileSystems.setDefaultPipelineOptions(options);
    Write<TableRow> write =
        BigQueryIO.writeTableRows()
            .withoutValidation()
            .withCreateDisposition(Write.CreateDisposition.valueOf(options.getCreateDisposition()))
            .withWriteDisposition(
                options.getIsTruncate()
                    ? Write.WriteDisposition.WRITE_TRUNCATE
                    : Write.WriteDisposition.WRITE_APPEND)
            .withCustomGcsTempLocation(
                StaticValueProvider.of(options.getBigQueryLoadingTemporaryDirectory()))
            .to(options.getOutputTable());

    if (Write.CreateDisposition.valueOf(options.getCreateDisposition())
        != Write.CreateDisposition.CREATE_NEVER) {
      write = write.withJsonSchema(getGcsFileAsString(options.getBigQuerySchemaPath()));
    }

    return write;
  }
}

¿Qué sigue?

Obtén información sobre las plantillas de Dataflow.
Consulta la lista de plantillas que proporciona Google.