Modelo de validação de dados do Cloud Storage Avro para Spanner

Beta

Este recurso está sujeito aos Termos das Soluções Pré-GA na seção Termos Gerais de Serviço dos Termos Específicos de Serviço. Os recursos pré-GA estão disponíveis "no estado em que se encontram" e podem ter suporte limitado. Para mais informações, consulte as descrições da fase de lançamento.

O modelo de arquivos Avro do Cloud Storage para validação de dados do Spanner é um pipeline em lote que lê dados do Cloud Storage e do Spanner e os compara para validar a correção da migração.

Requisitos de pipeline

• O diretório do Cloud Storage para arquivos Avro precisa existir antes da execução do pipeline. Gere esses arquivos Avro executando o pipeline SourceDB para Cloud Spanner com o parâmetro gcsOutputDirectory.
• O conjunto de dados de destino do BigQuery para resultados da validação precisa existir antes da execução do pipeline.
• As tabelas do Spanner precisam existir antes da execução do pipeline.
• As tabelas do Spanner precisam ter um esquema compatível (diretamente ou por mapeamento de esquema).
• Para validar uma migração realizada pelo pipeline do SourceDB para o Cloud Spanner, transmita valores idênticos para todas as configurações compartilhadas (como sessionFilePath, schemaOverridesFilePath e transformationJarPath).

Parâmetros do modelo

Parâmetros obrigatórios

• gcsInputDirectory: esse diretório é usado para ler os arquivos AVRO dos registros lidos da origem. Por exemplo, gs://your-bucket/your-path.
• projectId: o nome do projeto do Cloud Spanner.
• instanceId: a instância de destino do Cloud Spanner.
• databaseId: o banco de dados de destino do Cloud Spanner.
• bigQueryDataset: o ID do conjunto de dados do BigQuery em que os resultados da validação serão armazenados. Por exemplo, validation_report_dataset.

Parâmetros opcionais

• spannerHost: o endpoint do Cloud Spanner para chamar no modelo. Por exemplo, https://batch-spanner.googleapis.com. O padrão é: https://batch-spanner.googleapis.com.
• spannerPriority: a prioridade da solicitação para chamadas do Cloud Spanner. O valor precisa ser um destes: [HIGH,MEDIUM,LOW]. O padrão é HIGH.
• sessionFilePath: o caminho do arquivo de sessão no Cloud Storage que contém informações de mapeamento da ferramenta de migração do Spanner. O padrão é vazio.
• schemaOverridesFilePath: um arquivo que especifica as substituições de nome da tabela e da coluna da origem para o Spanner. O padrão é vazio.
• tableOverrides: substituições de nome de tabela da origem para o Spanner. Eles são gravados no seguinte formato: [{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}]. Este exemplo mostra o mapeamento da tabela "Singers" para "Vocalists" e da tabela "Albums" para "Records". Por exemplo, [{Singers, Vocalists}, {Albums, Records}]. O padrão é vazio.
• columnOverrides: substituições de nome de coluna da origem para o Spanner. Eles são gravados no seguinte formato: [{SourceTableName1.SourceColumnName1, SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1, SourceTableName2.SpannerColumnName1}]. O SourceTableName precisa ser o mesmo na origem e no par do Spanner. Para substituir nomes de tabelas, use tableOverrides.O exemplo mostra o mapeamento de SingerName para TalentName e AlbumName para RecordName nas tabelas "Singers" e "Albums", respectivamente. Por exemplo, [{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]. O padrão é vazio.
• runId: um identificador exclusivo da execução da validação. Se não for fornecido, o nome do job do Dataflow será usado. Por exemplo, run_20230101_120000.
• transformationJarPath: local do jar personalizado no Cloud Storage que contém a lógica de transformação personalizada para processar registros. O padrão é vazio.
• transformationClassName: o nome de classe totalmente qualificado com lógica de transformação personalizada. É um campo obrigatório no caso de transformationJarPath é especificado. O padrão é vazio.
• transformationCustomParameters: string contendo os parâmetros personalizados que serão passados para a classe de transformação personalizada. O padrão é vazio.
• tables: uma lista separada por vírgulas de tabelas de origem a serem incluídas na execução da validação. O padrão é vazio.
• tableConfigurationFilePath: um caminho de arquivo do GCS que contém uma lista JSON de tabelas de origem a serem validadas. Precisa ser um arquivo JSON com a estrutura {"tableNames": ["table1", "table2"]}. Opcionalmente, optionalConfigurations.<sourceTable>.spannerQuery define a consulta do Spanner usada para ler essa tabela, por exemplo, para restringir ao shardIds selecionado. Chaveie uma tabela somente do Spanner pelo nome dela no Spanner. Consulte a seção "Validação de um subconjunto de fragmentos" do README.md do módulo para ver as regras de consulta. O padrão é vazio.
• shardIds: uma lista separada por vírgulas de IDs de fragmentos lógicos (os IDs usados pelo modelo sourcedb-to-spanner, que aparecem como o diretório <table>/<shardId>/ na saída Avro). Quando definido, apenas esses fragmentos são validados. As tabelas com um arquivo de sessão ShardIdColumn são filtradas nessa coluna. Todas as outras tabelas no escopo precisam de um spannerQuery em tableConfigurationFilePath. Caso contrário, o job falha. Por exemplo, shard_001,shard_007. O padrão é vazio.

Executar o modelo

Console

1. Acesse a página Criar job usando um modelo do Dataflow.
Acesse Criar job usando um modelo
2. No campo Nome do job, insira um nome exclusivo.
3. Opcional: em Endpoint regional, selecione um valor no menu suspenso. A região padrão é us-central1.

   Para ver uma lista de regiões em que é possível executar um job do Dataflow, consulte Locais do Dataflow.

4. No menu suspenso Modelo do Dataflow, selecione o modelo Arquivos Avro do Cloud Storage para validação de dados do Spanner.
5. Nos campos de parâmetro fornecidos, insira os valores de parâmetro.
6. Cliquem em Executar job.

CLI da gcloud

Observação: para usar a Google Cloud CLI para executar modelos flex você precisa ter a Google Cloud CLI na versão 284.0.0 ou mais recente.

No shell ou no terminal, execute o modelo:

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

Substitua:

• JOB_NAME: um nome de job de sua escolha
• VERSION: a versão do modelo que você quer usar

  Use estes valores:

  • latest para usar a versão mais recente do modelo, disponível na pasta mãe não datada no bucket: gs://dataflow-templates/latest/
  • o nome da versão, como 2023-09-12-00_RC00, para usar uma versão específica do modelo, que pode ser aninhada na respectiva pasta mãe datada no bucket: gs://dataflow-templates
  Cuidado: é possível que a versão mais recente dos modelos seja atualizada com alterações interruptivas. Os ambientes de produção precisam usar os modelos mantidos na pasta mãe datada mais recente para evitar que essas alterações interruptivas afetem os fluxos de trabalho de produção.
• REGION_NAME: a região em que você quer implantar o job do Dataflow, por exemplo, us-central1
• PATH_TO_AVRO_FILES: o caminho do Cloud Storage para os arquivos Avro.
• PROJECT_ID: o ID do projeto do Spanner.
• INSTANCE_ID: o ID da instância de destino do Spanner.
• DATABASE_ID: o ID do banco de dados de destino do Spanner.
• BIGQUERY_DATASET: o ID do conjunto de dados do BigQuery para os resultados da validação.

API

Para executar o modelo usando a API REST, envie uma solicitação POST HTTP. Para mais informações sobre a API e os respectivos escopos 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": {
       "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"
  }
}

Substitua:

• PROJECT_ID: o ID do projeto Google Cloud em que você quer executar o job do Dataflow
• JOB_NAME: um nome de job de sua escolha
• VERSION: a versão do modelo que você quer usar

  Use estes valores:

  • latest para usar a versão mais recente do modelo, disponível na pasta mãe não datada no bucket: gs://dataflow-templates/latest/
  • o nome da versão, como 2023-09-12-00_RC00, para usar uma versão específica do modelo, que pode ser aninhada na respectiva pasta mãe datada no bucket: gs://dataflow-templates
  Cuidado: é possível que a versão mais recente dos modelos seja atualizada com alterações interruptivas. Os ambientes de produção precisam usar os modelos mantidos na pasta mãe datada mais recente para evitar que essas alterações interruptivas afetem os fluxos de trabalho de produção.
• LOCATION: a região em que você quer implantar o job do Dataflow, por exemplo, us-central1
• PATH_TO_AVRO_FILES: o caminho do Cloud Storage para os arquivos Avro.
• PROJECT_ID: o ID do projeto do Spanner.
• INSTANCE_ID: o ID da instância de destino do Spanner.
• DATABASE_ID: o ID do banco de dados de destino do Spanner.
• BIGQUERY_DATASET: o ID do conjunto de dados do BigQuery para os resultados da validação.

Código-fonte do modelo

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()));
  }
}