Plantilla de Sourcedb a Spanner

Beta

Esta función está sujeta a las "Condiciones de las Ofertas de Fase Previa a la DG" de la sección Condiciones Generales del Servicio de las Condiciones Específicas del Servicio. Las funciones de fase previa a la DG están disponibles “tal cual” y podrían tener asistencia limitada. Para obtener más información, consulta las descripciones de la etapa de lanzamiento.

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ámetro	Descripción
sourceConfigURL	La string de la URL de la conexión de JDBC. Por ejemplo, jdbc:mysql://127.4.5.30:3306/my-db?autoReconnect=true&maxReconnects=10&unicode=true&characterEncoding=UTF-8 o la configuración del fragmento.
instanceId	La instancia de destino de Cloud Spanner.
databaseId	La base de datos de destino de Cloud Spanner.
projectId	Este 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.
jdbcDriverJars	Opcional: La 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	Opcional: El nombre de la clase del controlador de JDBC. Por ejemplo: com.mysql.jdbc.Driver. La configuración predeterminada es com.mysql.jdbc.Driver.
username	Opcional: El nombre de usuario para usar en la conexión de JDBC. La configuración predeterminada es vacía.
password	Opcional: La contraseña para usar en la conexión de JDBC. La configuración predeterminada es vacía.
tables	Opcional: Las tablas que se migrarán desde la fuente. La configuración predeterminada es vacía.
numPartitions	La cantidad de particiones (opcional). 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.
spannerHost	El extremo de Cloud Spanner al que se llamará en la plantilla (opcional). Por ejemplo: https://batch-spanner.googleapis.com. La configuración predeterminada es https://batch-spanner.googleapis.com.
maxConnections	Opcional: 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	Opcional: Ruta de acceso de la 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	Opcional: 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	Opcional: 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	Opcional: 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.
disabledAlgorithms	Opcional: Algoritmos separados por comas que se deben inhabilitar. Si este valor se establece como ninguno, 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://<BUCKET>/file.txt,projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<VERSION_ID>.

Ejecuta la plantilla

Console

1. Ve a la página Crear un trabajo a partir de una plantilla de Dataflow.
Ir a Crear un trabajo a partir de una plantilla
2. En el campo Nombre del trabajo, ingresa un nombre de trabajo único.
3. Opcional: Para Extremo regional, selecciona un valor del menú desplegable. La región predeterminada es us-central1.

   Para obtener una lista de regiones en las que puedes ejecutar un trabajo de Dataflow, consulta Ubicaciones de Dataflow.

4. En el menú desplegable Plantilla de Dataflow, selecciona the Sourcedb to Spanner template.
5. En los campos de parámetros proporcionados, ingresa los valores de tus parámetros.
6. Haz clic en Ejecutar trabajo.

CLI de gcloud

Nota: Para usar Google Cloud CLI a fin de ejecutar plantillas flex, debes tener Google Cloud CLI en la versión 284.0.0 o una versión posterior.

En tu shell o terminal, ejecuta la plantilla:

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

Reemplaza lo siguiente:

• JOB_NAME: Es el nombre del trabajo que elijas
• VERSION: Es la versión de la plantilla que deseas usar.

  Puedes usar los siguientes valores:

  • latest para usar la última versión de la plantilla, que está disponible en la carpeta superior non-dated en el bucket gs://dataflow-templates-REGION_NAME/latest/
  • el nombre de la versión, como 2023-09-12-00_RC00, para usar una versión específica de la plantilla, que se puede encontrar anidada en la carpeta superior con fecha correspondiente en el bucket gs://dataflow-templates-REGION_NAME/
  Precaución: La versión más reciente de las plantillas podría actualizarse con cambios rotundos. Los entornos de producción deben usar plantillas que se conserven en la carpeta superior con la fecha más reciente para evitar que estos cambios rotundos afecten los flujos de trabajo de producción.
• REGION_NAME: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
• SOURCE_CONFIG_URL: La URL para conectarse al host de la base de datos de origen. Puede ser cualquiera de las siguientes opciones: La URL de conexión de JDBC, que debe contener el host, el puerto y el nombre de la base de datos de origen y, de forma opcional, puede contener propiedades como autoReconnect, maxReconnects, etc. Formato: `jdbc:mysql://{host}:{port}/{dbName}?{parameters}`2. La ruta de acceso de configuración del fragmento
• INSTANCE_ID: El ID de la instancia de Cloud Spanner.
• DATABASE_ID: El ID de la base de datos de Cloud Spanner.
• PROJECT_ID: El ID del proyecto de Cloud Spanner.
• OUTPUT_DIRECTORY: El directorio de salida para los eventos con errores, omitidos o filtrados.

API

Para ejecutar la plantilla con la API de REST, envía una solicitud POST HTTP. Para obtener más información de la API y sus permisos de autorización, 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" }
  }
}

Reemplaza lo siguiente:

• PROJECT_ID: El ID del proyecto de Google Cloud en el que deseas ejecutar el trabajo de Dataflow.
• JOB_NAME: Es el nombre del trabajo que elijas
• VERSION: Es la versión de la plantilla que deseas usar.

  Puedes usar los siguientes valores:

  • latest para usar la última versión de la plantilla, que está disponible en la carpeta superior non-dated en el bucket gs://dataflow-templates-REGION_NAME/latest/
  • el nombre de la versión, como 2023-09-12-00_RC00, para usar una versión específica de la plantilla, que se puede encontrar anidada en la carpeta superior con fecha correspondiente en el bucket gs://dataflow-templates-REGION_NAME/
  Precaución: La versión más reciente de las plantillas podría actualizarse con cambios rotundos. Los entornos de producción deben usar plantillas que se conserven en la carpeta superior con la fecha más reciente para evitar que estos cambios rotundos afecten los flujos de trabajo de producción.
• LOCATION: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
• SOURCE_CONFIG_URL: La URL para conectarse al host de la base de datos de origen. Puede ser cualquiera de las siguientes opciones: La URL de conexión de JDBC, que debe contener el host, el puerto y el nombre de la base de datos de origen y, de forma opcional, puede contener propiedades como autoReconnect, maxReconnects, etc. Formato: `jdbc:mysql://{host}:{port}/{dbName}?{parameters}`2. La ruta de acceso de configuración del fragmento
• INSTANCE_ID: El ID de la instancia de Cloud Spanner.
• DATABASE_ID: El ID de la base de datos de Cloud Spanner.
• PROJECT_ID: El ID del proyecto de Cloud Spanner.
• OUTPUT_DIRECTORY: El directorio de salida para los eventos con errores, omitidos o filtrados.

Código fuente de la plantilla

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 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.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 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",
      "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 = 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 {

  private static final Logger LOG = LoggerFactory.getLogger(SourceDbToSpanner.class);

  /**
   * 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
    if (options.getSourceConfigURL().startsWith("gs://")) {
      List<Shard> shards =
          new ShardFileReader(new SecretManagerAccessorImpl())
              .readForwardMigrationShardingConfig(options.getSourceConfigURL());
      return PipelineController.executeShardedMigration(options, pipeline, shards, spannerConfig);
    } else {
      return PipelineController.executeSingleInstanceMigration(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()));
  }
}