Modello PostgreSQL a BigQuery

Il modello PostgreSQL to BigQuery è una pipeline batch che copia i dati da una tabella PostgreSQL in una tabella BigQuery esistente. Questa pipeline utilizza JDBC per connettersi a PostgreSQL. Per un ulteriore livello di protezione, puoi anche trasmettere una chiave Cloud KMS insieme ai parametri di nome utente, password e stringa di connessione codificati in Base64 criptati con la chiave Cloud KMS. Per saperne di più sulla crittografia dei parametri di nome utente, password e stringa di connessione, consulta l'endpoint di crittografia dell'API Cloud KMS.

Requisiti della pipeline

• La tabella BigQuery deve esistere prima dell'esecuzione della pipeline.
• La tabella BigQuery deve avere uno schema compatibile.
• Il database relazionale deve essere accessibile dalla subnet in cui viene eseguito Dataflow.

Parametri del modello

Parametri obbligatori

• driverJars: l'elenco separato da virgole dei file JAR del driver. Ad esempio, gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar.
• driverClassName: il nome della classe del driver JDBC. Ad esempio, com.mysql.jdbc.Driver.
• connectionURL: la stringa dell'URL di connessione JDBC. Ad esempio, jdbc:mysql://some-host:3306/sampledb. Puoi trasmettere questo valore come stringa criptata con una chiave Cloud KMS e poi codificata in Base64. Rimuovi i caratteri di spaziatura dalla stringa con codifica Base64. Tieni presente la differenza tra una stringa di connessione al database Oracle non RAC (jdbc:oracle:thin:@some-host:<port>:<sid>) e una stringa di connessione al database Oracle RAC (jdbc:oracle:thin:@//some-host[:<port>]/<service_name>). Ad esempio, jdbc:mysql://some-host:3306/sampledb.
• outputTable: la posizione della tabella di output BigQuery. Ad esempio, <PROJECT_ID>:<DATASET_NAME>.<TABLE_NAME>.
• bigQueryLoadingTemporaryDirectory: la directory temporanea per il processo di caricamento di BigQuery. Ad esempio, gs://your-bucket/your-files/temp_dir.

Parametri facoltativi

• connectionProperties: la stringa delle proprietà da utilizzare per la connessione JDBC. Il formato della stringa deve essere [propertyName=property;]*.Per ulteriori informazioni, consulta Proprietà di configurazione (https://dev.mysql.com/doc/connector-j/en/connector-j-reference-configuration-properties.html) nella documentazione di MySQL. Ad esempio: unicode=true;characterEncoding=UTF-8.
• username: il nome utente da utilizzare per la connessione JDBC. Può essere passato come stringa criptata con una chiave Cloud KMS oppure può essere un secret di Secret Manager nel formato projects/{project}/secrets/{secret}/versions/{secret_version}.
• password: la password da utilizzare per la connessione JDBC. Può essere passato come stringa criptata con una chiave Cloud KMS oppure può essere un secret di Secret Manager nel formato projects/{project}/secrets/{secret}/versions/{secret_version}.
• query: la query da eseguire sull'origine per estrarre i dati. Tieni presente che alcuni tipi JDBC SQL e BigQuery, sebbene condividano lo stesso nome, presentano alcune differenze. Alcune mappature importanti dei tipi SQL -> BigQuery da tenere a mente sono DATETIME --> TIMESTAMP. Se gli schemi non corrispondono, potrebbe essere necessario il cast del tipo. Ad esempio: select * from sampledb.sample_table.
• KMSEncryptionKey: la chiave di crittografia Cloud KMS da utilizzare per decriptare il nome utente, la password e la stringa di connessione. Se passi una chiave Cloud KMS, devi anche criptare il nome utente, la password e la stringa di connessione. Ad esempio, projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
• useColumnAlias: se impostato su true, la pipeline utilizza l'alias della colonna (AS) anziché il nome della colonna per mappare le righe a BigQuery. Il valore predefinito è false.
• isTruncate: se impostato su true, la pipeline tronca i dati prima di caricarli in BigQuery. Il valore predefinito è false, che fa sì che la pipeline aggiunga i dati.
• partitionColumn: se partitionColumn viene specificato insieme a table, JdbcIO legge la tabella in parallelo eseguendo più istanze della query sulla stessa tabella (subquery) utilizzando gli intervalli. Al momento supporta le colonne di partizione Long e DateTime. Passa il tipo di colonna tramite partitionColumnType.
• partitionColumnType: il tipo di partitionColumn, accetta long o datetime. Il valore predefinito è long.
• table: la tabella da cui leggere quando si utilizzano le partizioni. Questo parametro accetta anche una sottoquery tra parentesi. Ad esempio, (select id, name from Person) as subq.
• numPartitions: il numero di partizioni. Con il limite inferiore e superiore, questo valore forma gli incrementi di partizione per le espressioni della clausola WHERE generate, che vengono utilizzate per dividere uniformemente la colonna di partizione. Quando l'input è inferiore a 1, il numero viene impostato su 1.
• lowerBound: il limite inferiore da utilizzare nello schema di partizionamento. Se non viene fornito, questo valore viene dedotto automaticamente da Apache Beam per i tipi supportati. datetime partitionColumnType accetta il limite inferiore nel formato yyyy-MM-dd HH:mm:ss.SSSZ. Ad esempio: 2024-02-20 07:55:45.000+03:30.
• upperBound: il limite superiore da utilizzare nello schema di partizionamento. Se non viene fornito, questo valore viene dedotto automaticamente da Apache Beam per i tipi supportati. datetime partitionColumnType accetta il limite superiore nel formato yyyy-MM-dd HH:mm:ss.SSSZ. Ad esempio: 2024-02-20 07:55:45.000+03:30.
• fetchSize: il numero di righe da recuperare dal database alla volta. Non utilizzata per le letture partizionate. Il valore predefinito è 50.000.
• createDisposition: il valore BigQuery CreateDisposition da utilizzare. Ad esempio, CREATE_IF_NEEDED o CREATE_NEVER. Il valore predefinito è CREATE_NEVER.
• bigQuerySchemaPath: il percorso Cloud Storage dello schema JSON BigQuery. Se createDisposition è impostato su CREATE_IF_NEEDED, questo parametro deve essere specificato. Ad esempio: gs://your-bucket/your-schema.json.
• outputDeadletterTable: la tabella BigQuery da utilizzare per i messaggi che non sono riusciti a raggiungere la tabella di output, formattata come "PROJECT_ID:DATASET_NAME.TABLE_NAME". Se la tabella non esiste, viene creata quando viene eseguita la pipeline. Se questo parametro non viene specificato, la pipeline non riuscirà a causa di errori di scrittura.Questo parametro può essere specificato solo se useStorageWriteApi o useStorageWriteApiAtLeastOnce è impostato su true.
• disabledAlgorithms: algoritmi separati da virgole da disattivare. Se questo valore è impostato su none, nessun algoritmo viene disattivato. Utilizza questo parametro con cautela, perché gli algoritmi disattivati per impostazione predefinita potrebbero presentare vulnerabilità o problemi di prestazioni. Ad esempio: SSLv3, RC4.
• extraFilesToStage: percorsi Cloud Storage o secret Secret Manager separati da virgole per i file da preparare nel worker. Questi file vengono salvati nella directory /extra_files di ogni worker. Ad esempio, gs://<BUCKET_NAME>/file.txt,projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<VERSION_ID>.
• useStorageWriteApi: se true, la pipeline utilizza l'API BigQuery Storage Write (https://cloud.google.com/bigquery/docs/write-api). Il valore predefinito è false. Per saperne di più, consulta la pagina Utilizzo dell'API Storage Write (https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api).
• useStorageWriteApiAtLeastOnce: quando utilizzi l'API Storage Write, specifica la semantica di scrittura. Per utilizzare la semantica almeno una volta (https://beam.apache.org/documentation/io/built-in/google-bigquery/#at-least-once-semantics), imposta questo parametro su true. Per utilizzare la semantica di esecuzione esatta, imposta il parametro su false. Questo parametro si applica solo quando useStorageWriteApi è true. Il valore predefinito è false.

Esegui il modello

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. La regione predefinita è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the PostgreSQL to BigQuery template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire modelli flex, devi avere la versione Google Cloud CLI 284.0.0 o successive.

Nella shell o nel terminale, esegui il modello:

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/PostgreSQL_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

Sostituisci quanto segue:

• JOB_NAME: un nome univoco del job a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare l'ultima versione del modello, disponibile nella cartella principale senza data nel bucket: gs://dataflow-templates-REGION_NAME/latest/
  • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella principale con data nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione:l'ultima versione dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale con data più recente per evitare che queste modifiche che causano interruzioni influiscano sui workflow di produzione.
• REGION_NAME: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• JDBC_CONNECTION_URL: l'URL di connessione JDBC
• SOURCE_SQL_QUERY: la query SQL da eseguire sul database di origine
• DATASET: il tuo set di dati BigQuery
• TABLE_NAME: il nome della tabella BigQuery
• PATH_TO_TEMP_DIR_ON_GCS: il percorso Cloud Storage della directory temporanea
• CONNECTION_PROPERTIES: le proprietà di connessione JDBC, se necessario
• CONNECTION_USERNAME: il nome utente della connessione JDBC
• CONNECTION_PASSWORD: la password di connessione JDBC
• KMS_ENCRYPTION_KEY: la chiave di crittografia Cloud KMS

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per saperne di più sull'API e sui relativi ambiti di autorizzazione, 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/PostgreSQL_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" }
  }
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto Google Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome univoco del job a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare l'ultima versione del modello, disponibile nella cartella principale senza data nel bucket: gs://dataflow-templates-REGION_NAME/latest/
  • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella principale con data nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione:l'ultima versione dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale con data più recente per evitare che queste modifiche che causano interruzioni influiscano sui workflow di produzione.
• LOCATION: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• JDBC_CONNECTION_URL: l'URL di connessione JDBC
• SOURCE_SQL_QUERY: la query SQL da eseguire sul database di origine
• DATASET: il tuo set di dati BigQuery
• TABLE_NAME: il nome della tabella BigQuery
• PATH_TO_TEMP_DIR_ON_GCS: il percorso Cloud Storage della directory temporanea
• CONNECTION_PROPERTIES: le proprietà di connessione JDBC, se necessario
• CONNECTION_USERNAME: il nome utente della connessione JDBC
• CONNECTION_PASSWORD: la password di connessione JDBC
• KMS_ENCRYPTION_KEY: la chiave di crittografia Cloud KMS

Codice sorgente del modello

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;
  }
}

Passaggi successivi

• Scopri di più sui modelli Dataflow.
• Consulta l'elenco dei modelli forniti da Google.