Da Java Database Connectivity (JDBC) a modello BigQuery

Il modello JDBC to BigQuery è una pipeline batch che copia i dati da una tabella di database relazionale a una tabella BigQuery esistente. Questa pipeline utilizza JDBC per la connessione al database relazionale. Utilizza questo modello per copiare in BigQuery i dati da qualsiasi database relazionale con driver JDBC disponibili.

Per un ulteriore livello di protezione, puoi passare una chiave Cloud KMS, insieme a un nome utente, una password e parametri della stringa di connessione codificati in Base64 e criptati con la chiave Cloud KMS. Per ulteriori dettagli sulla crittografia di nome utente, password e parametri della stringa di connessione, consulta l'endpoint di crittografia dell'API Cloud KMS.

Requisiti della pipeline

• I driver JDBC per il database relazionale devono essere disponibili.
• 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

Parametro	Descrizione
driverJars	L'elenco separato da virgole di 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 passare questo valore come stringa criptata con una chiave Cloud KMS e poi codificata in Base64. Rimuovi gli spazi vuoti dalla stringa codificata in Base64. Nota 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 BigQuery in cui scrivere l'output. Il nome deve utilizzare il formato <project>:<dataset>.<table_name>. Lo schema della tabella deve corrispondere agli oggetti di input. Ad esempio: <my-project>:<my-dataset>.<my-table>.
bigQueryLoadingTemporaryDirectory	La directory temporanea per il processo di caricamento di BigQuery. Ad esempio: gs://your-bucket/your-files/temp_dir.
connectionProperties	Facoltativo: la stringa delle proprietà da utilizzare per la connessione JDBC. Utilizza il formato della stringa [propertyName=property;]*. Ad esempio: unicode=true;characterEncoding=UTF-8.
username	Facoltativo: il nome utente da utilizzare per la connessione JDBC. Puoi passare questo valore criptato da una chiave Cloud KMS come stringa con codifica Base64.
password	Facoltativo: la password da utilizzare per la connessione JDBC. Puoi passare questo valore criptato da una chiave Cloud KMS come stringa con codifica Base64.
query	(Facoltativo) La query da eseguire sull'origine per estrarre i dati. Ad esempio: select * from sampledb.sample_table.
KMSEncryptionKey	(Facoltativo) La chiave di crittografia di Cloud KMS da utilizzare per decriptare il nome utente, la password e la stringa di connessione. Se passi in una chiave Cloud KMS, il nome utente, la password e la stringa di connessione devono essere tutti criptati. Ad esempio: projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
useColumnAlias	Facoltativo: se abilitato (impostato su true), la pipeline utilizza l'alias di colonna ("AS") anziché il nome della colonna per mappare le righe a BigQuery. Il valore predefinito è false.
isTruncate	(Facoltativo) Se l'opzione è abilitata (impostata su true), la pipeline viene troncata prima di caricare i dati in BigQuery. Il valore predefinito è false, che causa l'aggiunta di dati da parte della pipeline.
partitionColumn	(Facoltativo) Se questo parametro viene fornito insieme a table, JdbcIO legge la tabella in parallelo eseguendo più istanze della query nella stessa tabella (sottoquery) utilizzando gli intervalli. Attualmente supporta solo Long colonne di partizione.
table	(Facoltativo) La tabella da cui leggere quando utilizzi le partizioni. Questo parametro accetta anche una sottoquery tra parentesi. Ad esempio: (select id, name from Person come subq).
numPartitions	(Facoltativo) Il numero di partizioni. Con i limiti inferiore e superiore, questo valore forma gli incrementi di partizione per le espressioni della clausola WHERE generate utilizzate per suddividere in modo uniforme la colonna di partizione. Quando l'input è inferiore a 1, il numero viene impostato su 1.
lowerBound	Facoltativo: il limite inferiore da utilizzare nello schema di partizione. Se non viene specificato, questo valore viene dedotto automaticamente da Apache Beam per i tipi supportati.
upperBound	Facoltativo: il limite superiore da utilizzare nello schema di partizione. Se non viene specificato, questo valore viene dedotto automaticamente da Apache Beam per i tipi supportati.
disabledAlgorithms	(Facoltativo) Algoritmi separati da virgole da disattivare. Se questo valore viene impostato su none, nessun algoritmo viene disattivato. Da usare con cautela, perché gli algoritmi disattivati per impostazione predefinita sono noti per avere vulnerabilità o problemi di prestazioni. Ad esempio: SSLv3, RC4.
extraFilesToStage	(Facoltativo) Percorsi Cloud Storage o secret di Secret Manager separati da virgole per i file da inserire nel worker. Questi file vengono salvati nella directory /extra_files di ciascun worker. Ad esempio: gs://your-bucket/file.txt,projects/project-id/secrets/secret-id/versions/version-id.
useStorageWriteApi	Facoltativo: se true, la pipeline utilizza l' API BigQuery Storage Write. Il valore predefinito è false. Per ulteriori informazioni, consulta la pagina relativa all' utilizzo dell'API Storage Write.
useStorageWriteApiAtLeastOnce	Facoltativo: quando si utilizza l'API Storage Write, specifica la semantica di scrittura. Per utilizzare la semantica at-least-once, imposta questo parametro su true. Per usare la semantica "exactly-once", imposta il parametro su false. Questo parametro si applica solo quando useStorageWriteApi è true. Il valore predefinito è false.

Esegui il modello

Nota: se cerchi istruzioni per eseguire il modello da JDBC a BigQuery Legacy (Classic), consulta la documentazione relativa ai modelli da JDBC a BigQuery (legacy).

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, vedi Località Dataflow.

4. Nel menu a discesa Modello Dataflow, seleziona the JDBC 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 flex i modelli, devi avere Google Cloud CLI versione 284.0.0 o successiva.

Nella shell o nel terminale, esegui il modello:

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,\

Sostituisci quanto segue:

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

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella padre senza data del bucket: gs://dataflow-templates-REGION_NAME/latest/
  • il nome della versione, come 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella padre con data all'interno del bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale data più recente per evitare che queste modifiche che provocano errori influiscano sui flussi di lavoro di produzione.
• REGION_NAME: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• DRIVER_JARS: percorso o percorsi Cloud Storage dei driver JDBC separati da virgole
• DRIVER_CLASS_NAME: nome della classe del driver JDBC
• CONNECTION_URL: la stringa dell'URL di connessione JDBC.
• OUTPUT_TABLE: la tabella di output di BigQuery
• BIG_QUERY_LOADING_TEMPORARY_DIRECTORY: la directory temporanea per il processo di caricamento di BigQuery

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni 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",
     "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" }
  }
}

Sostituisci quanto segue:

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

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella padre senza data del bucket: gs://dataflow-templates-REGION_NAME/latest/
  • il nome della versione, come 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella padre con data all'interno del bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale data più recente per evitare che queste modifiche che provocano errori influiscano sui flussi di lavoro di produzione.
• LOCATION: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• DRIVER_JARS: percorso o percorsi Cloud Storage dei driver JDBC separati da virgole
• DRIVER_CLASS_NAME: nome della classe del driver JDBC
• CONNECTION_URL: la stringa dell'URL di connessione JDBC.
• OUTPUT_TABLE: la tabella di output di BigQuery
• BIG_QUERY_LOADING_TEMPORARY_DIRECTORY: la directory temporanea per il processo di caricamento di BigQuery

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

Passaggi successivi

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