Modelo Sourcedb para Spanner

Beta

Esta funcionalidade está sujeita aos "Termos das Ofertas de Pré-DG" na secção Termos Gerais de Serviço dos Termos Específicos do Serviço. As funcionalidades de pré-DG estão disponíveis "tal como estão" e podem ter apoio técnico limitado. Para mais informações, consulte as descrições das fases de lançamento.

O modelo SourceDB para Spanner é um pipeline em lote que copia dados de uma base de dados relacional para uma base de dados do Spanner existente. Este pipeline usa JDBC para se ligar à base de dados relacional. Pode usar este modelo para copiar dados de qualquer base de dados relacional com controladores JDBC disponíveis para o Spanner. Isto só suporta um conjunto limitado de tipos do MySQL

Para uma camada adicional de proteção, também pode transmitir uma chave do Cloud KMS juntamente com parâmetros de nome de utilizador, palavra-passe e string de ligação codificados em Base64 encriptados com a chave do Cloud KMS. Consulte o ponto final da API Cloud KMS de encriptação para ver detalhes adicionais sobre a encriptação do nome de utilizador, da palavra-passe e dos parâmetros da string de ligação.

Requisitos do pipeline

• Os controladores JDBC para a base de dados relacional têm de estar disponíveis.
• As tabelas do Spanner têm de existir antes da execução do pipeline.
• As tabelas do Spanner têm de ter um esquema compatível.
• A base de dados relacional tem de estar acessível a partir da sub-rede onde o Dataflow é executado.

Parâmetros de modelos

Parâmetro	Descrição
sourceConfigURL	A string do URL de ligação JDBC. Por exemplo, jdbc:mysql://127.4.5.30:3306/my-db?autoReconnect=true&maxReconnects=10&unicode=true&characterEncoding=UTF-8 ou a configuração de fragmentação.
instanceId	A instância do Cloud Spanner de destino.
databaseId	A base de dados do Cloud Spanner de destino.
projectId	Este é o nome do projeto do Cloud Spanner.
outputDirectory	Este diretório é usado para transferir os registos com falhas/ignorados/filtrados numa migração.
jdbcDriverJars	Opcional: a lista de ficheiros JAR de controladores separados por vírgulas. Por exemplo: gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar. A predefinição é vazio.
jdbcDriverClassName	Opcional: o nome da classe do controlador JDBC. Por exemplo: com.mysql.jdbc.Driver. A predefinição é: com.mysql.jdbc.Driver.
username	Opcional: o nome de utilizador a usar para a ligação JDBC. A predefinição é vazio.
password	Opcional: a palavra-passe a usar para a ligação JDBC. A predefinição é vazio.
tables	Opcional: tabelas a migrar da origem. A predefinição é vazio.
numPartitions	Opcional: o número de partições. Isto, juntamente com o limite inferior e superior, forma passos de partições para expressões da cláusula WHERE geradas usadas para dividir a coluna de partição uniformemente. Quando o valor introduzido é inferior a 1, o número é definido como 1. Predefinição: 0.
spannerHost	Opcional: o ponto final do Cloud Spanner a chamar no modelo. Por exemplo: https://batch-spanner.googleapis.com. A predefinição é: https://batch-spanner.googleapis.com.
maxConnections	Opcional: configura o conjunto de ligações JDBC em cada trabalhador com o número máximo de ligações. Use um número negativo para não ter limite. Por exemplo: -1. Predefinição: 0.
sessionFilePath	Opcional: caminho da sessão no Cloud Storage que contém informações de mapeamento da ferramenta de migração do Spanner. A predefinição é vazia.
transformationJarPath	Opcional: localização do JAR personalizado no Cloud Storage que contém a lógica de transformação personalizada para o processamento de registos. A predefinição é vazia.
transformationClassName	Opcional: nome de classe totalmente qualificado com a lógica de transformação personalizada. É um campo obrigatório se transformationJarPath for especificado. A predefinição é vazio.
transformationCustomParameters	Opcional: string que contém quaisquer parâmetros personalizados a transmitir à classe de transformação personalizada. A predefinição é vazio.
disabledAlgorithms	Opcional: algoritmos separados por vírgulas a desativar. Se este valor for definido como nenhum, nenhum algoritmo é desativado. Use este parâmetro com cuidado, porque os algoritmos desativados por predefinição podem ter vulnerabilidades ou problemas de desempenho. Por exemplo: SSLv3, RC4.
extraFilesToStage	Opcional: caminhos do Cloud Storage separados por vírgulas ou segredos do Secret Manager para ficheiros a preparar no trabalhador. Estes ficheiros são guardados no diretório /extra_files em cada trabalhador. Por exemplo: gs://<BUCKET>/file.txt,projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<VERSION_ID>.

Execute o modelo

Consola

1. Aceda à página do fluxo de dados Criar tarefa a partir de um modelo.
Aceda a Criar tarefa a partir de modelo
2. No campo Nome da tarefa, introduza um nome exclusivo para a tarefa.
3. Opcional: para Ponto final regional, selecione um valor no menu pendente. A região predefinida é us-central1.

   Para ver uma lista das regiões onde pode executar uma tarefa do Dataflow, consulte as localizações do Dataflow.

4. No menu pendente Modelo do fluxo de dados, selecione the Sourcedb to Spanner template.
5. Nos campos de parâmetros fornecidos, introduza os valores dos parâmetros.
6. Clique em Executar tarefa.

CLI gcloud

Nota: para usar a CLI do Google Cloud para executar modelos flex, tem de ter a versão CLI do Google Cloud 284.0.0 ou posterior.

Na shell ou no terminal, execute o modelo:

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

Substitua o seguinte:

• JOB_NAME: um nome de tarefa exclusivo à sua escolha
• VERSION: a versão do modelo que quer usar

  Pode usar os seguintes valores:

  • latest para usar a versão mais recente do modelo, que está disponível na pasta principal sem data no contentor: gs://dataflow-templates-REGION_NAME/latest/
  • o nome da versão, como 2023-09-12-00_RC00, para usar uma versão específica do modelo, que pode ser encontrada aninhada na pasta principal com a data correspondente no contentor: gs://dataflow-templates-REGION_NAME/
  Atenção: a versão mais recente dos modelos pode ser atualizada com alterações destrutivas. Os seus ambientes de produção devem usar modelos mantidos na pasta principal datada mais recente para evitar que estas alterações significativas afetem os seus fluxos de trabalho de produção.
• REGION_NAME: a região onde quer implementar a tarefa do Dataflow, por exemplo, us-central1
• SOURCE_CONFIG_URL: o URL para estabelecer ligação ao anfitrião da base de dados de origem. Pode ser uma das seguintes opções: 1. O URL de ligação JDBC, que tem de conter o anfitrião, a porta e o nome da base de dados de origem, e pode conter opcionalmente propriedades como autoReconnect, maxReconnects, etc. Formato: `jdbc:mysql://{host}:{port}/{dbName}?{parameters}`2. O caminho da configuração do fragmento
• INSTANCE_ID: o ID da instância do Cloud Spanner.
• DATABASE_ID: o ID da base de dados do Cloud Spanner.
• PROJECT_ID: o ID do projeto do Cloud Spanner.
• OUTPUT_DIRECTORY: o diretório de saída para eventos com falhas/ignorados/filtrados

API

Para executar o modelo através da API REST, envie um pedido HTTP POST. Para mais informações sobre a API e os respetivos âmbitos de autorização, consulte 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" }
  }
}

Substitua o seguinte:

• PROJECT_ID: o ID do projeto onde quer executar a tarefa do Dataflow Google Cloud
• JOB_NAME: um nome de tarefa exclusivo à sua escolha
• VERSION: a versão do modelo que quer usar

  Pode usar os seguintes valores:

  • latest para usar a versão mais recente do modelo, que está disponível na pasta principal sem data no contentor: gs://dataflow-templates-REGION_NAME/latest/
  • o nome da versão, como 2023-09-12-00_RC00, para usar uma versão específica do modelo, que pode ser encontrada aninhada na pasta principal com a data correspondente no contentor: gs://dataflow-templates-REGION_NAME/
  Atenção: a versão mais recente dos modelos pode ser atualizada com alterações destrutivas. Os seus ambientes de produção devem usar modelos mantidos na pasta principal datada mais recente para evitar que estas alterações significativas afetem os seus fluxos de trabalho de produção.
• LOCATION: a região onde quer implementar a tarefa do Dataflow, por exemplo, us-central1
• SOURCE_CONFIG_URL: o URL para estabelecer ligação ao anfitrião da base de dados de origem. Pode ser uma das seguintes opções: 1. O URL de ligação JDBC, que tem de conter o anfitrião, a porta e o nome da base de dados de origem, e pode conter opcionalmente propriedades como autoReconnect, maxReconnects, etc. Formato: `jdbc:mysql://{host}:{port}/{dbName}?{parameters}`2. O caminho da configuração do fragmento
• INSTANCE_ID: o ID da instância do Cloud Spanner.
• DATABASE_ID: o ID da base de dados do Cloud Spanner.
• PROJECT_ID: o ID do projeto do Cloud Spanner.
• OUTPUT_DIRECTORY: o diretório de saída para eventos com falhas/ignorados/filtrados

Código-fonte do modelo

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.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.options.SourceDbToSpannerOptions;
import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard;
import com.google.cloud.teleport.v2.spanner.migrations.utils.SecretManagerAccessorImpl;
import com.google.cloud.teleport.v2.spanner.migrations.utils.ShardFileReader;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.util.List;
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();

    // Parse the user options passed from the command-line
    SourceDbToSpannerOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(SourceDbToSpannerOptions.class);
    run(options);
  }

  /**
   * 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) {
    // TODO - Validate if options are as expected
    Pipeline pipeline = Pipeline.create(options);

    SpannerConfig spannerConfig = createSpannerConfig(options);

    // Decide type and source of migration
    // TODO(vardhanvthigle): Move this within pipelineController.
    switch (options.getSourceDbDialect()) {
      case SourceDbToSpannerOptions.CASSANDRA_SOURCE_DIALECT:
        Preconditions.checkArgument(
            StringUtils.isNotEmpty(options.getSourceConfigURL()),
            "Cassandra Dialect needs sourceConfigURL to be set.");
        return PipelineController.executeCassandraMigration(options, pipeline, spannerConfig);
      case SourceDbToSpannerOptions.ASTRA_DB_SOURCE_DIALECT:
        return PipelineController.executeCassandraMigration(options, pipeline, spannerConfig);

      default:
        /* Implementation detail, not having a default leads to failure in compile time checks enforced here */
        /* Making jdbc as default case which includes MYSQL and PG. */
        Preconditions.checkArgument(
            StringUtils.isNotEmpty(options.getSourceConfigURL()),
            "JDBC based source needs sourceConfigURL to be set.");
        return executeJdbcMigration(options, pipeline, spannerConfig);
    }
  }

  // TODO(vardhanvthigle): Move this within pipelineController.
  private static PipelineResult executeJdbcMigration(
      SourceDbToSpannerOptions options, Pipeline pipeline, SpannerConfig spannerConfig) {
    if (options.getSourceConfigURL().startsWith("gs://")) {
      List<Shard> shards =
          new ShardFileReader(new SecretManagerAccessorImpl())
              .readForwardMigrationShardingConfig(options.getSourceConfigURL());
      return PipelineController.executeJdbcShardedMigration(
          options, pipeline, shards, spannerConfig);
    } else {
      return PipelineController.executeJdbcSingleInstanceMigration(
          options, pipeline, spannerConfig);
    }
  }

  @VisibleForTesting
  static SpannerConfig createSpannerConfig(SourceDbToSpannerOptions options) {
    return 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()));
  }
}