Plantilla de SQL Server a BigQuery

La plantilla de SQL Server a BigQuery es una canalización por lotes que copia datos de una tabla de SQL Server en una tabla de BigQuery. Esta canalización usa JDBC para conectarse a SQL Server. Para añadir una capa de protección adicional, también puedes incluir una clave de Cloud KMS junto con los parámetros de nombre de usuario, contraseña y cadena de conexión codificados en Base64 y encriptados con la clave de Cloud KMS. Para obtener más información sobre cómo cifrar los parámetros de nombre de usuario, contraseña y cadena de conexión, consulta el endpoint de cifrado de la API Cloud KMS.

Requisitos del flujo de procesamiento

• La tabla de BigQuery debe existir antes de ejecutar el flujo de procesamiento.
• La tabla de BigQuery debe tener un esquema compatible.
• Se debe poder acceder a la base de datos relacional desde la subred en la que se ejecuta Dataflow.

Parámetros de plantilla

Parámetros obligatorios

• driverJars lista separada por comas de archivos JAR del controlador. Por ejemplo, gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar.
• driverClassName el nombre de la clase del controlador JDBC. Por ejemplo, com.mysql.jdbc.Driver.
• connectionURL: la cadena de URL de conexión JDBC. Por ejemplo, jdbc:mysql://some-host:3306/sampledb. Puedes introducir este valor como una cadena cifrada con una clave de Cloud KMS y, a continuación, codificada en Base64. Quita los espacios de la cadena codificada en Base64. Ten en cuenta la diferencia entre una cadena de conexión de base de datos de Oracle no 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>). Por ejemplo, jdbc:mysql://some-host:3306/sampledb.
• outputTable la ubicación de la tabla de salida de BigQuery. Por ejemplo, <PROJECT_ID>:<DATASET_NAME>.<TABLE_NAME>.
• bigQueryLoadingTemporaryDirectory directorio temporal del proceso de carga de BigQuery. Por ejemplo, gs://your-bucket/your-files/temp_dir.

Parámetros opcionales

• connectionProperties la cadena de propiedades que se va a usar para la conexión JDBC. El formato de la cadena debe ser [propertyName=property;]*.Para obtener más información, consulta Configuration Properties (https://dev.mysql.com/doc/connector-j/en/connector-j-reference-configuration-properties.html) en la documentación de MySQL. Por ejemplo, unicode=true;characterEncoding=UTF-8.
• username: el nombre de usuario que se usará para la conexión JDBC. Se puede enviar como una cadena cifrada con una clave de Cloud KMS o como un secreto de Secret Manager con el formato projects/{project}/secrets/{secret}/versions/{secret_version}.
• password: la contraseña que se usará para la conexión JDBC. Se puede enviar como una cadena cifrada con una clave de Cloud KMS o como un secreto de Secret Manager con el formato projects/{project}/secrets/{secret}/versions/{secret_version}.
• Consulta: la consulta que se va a ejecutar en la fuente para extraer los datos. Ten en cuenta que algunos tipos de JDBC SQL y BigQuery, aunque tengan el mismo nombre, presentan algunas diferencias. Estas son algunas asignaciones importantes de tipos de SQL a BigQuery que debes tener en cuenta:DATETIME --> TIMESTAMP Puede que sea necesario convertir tipos si los esquemas no coinciden. Por ejemplo, select * from sampledb.sample_table.
• KMSEncryptionKey clave de encriptado de Cloud KMS que se va a usar para desencriptar el nombre de usuario, la contraseña y la cadena de conexión. Si proporcionas una clave de Cloud KMS, también debes cifrar el nombre de usuario, la contraseña y la cadena de conexión. Por ejemplo, projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
• useColumnAlias si se define como true, la canalización usa el alias de columna (AS) en lugar del nombre de columna para asignar las filas a BigQuery. El valor predeterminado es false.
• isTruncate si se define como true, la canalización se trunca antes de cargar los datos en BigQuery. El valor predeterminado es false, lo que hace que la canalización añada datos.
• partitionColumn si se especifica partitionColumn junto con table, JdbcIO lee la tabla en paralelo ejecutando varias instancias de la consulta en la misma tabla (subconsulta) mediante intervalos. Actualmente, admite columnas de partición Long y DateTime. Pasa el tipo de columna por partitionColumnType.
• partitionColumnType el tipo de partitionColumn. Puede ser long o datetime. El valor predeterminado es "long".
• table: la tabla de la que se leerá al usar particiones. Este parámetro también acepta una subconsulta entre paréntesis. Por ejemplo, (select id, name from Person) as subq.
• numPartitions el número de particiones. Junto con los límites inferior y superior, este valor forma pasos de partición para las expresiones de la cláusula WHERE generadas que se usan para dividir la columna de partición de forma uniforme. Si el valor introducido es inferior a 1, el número se establece en 1.
• lowerBound el límite inferior que se va a usar en el esquema de partición. Si no se proporciona, Apache Beam infiere automáticamente este valor para los tipos admitidos. datetime partitionColumnType acepta el límite inferior en el formato yyyy-MM-dd HH:mm:ss.SSSZ. Por ejemplo, 2024-02-20 07:55:45.000+03:30.
• upperBound límite superior que se va a usar en el esquema de partición. Si no se proporciona, Apache Beam infiere automáticamente este valor para los tipos admitidos. datetime partitionColumnType acepta el límite superior en el formato yyyy-MM-dd HH:mm:ss.SSSZ. Por ejemplo, 2024-02-20 07:55:45.000+03:30.
• fetchSize número de filas que se van a obtener de la base de datos a la vez. No se usa para lecturas particionadas. El valor predeterminado es 50.000.
• createDisposition el valor de CreateDisposition de BigQuery que se va a usar. Por ejemplo, CREATE_IF_NEEDED o CREATE_NEVER. El valor predeterminado es CREATE_NEVER.
• bigQuerySchemaPath la ruta de Cloud Storage del esquema JSON de BigQuery. Si createDisposition se define como CREATE_IF_NEEDED, este parámetro debe especificarse. Por ejemplo, gs://your-bucket/your-schema.json.
• outputDeadletterTable la tabla de BigQuery que se va a usar para los mensajes que no se hayan podido enviar a la tabla de salida, con el formato "PROJECT_ID:DATASET_NAME.TABLE_NAME". Si la tabla no existe, se crea cuando se ejecuta la canalización. Si no se especifica este parámetro, la canalización fallará si se producen errores de escritura.Este parámetro solo se puede especificar si useStorageWriteApi o useStorageWriteApiAtLeastOnce tienen el valor "true".
• disabledAlgorithms algoritmos separados por comas que se van a inhabilitar. Si se le asigna el valor none, no se inhabilita ningún algoritmo. Usa este parámetro con precaución, ya que los algoritmos inhabilitados de forma predeterminada pueden tener vulnerabilidades o problemas de rendimiento. Por ejemplo, SSLv3, RC4.
• extraFilesToStage rutas de Cloud Storage o secretos de Secret Manager separados por comas de los archivos que se van a almacenar en el trabajador. Estos archivos se guardan en el directorio /extra_files de cada trabajador. Por ejemplo, gs://<BUCKET_NAME>/file.txt,projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<VERSION_ID>.
• useStorageWriteApi si es true, la canalización usa la API Storage Write de BigQuery (https://cloud.google.com/bigquery/docs/write-api). El valor predeterminado es false. Para obtener más información, consulta el artículo sobre cómo usar la API Storage Write (https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api).
• useStorageWriteApiAtLeastOnce cuando se usa la API Storage Write, especifica la semántica de escritura. Para usar la semántica de al menos una vez (https://beam.apache.org/documentation/io/built-in/google-bigquery/#at-least-once-semantics), asigna a este parámetro el valor true. Para usar la semántica de entrega única, asigna el valor false al parámetro. Este parámetro solo se aplica cuando useStorageWriteApi es true. El valor predeterminado es false.

Ejecutar la plantilla

Consola

1. Ve a la página Crear tarea a partir de plantilla de Dataflow.
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 SQL Server 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

Nota: Para usar la CLI de Google Cloud y ejecutar plantillas de flex, debes tener la versión CLI de Google Cloud 284.0.0 o una posterior.

En tu shell o terminal, ejecuta la plantilla:

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/SQLServer_to_BigQuery \
    --parameters \
connectionURL=JDBC_CONNECTION_URL,\
query=SOURCE_SQL_QUERY,\
outputTable=PROJECT_ID:DATASET.TABLE_NAME,
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS,\
connectionProperties=CONNECTION_PROPERTIES,\
username=CONNECTION_USERNAME,\
password=CONNECTION_PASSWORD,\
KMSEncryptionKey=KMS_ENCRYPTION_KEY

Haz los cambios siguientes:

• 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/
  Precaución: La última versión de las plantillas puede actualizarse con cambios importantes. Tus entornos de producción deben usar plantillas que se encuentren en la carpeta principal con fecha más reciente para evitar que estos cambios afecten a tus flujos de trabajo de producción.
• REGION_NAME: la región en la que quieras desplegar tu trabajo de Dataflow. Por ejemplo, us-central1
• JDBC_CONNECTION_URL: la URL de conexión de JDBC
• SOURCE_SQL_QUERY: la consulta de SQL que se va a ejecutar en la base de datos de origen
• DATASET: tu conjunto de datos de BigQuery
• TABLE_NAME: nombre de la tabla de BigQuery
• PATH_TO_TEMP_DIR_ON_GCS: la ruta de Cloud Storage al directorio temporal
• CONNECTION_PROPERTIES: las propiedades de conexión JDBC, si es necesario
• CONNECTION_USERNAME: nombre de usuario de la conexión JDBC
• CONNECTION_PASSWORD: la contraseña de la conexión JDBC
• KMS_ENCRYPTION_KEY: la clave de cifrado de Cloud KMS

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
{
  "launchParameter": {
    "jobName": "JOB_NAME",
    "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/SQLServer_to_BigQuery",
    "parameters": {
      "connectionURL": "JDBC_CONNECTION_URL",
      "query": "SOURCE_SQL_QUERY",
      "outputTable": "PROJECT_ID:DATASET.TABLE_NAME",
      "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS",
      "connectionProperties": "CONNECTION_PROPERTIES",
      "username": "CONNECTION_USERNAME",
      "password": "CONNECTION_PASSWORD",
      "KMSEncryptionKey":"KMS_ENCRYPTION_KEY"
    },
    "environment": { "zone": "us-central1-f" }
  }
}

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/
  Precaución: La última versión de las plantillas puede actualizarse con cambios importantes. Tus entornos de producción deben usar plantillas que se encuentren en la carpeta principal con fecha más reciente para evitar que estos cambios afecten a tus flujos de trabajo de producción.
• LOCATION: la región en la que quieras desplegar tu trabajo de Dataflow. Por ejemplo, us-central1
• JDBC_CONNECTION_URL: la URL de conexión de JDBC
• SOURCE_SQL_QUERY: la consulta de SQL que se va a ejecutar en la base de datos de origen
• DATASET: tu conjunto de datos de BigQuery
• TABLE_NAME: nombre de la tabla de BigQuery
• PATH_TO_TEMP_DIR_ON_GCS: la ruta de Cloud Storage al directorio temporal
• CONNECTION_PROPERTIES: las propiedades de conexión JDBC, si es necesario
• CONNECTION_USERNAME: nombre de usuario de la conexión JDBC
• CONNECTION_PASSWORD: la contraseña de la conexión JDBC
• KMS_ENCRYPTION_KEY: la clave de cifrado de Cloud KMS

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.transforms.BigQueryConverters.wrapBigQueryInsertError;
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.ErrorProto;
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.coders.FailsafeElementCoder;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.options.JdbcToBigQueryOptions;
import com.google.cloud.teleport.v2.transforms.ErrorConverters;
import com.google.cloud.teleport.v2.utils.BigQueryIOUtils;
import com.google.cloud.teleport.v2.utils.GCSAwareValueProvider;
import com.google.cloud.teleport.v2.utils.JdbcConverters;
import com.google.cloud.teleport.v2.utils.ResourceUtils;
import com.google.cloud.teleport.v2.utils.SecretManagerUtils;
import com.google.cloud.teleport.v2.values.FailsafeElement;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.StringUtf8Coder;
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.BigQueryInsertError;
import org.apache.beam.sdk.io.gcp.bigquery.InsertRetryPolicy;
import org.apache.beam.sdk.io.gcp.bigquery.TableRowJsonCoder;
import org.apache.beam.sdk.io.gcp.bigquery.WriteResult;
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.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TypeDescriptor;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;

/**
 * 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-and-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.",
      "If BigQuery table already exist before pipeline execution, it must have a compatible schema.",
      "The relational database must be accessible from the subnet where Dataflow runs."
    })
public class JdbcToBigQuery {

  /** Coder for FailsafeElement. */
  private static final FailsafeElementCoder<String, String> FAILSAFE_ELEMENT_CODER =
      FailsafeElementCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of());

  /**
   * 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);
    if (!options.getUseStorageWriteApi()
        && !options.getUseStorageWriteApiAtLeastOnce()
        && !Strings.isNullOrEmpty(options.getOutputDeadletterTable())) {
      throw new IllegalArgumentException(
          "outputDeadletterTable can only be specified if BigQuery Storage Write API is enabled either with useStorageWriteApi or useStorageWriteApiAtLeastOnce.");
    }

    // 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(
                    maybeParseSecret(options.getConnectionURL()), options.getKMSEncryptionKey()))
            .withUsername(
                maybeDecrypt(
                    maybeParseSecret(options.getUsername()), options.getKMSEncryptionKey()))
            .withPassword(
                maybeDecrypt(
                    maybeParseSecret(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
      JdbcIO.ReadWithPartitions<TableRow, ?> readIO = null;
      final String partitionColumnType = options.getPartitionColumnType();
      if (partitionColumnType == null || "long".equals(partitionColumnType)) {
        JdbcIO.ReadWithPartitions<TableRow, Long> longTypeReadIO =
            JdbcIO.<TableRow, Long>readWithPartitions(TypeDescriptors.longs())
                .withDataSourceConfiguration(dataSourceConfiguration)
                .withTable(options.getTable())
                .withPartitionColumn(options.getPartitionColumn())
                .withRowMapper(JdbcConverters.getResultSetToTableRow(options.getUseColumnAlias()));
        if (options.getLowerBound() != null && options.getUpperBound() != null) {
          // Check if lower bound and upper bound are long type.
          try {
            longTypeReadIO =
                longTypeReadIO
                    .withLowerBound(Long.valueOf(options.getLowerBound()))
                    .withUpperBound(Long.valueOf(options.getUpperBound()));
          } catch (NumberFormatException e) {
            throw new NumberFormatException(
                "Expected Long values for lowerBound and upperBound, received : " + e.getMessage());
          }
        }
        readIO = longTypeReadIO;
      } else if ("datetime".equals(partitionColumnType)) {
        JdbcIO.ReadWithPartitions<TableRow, DateTime> dateTimeReadIO =
            JdbcIO.<TableRow, DateTime>readWithPartitions(TypeDescriptor.of(DateTime.class))
                .withDataSourceConfiguration(dataSourceConfiguration)
                .withTable(options.getTable())
                .withPartitionColumn(options.getPartitionColumn())
                .withRowMapper(JdbcConverters.getResultSetToTableRow(options.getUseColumnAlias()));
        if (options.getLowerBound() != null && options.getUpperBound() != null) {
          DateTimeFormatter dateFormatter =
              DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSSZ").withOffsetParsed();
          // Check if lowerBound and upperBound are DateTime type.
          try {
            dateTimeReadIO =
                dateTimeReadIO
                    .withLowerBound(dateFormatter.parseDateTime(options.getLowerBound()))
                    .withUpperBound(dateFormatter.parseDateTime(options.getUpperBound()));
          } catch (IllegalArgumentException e) {
            throw new IllegalArgumentException(
                "Expected DateTime values in the format for lowerBound and upperBound, received : "
                    + e.getMessage());
          }
        }
        readIO = dateTimeReadIO;
      } else {
        throw new IllegalStateException("Received unsupported partitionColumnType.");
      }
      if (options.getNumPartitions() != null) {
        readIO = readIO.withNumPartitions(options.getNumPartitions());
      }
      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(new GCSAwareValueProvider(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
     */
    WriteResult writeResult = rows.apply("Write to BigQuery", writeToBQ);

    /*
     * Step 3.
     * If using Storage Write API, capture failed inserts and either
     *   a) write error rows to DLQ
     *   b) fail the pipeline
     */
    if (options.getUseStorageWriteApi() || options.getUseStorageWriteApiAtLeastOnce()) {
      PCollection<BigQueryInsertError> insertErrors =
          BigQueryIOUtils.writeResultToBigQueryInsertErrors(writeResult, options);

      if (!Strings.isNullOrEmpty(options.getOutputDeadletterTable())) {
        /*
         * Step 3a.
         * Elements that failed inserts into BigQuery are extracted and converted to FailsafeElement
         */
        PCollection<FailsafeElement<String, String>> failedInserts =
            insertErrors
                .apply(
                    "WrapInsertionErrors",
                    MapElements.into(FAILSAFE_ELEMENT_CODER.getEncodedTypeDescriptor())
                        .via((BigQueryInsertError e) -> wrapBigQueryInsertError(e)))
                .setCoder(FAILSAFE_ELEMENT_CODER);

        /*
         * Step 3a Contd.
         * Insert records that failed insert into deadletter table
         */
        failedInserts.apply(
            "WriteFailedRecords",
            ErrorConverters.WriteStringMessageErrors.newBuilder()
                .setErrorRecordsTable(options.getOutputDeadletterTable())
                .setErrorRecordsTableSchema(ResourceUtils.getDeadletterTableSchemaJson())
                .setUseWindowedTimestamp(false)
                .build());
      } else {
        /*
         * Step 3b.
         * Fail pipeline upon write errors if no DLQ was specified
         */
        insertErrors.apply(ParDo.of(new ThrowWriteErrorsDoFn()));
      }
    }

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

  static class ThrowWriteErrorsDoFn extends DoFn<BigQueryInsertError, Void> {
    @ProcessElement
    public void processElement(ProcessContext c) {
      BigQueryInsertError insertError = Objects.requireNonNull(c.element());
      List<String> errorMessages =
          insertError.getError().getErrors().stream()
              .map(ErrorProto::getMessage)
              .collect(Collectors.toList());
      String stackTrace = String.join("\nCaused by:", errorMessages);

      throw new IllegalStateException(
          String.format(
              "Failed to insert row %s.\nCaused by: %s", insertError.getRow(), stackTrace));
    }
  }

  /**
   * 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()))
            .withExtendedErrorInfo()
            .to(options.getOutputTable());

    if (Write.CreateDisposition.valueOf(options.getCreateDisposition())
        != Write.CreateDisposition.CREATE_NEVER) {
      write = write.withJsonSchema(getGcsFileAsString(options.getBigQuerySchemaPath()));
    }
    if (options.getUseStorageWriteApi() || options.getUseStorageWriteApiAtLeastOnce()) {
      write = write.withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors());
    }

    return write;
  }

  /**
   * Retrieves a secret value from SecretManagerUtils if the input string matches the specified
   * pattern.
   *
   * @param secret The input string representing a potential secret.
   * @return The secret value if the input matches the pattern and the secret is found, otherwise
   *     the original input string.
   */
  private static String maybeParseSecret(String secret) {
    // Check if the input string is not null.
    if (secret != null) {
      // Check if the input string matches the pattern for secrets stored in SecretManagerUtils.
      if (secret.matches("projects/.*/secrets/.*/versions/.*")) { // Use .* to match any characters
        // Retrieve the secret value from SecretManagerUtils.
        return SecretManagerUtils.getSecret(secret);
      }
    }
    // If the input is null or doesn't match the pattern, return the original input.
    return secret;
  }
}

Siguientes pasos

• Consulta información sobre las plantillas de Dataflow.
• Consulta la lista de plantillas proporcionadas por Google.