Plantilla de validación de datos de Avro de Cloud Storage a Spanner

Beta

Esta función está sujeta a las "Condiciones de la oferta previa a la DG" en la sección Condiciones Generales del Servicio de las Condiciones Específicas del Servicio. Las funciones de la fase previa a la DG están disponibles “sin modificaciones” y pueden tener asistencia limitada. Para obtener más información, consulta las descripciones de la etapa de lanzamiento.

La plantilla de Cloud Storage Avro a Spanner Data Validation es una canalización por lotes que lee datos de Cloud Storage y Spanner, y los compara para validar la corrección de la migración.

Requisitos de la canalización

• El directorio de Cloud Storage para los archivos Avro debe existir antes de ejecutar la canalización. Para generar estos archivos Avro, ejecuta la canalización de SourceDB a Cloud Spanner con el parámetro gcsOutputDirectory.
• El conjunto de datos de destino de BigQuery para los resultados de la validación debe existir antes de ejecutar la canalización.
• Las tablas de Spanner deben existir antes de ejecutar la canalización.
• Las tablas de Spanner deben tener un esquema compatible (ya sea directamente o a través de la asignación de esquemas).
• Para validar una migración realizada por el canal de SourceDB a Cloud Spanner, debes pasar valores idénticos para cualquier configuración compartida (como sessionFilePath, schemaOverridesFilePath y transformationJarPath).

Parámetros de la plantilla

Parámetros obligatorios

• gcsInputDirectory: Este directorio se usa para leer los archivos AVRO de los registros leídos desde la fuente. Por ejemplo, gs://your-bucket/your-path
• projectId: Es el nombre del proyecto de Cloud Spanner.
• instanceId: Es la instancia de destino de Cloud Spanner.
• databaseId: Es la base de datos de destino de Cloud Spanner.
• bigQueryDataset: Es el ID del conjunto de datos de BigQuery en el que se almacenarán los resultados de la validación. Por ejemplo, validation_report_dataset

Parámetros opcionales

• 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.
• 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.
• 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.
• 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.
• tableOverrides: Son los reemplazos del nombre de la tabla desde la fuente hasta Spanner. Se escriben en el siguiente formato: [{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}]. En este ejemplo, se 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.
• runId: Es un identificador único para la ejecución de validación. Si no se proporciona, se usará el nombre del trabajo de Dataflow. Por ejemplo, run_20230101_120000
• 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.
• tables: Es una lista separada por comas de las tablas fuente que se incluirán en la ejecución de la validación. La configuración predeterminada es vacía.
• tableConfigurationFilePath: Es la ruta de acceso a un archivo de GCS que contiene una lista en formato JSON de las tablas de origen que se validarán. Debe ser un archivo JSON con la estructura {"tableNames": ["table1", "table2"]}. De manera opcional, optionalConfigurations.<sourceTable>.spannerQuery establece la consulta de Spanner que se usa para leer esa tabla, p.ej., para restringirla al shardIds seleccionado. Clave una tabla solo de Spanner por su nombre de Spanner. Consulta la sección "Validating a subset of shards" del archivo README.md del módulo para conocer las reglas de consulta. La configuración predeterminada es vacía.
• shardIds: Es una lista separada por comas de IDs de fragmentos lógicos (los IDs que usa la plantilla sourcedb-to-spanner, que aparecen como el directorio <table>/<shardId>/ en el resultado de Avro). Cuando se establece, solo se validan esos fragmentos. Las tablas con un archivo de sesión ShardIdColumn se filtran en esa columna. Todas las demás tablas dentro del alcance necesitan un spannerQuery en tableConfigurationFilePath. De lo contrario, el trabajo fallará. Por ejemplo, shard_001,shard_007. La configuración predeterminada es vacía.

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 la plantilla Validación de datos de Spanner a partir de archivos Avro de Cloud Storage.
5. En los campos de parámetros proporcionados, ingresa los valores de tus parámetros.
6. Haz clic en Ejecutar trabajo.

gcloud CLI

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/VERSION/flex/Avro_to_Spanner_Data_Validator \
    --region=REGION_NAME \
    --parameters \
       gcsInputDirectory=PATH_TO_AVRO_FILES,\
       projectId=PROJECT_ID,\
       instanceId=INSTANCE_ID,\
       databaseId=DATABASE_ID,\
       bigQueryDataset=BIGQUERY_DATASET

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/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/
  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
• PATH_TO_AVRO_FILES: Es la ruta de Cloud Storage a los archivos Avro.
• PROJECT_ID: Es el ID del proyecto de Spanner.
• INSTANCE_ID: Es el ID de la instancia de Spanner de destino.
• DATABASE_ID: Es el ID de la base de datos de Spanner de destino.
• BIGQUERY_DATASET: Es el ID del conjunto de datos de BigQuery para los resultados de la validación.

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": {
       "gcsInputDirectory": "PATH_TO_AVRO_FILES",
       "projectId": "PROJECT_ID",
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "bigQueryDataset": "BIGQUERY_DATASET"
     },
     "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/Avro_to_Spanner_Data_Validator"
  }
}

Reemplaza lo siguiente:

• PROJECT_ID: El Google Cloud ID del proyecto 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/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/
  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
• PATH_TO_AVRO_FILES: Es la ruta de Cloud Storage a los archivos Avro.
• PROJECT_ID: Es el ID del proyecto de Spanner.
• INSTANCE_ID: Es el ID de la instancia de Spanner de destino.
• DATABASE_ID: Es el ID de la base de datos de Spanner de destino.
• BIGQUERY_DATASET: Es el ID del conjunto de datos de BigQuery para los resultados de la validación.

Código fuente de la plantilla

Java

/*
 * Copyright (C) 2026 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.constants.GCSSpannerDVConstants.SOURCE_TAG;
import static com.google.cloud.teleport.v2.constants.GCSSpannerDVConstants.SPANNER_TAG;

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.config.TableConfiguration;
import com.google.cloud.teleport.v2.dto.ComparisonRecord;
import com.google.cloud.teleport.v2.fn.SchemaMapperProviderFn;
import com.google.cloud.teleport.v2.options.GCSSpannerDVOptions;
import com.google.cloud.teleport.v2.spanner.ddl.Ddl;
import com.google.cloud.teleport.v2.spanner.migrations.schema.ISchemaMapper;
import com.google.cloud.teleport.v2.spanner.migrations.transformation.CustomTransformation;
import com.google.cloud.teleport.v2.transforms.MatchRecordsTransform;
import com.google.cloud.teleport.v2.transforms.ReportResultsTransform;
import com.google.cloud.teleport.v2.transforms.SourceReaderTransform;
import com.google.cloud.teleport.v2.transforms.SpannerInformationSchemaProcessorTransform;
import com.google.cloud.teleport.v2.transforms.SpannerReaderTransform;
import com.google.common.annotations.VisibleForTesting;
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.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.PCollectionView;
import org.joda.time.Instant;

@Template(
    name = "Avro_to_Spanner_Data_Validator",
    category = TemplateCategory.BATCH,
    displayName = "Cloud Storage Avro files to Spanner Data Validation",
    description =
        "Batch pipeline that reads data from Cloud Storage and Spanner and compares them to validate migration"
            + " correctness.",
    optionsClass = GCSSpannerDVOptions.class,
    flexContainerName = "avro-to-spanner-dv",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/avro-to-spanner-dv",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The Cloud Storage directory for Avro files must exist before pipeline execution.",
      "The target BigQuery dataset for validation results must exist before pipeline execution.",
      "The Spanner tables must exist before pipeline execution.",
      "The Spanner tables must have a compatible schema (either directly or through schema mapping)."
    })
public class GCSSpannerDV {

  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    GCSSpannerDVOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(GCSSpannerDVOptions.class);
    run(options);
  }

  public static PipelineResult run(GCSSpannerDVOptions options) {
    Pipeline pipeline = Pipeline.create(options);

    TableConfiguration tableConfig = TableConfiguration.parseFromOptions(options);

    SpannerConfig spannerConfig = createSpannerConfig(options);

    // Fetch Spanner DDL using Info schema
    final PCollectionView<Ddl> ddlView =
        pipeline.apply(
            "ReadSpannerInformationSchema",
            new SpannerInformationSchemaProcessorTransform(spannerConfig));

    // Get Schema mapper provider, we get Ddl from a side input
    // so the mapper has to be initialized lazily
    SerializableFunction<Ddl, ISchemaMapper> schemaMapperProvider =
        new SchemaMapperProviderFn(
            options.getSessionFilePath(),
            options.getSchemaOverridesFilePath(),
            options.getTableOverrides(),
            options.getColumnOverrides());

    CustomTransformation customTransformation =
        CustomTransformation.builder(
                options.getTransformationJarPath(), options.getTransformationClassName())
            .setCustomParameters(options.getTransformationCustomParameters())
            .build();

    // Get Source records hashes
    PCollection<ComparisonRecord> sourceRecords =
        pipeline.apply(
            "ReadSourceRecords",
            new SourceReaderTransform(
                options.getGcsInputDirectory(),
                ddlView,
                schemaMapperProvider,
                customTransformation,
                tableConfig));

    // Get Spanner records hashes
    PCollection<ComparisonRecord> spannerRecords =
        pipeline.apply(
            "ReadSpannerRecords",
            new SpannerReaderTransform(spannerConfig, ddlView, schemaMapperProvider, tableConfig));

    PCollectionTuple inputs =
        PCollectionTuple.of(SOURCE_TAG, sourceRecords).and(SPANNER_TAG, spannerRecords);

    // Match records to determine equivalence
    PCollectionTuple matchResults = inputs.apply("MatchRecords", new MatchRecordsTransform());

    // Report results of the validation
    Instant startTimestamp = Instant.now();
    String runId = options.getRunId();
    if (runId == null) {
      runId = String.format("%s_%s", options.getJobName(), startTimestamp);
    }

    matchResults.apply(
        "ReportResults",
        new ReportResultsTransform(options.getBigQueryDataset(), runId, startTimestamp));

    return pipeline.run();
  }

  @VisibleForTesting
  static SpannerConfig createSpannerConfig(GCSSpannerDVOptions 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()));
  }
}