Cloud Storage Avro 到 Spanner 資料驗證範本

Beta 版

這項功能適用《服務專屬條款》中「一般服務條款」一節的《正式發布前產品條款》。正式發布前功能是依「原樣」提供，支援服務可能受限。 詳情請參閱推出階段說明。

「Cloud Storage Avro 檔案至 Spanner 資料驗證」範本是一種批次管道，可從 Cloud Storage 和 Spanner 讀取資料並進行比較，以驗證遷移作業是否正確。

管道相關規定

• 在管道執行作業前，Avro 檔案的 Cloud Storage 目錄必須已存在。執行 SourceDB to Cloud Spanner 管道並使用 gcsOutputDirectory 參數，即可產生這些 Avro 檔案。
• 在管道執行作業前，驗證結果的目標 BigQuery 資料集必須已存在。
• Spanner 資料表必須在管道執行作業前就已存在。
• Spanner 資料表必須具有相容的結構定義 (直接或透過結構定義對應)。
• 如要驗證 SourceDB to 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}]。請注意，來源和 Spanner 配對中的 SourceTableName 應保持不變。如要覆寫資料表名稱，請使用 tableOverrides。以下範例分別顯示 Singers 和 Albums 資料表中的 SingerName 對應至 TalentName，以及 AlbumName 對應至 RecordName。例如，[{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]。預設為空白。
• runId：驗證執行的專屬 ID。如未提供，系統會使用 Dataflow 工作名稱。例如：run_20230101_120000。
• transformationJarPath：Cloud Storage 中的自訂 JAR 位置，內含用於處理記錄的自訂轉換邏輯。預設為空白。
• transformationClassName：具有自訂轉換邏輯的完整類別名稱。如果指定 transformationJarPath，這個欄位就必須填寫。預設為空白。
• transformationCustomParameters：字串，包含要傳遞至自訂轉換類別的任何自訂參數。預設為空白。
• 資料表：以半形逗號分隔的來源資料表清單，用於驗證作業。預設為空白。
• tableConfigurationFilePath：GCS 檔案路徑，內含要驗證的來源資料表 JSON 清單。這必須是結構為 {"tableNames": ["table1", "table2"]} 的 JSON 檔案。(選用) optionalConfigurations.<sourceTable>.spannerQuery 設定用於讀取該資料表的 Spanner 查詢，例如將其限制為所選 shardIds。以 Spanner 名稱為僅限 Spanner 的資料表建立索引鍵。如需查詢規則，請參閱模組 README.md 的「Validating a subset of shards」一節。預設為空白。
• shardIds：以半形逗號分隔的邏輯分片 ID 清單 (sourcedb-to-spanner 範本使用的 ID，會顯示為 Avro 輸出中的 <table>/<shardId>/ 目錄)。設定後，系統只會驗證這些分片。如果資料表有工作階段檔案 ShardIdColumn，系統會根據該欄篩選資料；其他範圍內的資料表都必須有 spannerQuery 欄 (位於 tableConfigurationFilePath 中)，否則工作會失敗。例如 shard_001,shard_007。預設為空白。

執行範本

控制台

1. 前往 Dataflow 的「Create job from template」(依據範本建立工作) 頁面。
前往「依範本建立工作」
2. 在「Job name」(工作名稱) 欄位中，輸入專屬工作名稱。
3. 選用：如要使用區域端點，請從下拉式選單中選取值。預設區域為 us-central1。

   如需可執行 Dataflow 工作的區域清單，請參閱「Dataflow 位置」。

4. 從「Dataflow template」(Dataflow 範本) 下拉式選單中，選取「Cloud Storage Avro files to Spanner Data Validation」(將 Cloud Storage Avro 檔案匯入 Spanner 資料驗證) 範本。
5. 在提供的參數欄位中輸入參數值。
6. 按一下「Run Job」(執行工作)。

gcloud CLI

注意：如要使用 the Google Cloud CLI 執行 flex範本，您必須安裝 Google Cloud CLI 284.0.0 以上版本。

在殼層或終端機中執行範本：

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，使用範本的最新版本，該版本位於 bucket 中非依日期命名的上層資料夾：gs://dataflow-templates/latest/
  • 版本名稱，例如 2023-09-12-00_RC00，可使用特定版本的範本，該範本會以巢狀結構存放在 bucket 中相應的依日期命名的上層資料夾中：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，使用範本的最新版本，該版本位於 bucket 中非依日期命名的上層資料夾：gs://dataflow-templates/latest/
  • 版本名稱，例如 2023-09-12-00_RC00，可使用特定版本的範本，該範本會以巢狀結構存放在 bucket 中相應的依日期命名的上層資料夾中：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()));
  }
}