Modello SourceDB to Spanner

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 SourceDB to Spanner è una pipeline batch che copia i dati da un database relazionale in un database Spanner esistente. Questa pipeline utilizza JDBC per connettersi al database relazionale. Puoi utilizzare questo modello per copiare i dati da qualsiasi database relazionale con driver JDBC disponibili in Spanner. Supporta solo un set limitato di tipi di MySQL

Per un ulteriore livello di protezione, puoi anche passare una chiave Cloud KMS insieme a parametri di nome utente, password e stringa di connessione codificati in Base64 e criptati con la chiave Cloud KMS. Per ulteriori dettagli sulla crittografia dei parametri di nome utente, password e 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.
• Le tabelle Spanner devono esistere prima dell'esecuzione della pipeline.
• Le tabelle Spanner devono avere uno schema compatibile.
• Il database relazionale deve essere accessibile dalla subnet in cui viene eseguito Dataflow.

Parametri del modello

Parametro	Descrizione
sourceConfigURL	La stringa dell'URL di connessione JDBC. Ad esempio, jdbc:mysql://127.4.5.30:3306/my-db?autoReconnect=true&maxReconnects=10&unicode=true&characterEncoding=UTF-8 o la configurazione dello shard.
instanceId	L'istanza Cloud Spanner di destinazione.
databaseId	Il database Cloud Spanner di destinazione.
projectId	Questo è il nome del progetto Cloud Spanner.
outputDirectory	Questa directory viene utilizzata per scaricare i record non riusciti/saltati/filtrati in una migrazione.
jdbcDriverJars	(Facoltativo) L'elenco separato da virgole dei file JAR del driver. Ad esempio: gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar. Il valore predefinito è vuoto.
jdbcDriverClassName	(Facoltativo) Il nome della classe del driver JDBC. Ad esempio: com.mysql.jdbc.Driver. Il valore predefinito è: com.mysql.jdbc.Driver.
username	(Facoltativo) Il nome utente da utilizzare per la connessione JDBC. Il valore predefinito è vuoto.
password	(Facoltativo) La password da utilizzare per la connessione JDBC. Il valore predefinito è vuoto.
tables	(Facoltativo) Le tabelle di cui eseguire la migrazione dall'origine. Il valore predefinito è vuoto.
numPartitions	(Facoltativo) Il numero di partizioni. Questo, insieme al limite inferiore e superiore, forma gli intervalli di partizioni per le espressioni della clausola WHERE generate utilizzate per dividere uniformemente la colonna di partizione. Se l'input è inferiore a 1, il numero viene impostato su 1. Il valore predefinito è: 0.
spannerHost	(Facoltativo) L'endpoint Cloud Spanner da chiamare nel modello. Ad esempio: https://batch-spanner.googleapis.com. Il valore predefinito è: https://batch-spanner.googleapis.com.
maxConnections	(Facoltativo) Configura il pool di connessioni JDBC su ogni worker con il numero massimo di connessioni. Utilizza un numero negativo per non avere limiti. Ad esempio: -1. Il valore predefinito è: 0.
sessionFilePath	(Facoltativo) Il percorso della sessione in Cloud Storage che contiene le informazioni di mapping di Spanner Migration Tool. Il valore predefinito è vuoto.
transformationJarPath	(Facoltativo) La posizione del file JAR personalizzato in Cloud Storage che contiene la logica di trasformazione personalizzata per l'elaborazione dei record. Il valore predefinito è vuoto.
transformationClassName	(Facoltativo) Il nome della classe completo con la logica di trasformazione personalizzata. È un campo obbligatorio se viene specificato transformationJarPath. Il valore predefinito è vuoto.
transformationCustomParameters	(Facoltativo) La stringa contenente eventuali parametri personalizzati da passare alla classe di trasformazione personalizzata. Il valore predefinito è vuoto.
disabledAlgorithms	(Facoltativo) Gli algoritmi separati da virgole da disattivare. Se questo valore è impostato su none, non viene disattivato alcun algoritmo. Utilizza questo parametro con cautela, perché gli algoritmi disattivati per impostazione predefinita potrebbero presentare vulnerabilità o problemi di prestazioni. Ad esempio: SSLv3, RC4.
extraFilesToStage	(Facoltativo) I percorsi di Cloud Storage o i secret di 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>/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 Sourcedb to Spanner.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud CLI

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/Sourcedb_to_Spanner_Flex \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --parameters \
       sourceConfigURL=SOURCE_CONFIG_URL,\
       instanceId=INSTANCE_ID,\
       databaseId=DATABASE_ID,\
       projectId=PROJECT_ID,\
       outputDirectory=OUTPUT_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 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
• SOURCE_CONFIG_URL: l'URL per connetterti all'host del database di origine. Può essere uno dei seguenti: L'URL di connessione JDBC, che deve contenere l'host, la porta e il nome del database di origine e può contenere facoltativamente proprietà come autoReconnect, maxReconnects e così via. Formato: `jdbc:mysql://{host}:{port}/{dbName}?{parameters}`2. Il percorso di configurazione dello shard
• INSTANCE_ID: l'ID dell'istanza Cloud Spanner.
• DATABASE_ID: l'ID del database Cloud Spanner.
• PROJECT_ID: l'ID del progetto Cloud Spanner.
• OUTPUT_DIRECTORY: la directory di output per gli eventi non riusciti/saltati/filtrati

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta HTTP POST. Per ulteriori 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": {
       "sourceConfigURL": "SOURCE_CONFIG_URL",
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "projectId": "PROJECT_ID",
       "outputDirectory": "OUTPUT_DIRECTORY",
     },
     "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Sourcedb_to_Spanner_Flex",
     "environment": { "maxWorkers": "10" }
  }
}

Sostituisci quanto segue:

• PROJECT_ID: l' Google Cloud ID progetto 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 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
• SOURCE_CONFIG_URL: l'URL per connetterti all'host del database di origine. Può essere uno dei seguenti: L'URL di connessione JDBC, che deve contenere l'host, la porta e il nome del database di origine e può contenere facoltativamente proprietà come autoReconnect, maxReconnects e così via. Formato: `jdbc:mysql://{host}:{port}/{dbName}?{parameters}`2. Il percorso di configurazione dello shard
• INSTANCE_ID: l'ID dell'istanza Cloud Spanner.
• DATABASE_ID: l'ID del database Cloud Spanner.
• PROJECT_ID: l'ID del progetto Cloud Spanner.
• OUTPUT_DIRECTORY: la directory di output per gli eventi non riusciti/saltati/filtrati

Codice sorgente del modello

Java

/*
 * Copyright (C) 2024 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 com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.CommonTemplateJvmInitializer;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.options.SourceDbToSpannerOptions;
import com.google.cloud.teleport.v2.source.ISrcToSpSourceConnector;
import com.google.cloud.teleport.v2.source.SourceConnectorFactory;
import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard;
import com.google.cloud.teleport.v2.spanner.migrations.source.config.JdbcShardConfig;
import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConnectionConfig;
import com.google.cloud.teleport.v2.spanner.migrations.utils.DataflowWorkerMachineTypeUtils;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.util.Optional;
import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.commons.lang3.StringUtils;

/**
 * A template that copies data from a relational database using JDBC to an existing Spanner
 * database.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/sourcedb-to-spanner/README_Sourcedb_to_Spanner_Flex.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Sourcedb_to_Spanner_Flex",
    category = TemplateCategory.BATCH,
    displayName = "Sourcedb to Spanner",
    description = {
      "The SourceDB to Spanner template is a batch pipeline that copies data from a relational"
          + " database into an existing Spanner database. 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 Spanner. This currently only supports a limited set of types of MySQL",
    },
    optionsClass = SourceDbToSpannerOptions.class,
    flexContainerName = "source-db-to-spanner",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/sourcedb-to-spanner",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The JDBC drivers for the relational database must be available.",
      "The Spanner tables must exist before pipeline execution.",
      "The Spanner tables must have a compatible schema.",
      "The relational database must be accessible from the subnet where Dataflow runs."
    })
public class SourceDbToSpanner {

  /**
   * Main entry point for executing the pipeline. This will run the pipeline asynchronously. If
   * blocking execution is required, use the {@link SourceDbToSpanner#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();

    SourceDbToSpannerOptions options = getSourceDbToSpannerOptions(args);
    run(options);
  }

  @VisibleForTesting
  protected static SourceDbToSpannerOptions getSourceDbToSpannerOptions(String[] args) {
    // Parse the user options passed from the command-line
    SourceDbToSpannerOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(SourceDbToSpannerOptions.class);
    // Stage SSL certificates to extraFiles if required as per the pipeline options.
    // Ref https://cloud.google.com/dataflow/docs/guides/templates/ssl-certificates
    new CommonTemplateJvmInitializer().beforeProcessing(options);
    return options;
  }

  /**
   * Validates the provided pipeline options. TODO: move this to source connector.
   *
   * @param options The execution parameters to the pipeline.
   * @param sourceConnectionConfig Parsed source connection config.
   * @throws IllegalArgumentException if the provided options are invalid for the pipeline.
   */
  @VisibleForTesting
  static void validateOptions(
      SourceDbToSpannerOptions options, SourceConnectionConfig sourceConnectionConfig) {
    if (SourceDbToSpannerOptions.PG_SOURCE_DIALECT.equals(options.getSourceDbDialect())) {
      Preconditions.checkArgument(
          (sourceConnectionConfig instanceof JdbcShardConfig),
          "Postgresql dialect should have JDBC source config.");
      for (Shard shard : ((JdbcShardConfig) sourceConnectionConfig).getShardConfigs()) {
        if (StringUtils.isNotBlank(shard.getNamespace())
            && !shard.getNamespace().equals("public")) {
          throw new IllegalArgumentException(
              "Non-public namespaces are currently unsupported for PostgreSQL migrations.");
        }
      }
    }
  }

  /**
   * Create the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  @VisibleForTesting
  static PipelineResult run(SourceDbToSpannerOptions options) {
    Pipeline pipeline = Pipeline.create(options);
    String workerMachineType =
        pipeline.getOptions().as(DataflowPipelineWorkerPoolOptions.class).getWorkerMachineType();
    Optional<Integer> resourceHintsMinCpus =
        DataflowWorkerMachineTypeUtils.getMinCpuResourceHint(pipeline.getOptions());
    DataflowWorkerMachineTypeUtils.validateMachineSpecs(workerMachineType, 4, resourceHintsMinCpus);

    SpannerConfig spannerConfig = createSpannerConfig(options);
    SourceConnectionConfig sourceConnectionConfig =
        PipelineController.getSourceConnectionConfig(
            options.getSourceDbDialect(), options.getSourceConfigURL());

    validateOptions(options, sourceConnectionConfig);

    // Decide type and source of migration
    ISrcToSpSourceConnector connector = SourceConnectorFactory.getSourceConnectorByDialect(options);
    return connector.executeMigration(options, sourceConnectionConfig, pipeline, spannerConfig);
  }

  @VisibleForTesting
  static SpannerConfig createSpannerConfig(SourceDbToSpannerOptions options) {
    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withProjectId(ValueProvider.StaticValueProvider.of(options.getProjectId()))
            .withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()))
            .withInstanceId(ValueProvider.StaticValueProvider.of(options.getInstanceId()))
            .withDatabaseId(ValueProvider.StaticValueProvider.of(options.getDatabaseId()))
            .withRpcPriority(ValueProvider.StaticValueProvider.of(options.getSpannerPriority()));
    if (options.getMaxCommitDelay() >= 0) {
      spannerConfig = spannerConfig.withMaxCommitDelay(options.getMaxCommitDelay());
    }
    return spannerConfig;
  }
}