Plantilla de Sourcedb a Spanner

La plantilla de SourceDB a Spanner es una canalización por lotes que copia los datos de una base de datos relacional a una base de datos de Spanner existente. Esta canalización utiliza JDBC para conectarse a las bases de datos relacionales. Puedes usar esta plantilla para copiar datos desde cualquier base de datos relacional con controladores de JDBC disponibles en Spanner. Solo admite un conjunto limitado de tipos de MySQL.

Para obtener una capa adicional de protección, también puedes pasar una clave de Cloud KMS junto con un nombre de usuario, contraseña y parámetros de cadena de conexión codificados en Base64 encriptados con la clave de Cloud KMS. Consulta el extremo de encriptación de la API de Cloud KMS para obtener más información acerca de cómo encriptar los parámetros de nombre de usuario, contraseña y cadena de conexión.

Requisitos de la canalización

• Los controladores de JDBC de la base de datos relacional deben estar disponibles.
• Las tablas de Spanner deben existir antes de ejecutar la canalización.
• Las tablas de Spanner deben tener un esquema compatible.
• La base de datos relacional debe ser accesible desde la subred en la que se ejecuta Dataflow.

Parámetros de la plantilla

Parámetros obligatorios

• sourceConfigURL: Es la URL del archivo de configuración de la conexión de origen. El formato del archivo depende del tipo de fuente. En el caso de Astra, apuntará a un archivo de configuración de conexión de Astra (ejemplo). En el caso de JDBC, apuntará a un archivo de configuración de fragmentación de JDBC (ejemplo). En el caso de Cassandra, apuntará a un archivo de configuración del controlador de Cassandra (ejemplo). Este parámetro es obligatorio. Por ejemplo, gs://your-bucket/source-config.json. La configuración predeterminada es vacía.
• instanceId: Es la instancia de destino de Cloud Spanner.
• databaseId: Es la base de datos de destino de Cloud Spanner.
• projectId: Es el nombre del proyecto de Cloud Spanner.
• outputDirectory: Este directorio se usa para volcar los registros con errores, omitidos o filtrados en una migración.

Parámetros opcionales

• sourceDbDialect: Los valores posibles son CASSANDRA, MYSQL, POSTGRESQL, ORACLE y SQLSERVER. El valor predeterminado es MYSQL.
• jdbcDriverJars: Lista separada por comas de los archivos JAR del controlador. Por ejemplo, gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar La configuración predeterminada es vacía.
• jdbcDriverClassName: Es el nombre de la clase del controlador de JDBC. Por ejemplo, com.mysql.jdbc.Driver La configuración predeterminada es com.mysql.jdbc.Driver.
• tables: Tablas que se migrarán desde la fuente. La configuración predeterminada es vacía.
• numPartitions: Es la cantidad de particiones. Esto, junto con el límite inferior y superior, forma segmentaciones de particiones para las expresiones de cláusula WHERE generadas que se usan para dividir la columna de partición de manera uniforme. Cuando la entrada es menor que 1, se establece el número en 1. La configuración predeterminada es 0.
• fetchSize: Es la cantidad de filas que se recuperarán por lectura de página para la fuente de JDBC. Si no se configura, se infiere automáticamente a partir del tipo de máquina del trabajador y el tamaño estimado de la fila, y se recurre a 50,000 filas si no se puede inferir (por ejemplo, cuando no se especifica el tipo de máquina del trabajador). Si el dialecto de origen es MySQL, consulta la siguiente nota. En última instancia, esto se traduce en una llamada a Statement.setFetchSize en la capa de JDBC. SOLO se debe usar si el valor predeterminado arroja errores de memoria.Nota para la fuente de MySQL: El conector de MySQL ignora FetchSize, a menos que useCursorFetch=true también forme parte de las propiedades de conexión. En el caso del dialecto de MySQL, la canalización agregará useCursorFetch=true a las propiedades de conexión de forma predeterminada, a menos que el parámetro fetchSize se establezca explícitamente en 0.
• spannerHost: Es el extremo de Cloud Spanner al que se llamará en la plantilla. Por ejemplo, https://batch-spanner.googleapis.com La configuración predeterminada es https://batch-spanner.googleapis.com.
• maxConnections: Configura el grupo de conexiones de JDBC en cada trabajador con la cantidad máxima de conexiones. Usa un número negativo para no tener límite. Por ejemplo, -1 La configuración predeterminada es 0.
• sessionFilePath: Ruta del archivo de sesión en Cloud Storage que contiene información de asignación de la herramienta de migración de Spanner. La configuración predeterminada es vacía.
• transformationJarPath: Es la ubicación de archivo JAR personalizada en Cloud Storage que contiene la lógica de transformación personalizada para procesar registros. La configuración predeterminada es vacía.
• transformationClassName: Es el nombre de clase completamente calificado que tiene la lógica de transformación personalizada. Es un campo obligatorio en caso de que se especifique transformJarPath. La configuración predeterminada es vacía.
• transformationCustomParameters: Es la cadena que contiene cualquier parámetro personalizado que se pasará a la clase de transformación personalizada. La configuración predeterminada es vacía.
• insertOnlyModeForSpannerMutations: De forma predeterminada, la canalización usa Upserts para escribir filas en Spanner. Esto significa que se reemplazarían las filas existentes. Si el modo InsertOnly está habilitado, se usarán inserciones en lugar de upserts, y no se reemplazarán las filas existentes.
• batchSizeForSpannerMutations: Es el tamaño del lote en bytes para las mutaciones de Spanner. Si se establece en un valor inferior a 0, se usa el valor predeterminado de SpannerIO de Apache Beam, que es de 1 MB. Establece este valor en 0 o 10 para inhabilitar las mutaciones por lotes.
• spannerPriority: La prioridad de solicitud para llamadas de Cloud Spanner. El valor debe ser uno de los siguientes: [HIGH,MEDIUM,LOW]. El valor predeterminado es HIGH.
• tableOverrides: Son los reemplazos del nombre de la tabla desde la fuente hasta Spanner. Se escriben en el siguiente formato: [{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}]Este ejemplo muestra la asignación de la tabla Singers a Vocalists y la tabla Albums a Records. Por ejemplo, [{Singers, Vocalists}, {Albums, Records}] La configuración predeterminada es vacía.
• columnOverrides: Son las anulaciones del nombre de la columna desde la fuente hasta Spanner. Se escriben en el siguiente formato: [{SourceTableName1.SourceColumnName1, SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1, SourceTableName2.SpannerColumnName1}]. Ten en cuenta que SourceTableName debe seguir siendo el mismo en el par de origen y Spanner. Para anular los nombres de las tablas, usa tableOverrides.En el ejemplo, se muestra la asignación de SingerName a TalentName y de AlbumName a RecordName en las tablas Singers y Albums, respectivamente. Por ejemplo, [{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}] La configuración predeterminada es vacía.
• schemaOverridesFilePath: Es un archivo que especifica las anulaciones de la tabla y el nombre de la columna desde la fuente hasta Spanner. La configuración predeterminada es vacía.
• uniformizationStageCountHint: Es una sugerencia para la cantidad de etapas de uniformización. Actualmente, solo se aplica a las fuentes basadas en JDBC, como MySQL o PostgreSQL. Deja el valor 0 o el valor predeterminado para inhabilitar la uniformización. Se establece en -1 para una cantidad de etapas de log(numPartition). Si el espacio de clave primaria de origen se distribuye de manera uniforme (por ejemplo, una clave de incremento automático con espacios vacíos dispersos), es mejor dejarlo inhabilitado. Si tu espacio de claves no es uniforme, es posible que encuentres una VM rezagada en la ejecución de tu flujo de datos. En ese caso, puedes establecerlo en -1 para habilitar la uniformización. Si lo configuras manualmente en valores distintos de 0 o -1, podrás ajustar mejor la compensación entre la sobrecarga que agregan las etapas de uniformización y la mejora del rendimiento debido a una mejor distribución del trabajo.
• failureInjectionParameter: Es el parámetro de inyección de fallas. Solo se usa para pruebas. La configuración predeterminada es vacía.
• maxCommitDelay: Es el tiempo de demora máximo de la confirmación para optimizar la capacidad de procesamiento de escritura en Spanner. Consulta https://cloud.google.com/spanner/docs/throughput-optimized-writes.Set en -1 para permitir que Spanner elija el valor predeterminado. Se establece en un valor positivo para anular la mejor compensación posible entre la capacidad de procesamiento y la latencia.El valor predeterminado es -1.
• gcsOutputDirectory: Este directorio se usa para escribir los archivos AVRO de los registros leídos de la fuente. Por ejemplo, gs://your-bucket/your-path La configuración predeterminada es vacía.
• disabledAlgorithms: Algoritmos separados por comas que se deben inhabilitar. Si este valor se establece como none, no se inhabilita ningún algoritmo. Ten cuidado con este parámetro, ya que los algoritmos inhabilitados de forma predeterminada podrían tener vulnerabilidades o problemas de rendimiento. Por ejemplo, SSLv3, RC4.
• extraFilesToStage: Rutas de Cloud Storage separadas por comas o secretos de Secret Manager para los archivos que se deben almacenar en etapa intermedia en el trabajador. Estos archivos se guardan en el directorio /extra_files en cada trabajador. Por ejemplo, gs:///file.txt,projects//secrets//versions/

Ejecuta la plantilla

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates/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,\

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/VERSION/flex/Sourcedb_to_Spanner_Flex",
     "environment": { "maxWorkers": "10" }
  }
}

/*
 * 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.
 *
 * 
Check out 
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/sourcedb-to-spanner/README_Sourcedb_to_Spanner_Flex.md">README
 * 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 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;
  }
}