Modello Pub/Sub a Java Database Connectivity (JDBC)

Beta

Questa funzionalità è soggetta ai "Termini dell'offerta pre-GA" nella sezione dei Termini di servizio generali dei Termini specifici dei servizi. Le funzionalità pre-GA sono disponibili "così come sono" e potrebbero avere un supporto limitato. Per ulteriori informazioni, consulta le descrizioni della fase di lancio.

Il modello Da Pub/Sub a Java Database Connectivity (JDBC) è una pipeline in modalità flusso che importa i dati da una sottoscrizione Pub/Sub preesistente come stringhe JSON e scrive i record risultanti in JDBC.

Requisiti della pipeline

• La sottoscrizione Pub/Sub deve esistere prima dell'esecuzione della pipeline.
• L'origine JDBC deve esistere prima dell'esecuzione della pipeline.
• L'argomento messaggi non recapitabili di output di Pub/Sub deve esistere prima dell'esecuzione della pipeline.

Parametri del modello

Parametro	Descrizione
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 i caratteri di spaziatura dalla stringa con codifica in Base64.
driverJars	Percorsi Cloud Storage separati da virgole per i driver JDBC. Ad esempio, gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar.
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 in 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 in Base64.
connectionProperties	(Facoltativo) Stringa delle proprietà da utilizzare per la connessione JDBC. Il formato della stringa deve essere [propertyName=property;]*. Ad esempio, unicode=true;characterEncoding=UTF-8.
statement	Istruzione da eseguire sul database. L'istruzione deve specificare i nomi delle colonne della tabella in qualsiasi ordine. Solo i valori dei nomi delle colonne specificati vengono letti dal JSON e aggiunti all'istruzione. Ad esempio, INSERT INTO tableName (column1, column2) VALUES (?,?)
inputSubscription	La sottoscrizione Pub/Sub di input da cui leggere, nel formato projects/<project>/subscriptions/<subscription>.
outputDeadletterTopic	L'argomento Pub/Sub a cui inoltrare i messaggi non recapitabili. Ad esempio, projects/<project-id>/topics/<topic-name>.
KMSEncryptionKey	(Facoltativo) La chiave di crittografia Cloud KMS per decriptare il nome utente, la password e la stringa di connessione. Se viene passata la chiave Cloud KMS, il nome utente, la password e la stringa di connessione devono essere passati tutti in formato criptato.
extraFilesToStage	Percorsi Cloud Storage o secret di Secret Manager separati da virgole per i file da preparare nel worker. Questi file verranno salvati nella directory /extra_files di ogni worker. Ad esempio, gs://<my-bucket>/file.txt,projects/<project-id>/secrets/<secret-id>/versions/<version-id>.

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 job univoco.
3. (Facoltativo) Per Endpoint regionale, 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 il modello Da Pub/Sub a JDBC.
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 la versione 284.0.0 o successiva di Google Cloud CLI.

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/Pubsub_to_Jdbc \
    --region REGION_NAME \
    --parameters \
driverClassName=DRIVER_CLASS_NAME,\
connectionUrl=JDBC_CONNECTION_URL,\
driverJars=DRIVER_PATHS,\
username=CONNECTION_USERNAME,\
password=CONNECTION_PASSWORD,\
connectionProperties=CONNECTION_PROPERTIES,\
statement=SQL_STATEMENT,\
inputSubscription=INPUT_SUBSCRIPTION,\
outputDeadletterTopic=OUTPUT_DEADLETTER_TOPIC,\
KMSEncryptionKey=KMS_ENCRYPTION_KEY

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 principale non datata 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 datata nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale datata più recente per evitare che queste modifiche che causano interruzioni 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_CLASS_NAME: il nome della classe del driver
• JDBC_CONNECTION_URL: l'URL di connessione JDBC
• DRIVER_PATHS: il percorso o i percorsi Cloud Storage separati da virgole del o dei driver JDBC
• CONNECTION_USERNAME: il nome utente della connessione JDBC
• CONNECTION_PASSWORD: la password della connessione JDBC
• CONNECTION_PROPERTIES: le proprietà della connessione JDBC, se necessario
• SQL_STATEMENT: l'istruzione SQL da eseguire sul database
• INPUT_SUBSCRIPTION: la sottoscrizione Pub/Sub di input da cui leggere
• OUTPUT_DEADLETTER_TOPIC: l'argomento Pub/Sub a cui inoltrare i messaggi non recapitabili
• KMS_ENCRYPTION_KEY: la chiave di crittografia Cloud KMS

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta HTTP POST. 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
{
   "jobName": "JOB_NAME",
   "parameters": {
       "driverClassName": "DRIVER_CLASS_NAME",
       "connectionUrl": "JDBC_CONNECTION_URL",
       "driverJars": "DRIVER_PATHS",
       "username": "CONNECTION_USERNAME",
       "password": "CONNECTION_PASSWORD",
       "connectionProperties": "CONNECTION_PROPERTIES",
       "statement": "SQL_STATEMENT",
       "inputSubscription": "INPUT_SUBSCRIPTION",
       "outputDeadletterTopic": "OUTPUT_DEADLETTER_TOPIC",
       "KMSEncryptionKey":"KMS_ENCRYPTION_KEY"
   },
   "environment": { "zone": "us-central1-f" },
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto in cui vuoi eseguire il job Dataflow Google Cloud
• 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 principale non datata 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 datata nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale datata più recente per evitare che queste modifiche che causano interruzioni influiscano sui flussi di lavoro di produzione.
• LOCATION: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• DRIVER_CLASS_NAME: il nome della classe del driver
• JDBC_CONNECTION_URL: l'URL di connessione JDBC
• DRIVER_PATHS: il percorso o i percorsi Cloud Storage separati da virgole del o dei driver JDBC
• CONNECTION_USERNAME: il nome utente della connessione JDBC
• CONNECTION_PASSWORD: la password della connessione JDBC
• CONNECTION_PROPERTIES: le proprietà della connessione JDBC, se necessario
• SQL_STATEMENT: l'istruzione SQL da eseguire sul database
• INPUT_SUBSCRIPTION: la sottoscrizione Pub/Sub di input da cui leggere
• OUTPUT_DEADLETTER_TOPIC: l'argomento Pub/Sub a cui inoltrare i messaggi non recapitabili
• KMS_ENCRYPTION_KEY: la chiave di crittografia Cloud KMS

Codice sorgente del modello

Java

/*
 * Copyright (C) 2021 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.KMSUtils.maybeDecrypt;

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.io.DynamicJdbcIO;
import com.google.cloud.teleport.v2.options.PubsubToJdbcOptions;
import com.google.cloud.teleport.v2.transforms.ErrorConverters;
import com.google.cloud.teleport.v2.utils.JsonStringToQueryMapper;
import com.google.cloud.teleport.v2.values.FailsafeElement;
import com.google.common.base.Splitter;
import java.util.List;
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.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.io.jdbc.JdbcIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.values.PCollection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link PubsubToJdbc} streaming pipeline reads data from Google Cloud PubSub and publishes to
 * JDBC.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-googlecloud/README_Pubsub_to_Jdbc.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Pubsub_to_Jdbc",
    category = TemplateCategory.STREAMING,
    displayName = "Pub/Sub to JDBC",
    description =
        "The Pub/Sub to Java Database Connectivity (JDBC) template is a streaming pipeline that ingests data from a "
            + "pre-existing Cloud Pub/Sub subscription as JSON strings, and writes the resulting records to JDBC.",
    optionsClass = PubsubToJdbcOptions.class,
    flexContainerName = "jdbc-and-googlecloud",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/pubsub-to-jdbc",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The Cloud Pub/Sub subscription must exist prior to running the pipeline.",
      "The JDBC source must exist prior to running the pipeline.",
      "The Cloud Pub/Sub output deadletter topic must exist prior to running the pipeline.",
    },
    streaming = true,
    supportsAtLeastOnce = true)
public class PubsubToJdbc {

  /* Logger for class.*/
  private static final Logger LOG = LoggerFactory.getLogger(PubsubToJdbc.class);

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

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    PubsubToJdbcOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(PubsubToJdbcOptions.class);

    run(options);
  }

  /**
   * Runs a pipeline which reads message from Pub/Sub and writes to JDBC.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  public static PipelineResult run(PubsubToJdbcOptions options) {
    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    LOG.info("Starting Pubsub-to-Jdbc Pipeline.");

    /*
     * Steps:
     *  1) Read data from a Pub/Sub subscription
     *  2) Write to Jdbc Table
     *  3) Write errors to deadletter topic
     */
    PCollection<String> pubsubData =
        pipeline.apply(
            "readFromPubSubSubscription",
            PubsubIO.readStrings().fromSubscription(options.getInputSubscription()));

    JdbcIO.DataSourceConfiguration dataSourceConfiguration =
        JdbcIO.DataSourceConfiguration.create(
                ValueProvider.StaticValueProvider.of(options.getDriverClassName()),
                maybeDecrypt(options.getConnectionUrl(), options.getKMSEncryptionKey()))
            .withDriverJars(options.getDriverJars());
    if (options.getUsername() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withUsername(
              maybeDecrypt(options.getUsername(), options.getKMSEncryptionKey()).get());
    }
    if (options.getPassword() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withPassword(
              maybeDecrypt(options.getPassword(), options.getKMSEncryptionKey()).get());
    }
    if (options.getConnectionProperties() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withConnectionProperties(options.getConnectionProperties());
    }

    PCollection<FailsafeElement<String, String>> errors =
        pubsubData
            .apply(
                "writeToJdbc",
                DynamicJdbcIO.<String>write()
                    .withDataSourceConfiguration(dataSourceConfiguration)
                    .withStatement(options.getStatement())
                    .withPreparedStatementSetter(
                        new JsonStringToQueryMapper(getKeyOrder(options.getStatement()))))
            .setCoder(FAILSAFE_ELEMENT_CODER);

    errors.apply(
        "WriteFailedRecords",
        ErrorConverters.WriteStringMessageErrorsToPubSub.newBuilder()
            .setErrorRecordsTopic(options.getOutputDeadletterTopic())
            .build());

    return pipeline.run();
  }

  private static List<String> getKeyOrder(String statement) {
    int startIndex = statement.indexOf("(");
    int endIndex = statement.indexOf(")");
    String data = statement.substring(startIndex + 1, endIndex);
    return Splitter.on(',').trimResults().splitToList(data);
  }
}

Passaggi successivi

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