Cloud Storage Avro to Spanner Data Validation 模板

Beta 版

此功能受服务专用条款的“通用服务条款”部分中的“非正式版产品条款”的约束。 非正式版功能“按原样”提供，可能只能获得有限的支持。 如需了解详情，请参阅发布阶段说明。

Cloud Storage Avro files to Spanner Data Validation 模板是一种批处理流水线，可从 Cloud Storage 和 Spanner 读取数据并进行比较，以验证迁移的正确性。

流水线要求

• 在执行流水线之前，必须已存在用于存放 Avro 文件的 Cloud Storage 目录。通过运行 SourceDB to Cloud Spanner 流水线并使用 gcsOutputDirectory 参数来生成这些 Avro 文件。
• 在执行流水线之前，必须已存在用于存储验证结果的目标 BigQuery 数据集。
• 在执行流水线之前，Spanner 表必须已存在。
• Spanner 表必须具有兼容的架构（直接兼容或通过架构映射兼容）。
• 如需验证由 SourceDB 到 Cloud Spanner 流水线执行的迁移，您必须为所有共享配置（例如 sessionFilePath、schemaOverridesFilePath 和 transformationJarPath）传递相同的值。

模板参数

必需参数

• gcsInputDirectory：此目录用于读取从来源读取的记录的 AVRO 文件。例如 gs://your-bucket/your-path。
• projectId：这是 Cloud Spanner 项目的名称。
• instanceId：目标 Cloud Spanner 实例。
• databaseId：目标 Cloud Spanner 数据库。
• bigQueryDataset：将存储验证结果的 BigQuery 数据集 ID。例如 validation_report_dataset。

可选参数

• spannerHost：要在模板中调用的 Cloud Spanner 端点。例如 https://batch-spanner.googleapis.com。默认值为：https://batch-spanner.googleapis.com。
• spannerPriority：Cloud Spanner 调用的请求优先级。该值必须为以下值之一：[HIGH,MEDIUM,LOW]。默认值为 HIGH。
• sessionFilePath：Cloud Storage 中的会话文件路径，其中包含 Spanner 迁移工具的映射信息。默认值为空。
• schemaOverridesFilePath：一个文件，用于指定从源到 Spanner 的表和列名称替换。默认值为空。
• tableOverrides：这些是从源到 Spanner 的表名称替换。它们采用以下格式编写：[{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}]。此示例展示了如何将 Singers 表映射到 Vocalists 表，并将 Albums 表映射到 Records 表。例如 [{Singers, Vocalists}, {Albums, Records}]。默认值为空。
• columnOverrides：这些是从源到 Spanner 的列名称替换。它们的格式如下：[{SourceTableName1.SourceColumnName1, SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1, SourceTableName2.SpannerColumnName1}]。请注意，SourceTableName 在源和 Spanner 对中应保持不变。如需替换表名称，请使用 tableOverrides。此示例展示了如何分别将 Singers 表中的 SingerName 映射到 TalentName，以及将 Albums 表中的 AlbumName 映射到 RecordName。例如 [{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]。默认值为空。
• runId：验证运行的唯一标识符。如果未提供，则使用 Dataflow 作业名称。例如 run_20230101_120000。
• transformationJarPath：Cloud Storage 中的自定义 jar 位置，其中包含用于处理记录的自定义转换逻辑。默认值为空。
• transformationClassName：包含自定义转换逻辑的完全限定类名称。如果指定了 transformationJarPath，则这是必填字段。默认值为空。
• transformationCustomParameters：包含要传递给自定义转换类的任何自定义参数的字符串。默认值为空。
• tables：要纳入验证运行的源表的英文逗号分隔列表。默认值为空。
• tableConfigurationFilePath：包含要验证的源表的 JSON 列表的 GCS 文件路径。该文件必须是 JSON 文件，且结构为 {"tableNames": ["table1", "table2"]}。默认值为空。

运行模板

控制台

1. 转到 Dataflow 基于模板创建作业页面。
转到“基于模板创建作业”
2. 在作业名称字段中，输入唯一的作业名称。
3. 可选：对于区域性端点，从下拉菜单中选择一个值。默认区域为 us-central1。

   如需查看可以在其中运行 Dataflow 作业的区域列表，请参阅 Dataflow 位置。

4. 从 Dataflow 模板下拉菜单中，选择 Cloud Storage Avro 文件到 Spanner 数据验证模板。
5. 在提供的参数字段中，输入您的参数值。
6. 点击运行作业。

gcloud CLI

注意：如需使用 Google Cloud CLI 运行 flex模板，您必须具有 Google Cloud CLI 284.0.0 或更高版本。

在 shell 或终端中，运行模板：

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

替换以下内容：

• JOB_NAME：您选择的唯一性作业名称
• VERSION：您要使用的模板的版本

  您可使用以下值：

  • latest，以使用模板的最新版本，该模板在存储桶的未标示日期的父文件夹 (gs://dataflow-templates/latest/) 中可用
  • 版本名称（如 2023-09-12-00_RC00），以使用模板的特定版本，该版本嵌套在存储桶的相应日期父文件夹 (gs://dataflow-templates/) 中
  注意：最新版模板可能会随着重大更改而更新。为了防止这些重大更改影响您的生产工作流程，生产环境应使用有最近标示日期的父文件夹中保存的模板。
• REGION_NAME：要在其中部署 Dataflow 作业的区域，例如 us-central1
• PATH_TO_AVRO_FILES：Avro 文件的 Cloud Storage 路径。
• PROJECT_ID：Spanner 项目 ID。
• INSTANCE_ID：目标 Spanner 实例 ID。
• DATABASE_ID：目标 Spanner 数据库 ID。
• BIGQUERY_DATASET：验证结果的 BigQuery 数据集 ID。

API

如需使用 REST API 来运行模板，请发送 HTTP POST 请求。如需详细了解 API 及其授权范围，请参阅 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"
  }
}

替换以下内容：

• PROJECT_ID：您要在其中运行 Dataflow 作业的 Google Cloud 项目 ID
• JOB_NAME：您选择的唯一性作业名称
• VERSION：您要使用的模板的版本

  您可使用以下值：

  • latest，以使用模板的最新版本，该模板在存储桶的未标示日期的父文件夹 (gs://dataflow-templates/latest/) 中可用
  • 版本名称（如 2023-09-12-00_RC00），以使用模板的特定版本，该版本嵌套在存储桶的相应日期父文件夹 (gs://dataflow-templates/) 中
  注意：最新版模板可能会随着重大更改而更新。为了防止这些重大更改影响您的生产工作流程，生产环境应使用有最近标示日期的父文件夹中保存的模板。
• LOCATION：要在其中部署 Dataflow 作业的区域，例如 us-central1
• PATH_TO_AVRO_FILES：Avro 文件的 Cloud Storage 路径。
• PROJECT_ID：Spanner 项目 ID。
• INSTANCE_ID：目标 Spanner 实例 ID。
• DATABASE_ID：目标 Spanner 数据库 ID。
• BIGQUERY_DATASET：验证结果的 BigQuery 数据集 ID。

模板源代码

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