Cloud Storage Avro to Spanner データ検証テンプレート

ベータ版

この機能は、サービス固有規約の「一般サービス規約」セクションの「一般提供前のサービス規約」の対象となります。pre-GA の機能は「現状のまま」で提供されるものであり、サポートが制限されることがあります。 詳細については、リリース ステージの説明をご覧ください。

Cloud Storage Avro files to Spanner Data Validation テンプレートは、Cloud Storage と Spanner からデータを読み取り、それらを比較して移行の正確性を検証するバッチ パイプラインです。

パイプラインの要件

• パイプラインの実行前に、Avro ファイルの Cloud Storage ディレクトリが存在している必要があります。これらの Avro ファイルを生成するには、gcsOutputDirectory パラメータを使用して SourceDB to Cloud Spanner パイプラインを実行します。
• 検証結果のターゲット 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: Spanner Migration Tool のマッピング情報が含まれる Cloud Storage 内のセッション ファイルのパス。デフォルトは空です。
• 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 テーブルと Albums テーブルそれぞれで、SingerName を TalentName に、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 ファイルパス。これは、{"tableNames": ["table1", "table2"]} 構造の JSON ファイルである必要があります。必要に応じて、optionalConfigurations.<sourceTable>.spannerQuery は、そのテーブルの読み取りに使用される Spanner クエリを設定します（たとえば、選択した shardIds に制限するなど）。Spanner 名で Spanner 専用テーブルをキー設定します。クエリ ルールについては、モジュールの README.md の「シャードのサブセットの検証」セクションをご覧ください。デフォルトは空です。
• shardIds: 論理シャード ID のカンマ区切りのリスト（sourcedb-to-spanner テンプレートで使用される ID。Avro 出力では <table>/<shardId>/ ディレクトリとして表示されます）。設定すると、これらのシャードのみが検証されます。セッション ファイル ShardIdColumn を含むテーブルは、その列でフィルタされます。他のすべてのスコープ内テーブルには tableConfigurationFilePath に spannerQuery が必要です。そうでない場合、ジョブは失敗します。例: shard_001,shard_007デフォルトは空です。

テンプレートを実行する

コンソール

1. Dataflow の [テンプレートからジョブを作成] ページに移動します。
[テンプレートからジョブを作成] に移動
2. [ジョブ名] フィールドに、固有のジョブ名を入力します。
3. （省略可）[リージョン エンドポイント] で、プルダウン メニューから値を選択します。デフォルトのリージョンは us-central1 です。

   Dataflow ジョブを実行できるリージョンのリストについては、Dataflow のロケーションをご覧ください。

4. [Dataflow テンプレート] プルダウン メニューから、[Cloud Storage Avro files to Spanner Data Validation] テンプレートを選択します。
5. 表示されたパラメータ フィールドに、パラメータ値を入力します。
6. [ジョブを実行] をクリックします。

gcloud CLI

注: 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: 最新バージョンのテンプレートを使用します。このテンプレートは、バケット内の日付のない親フォルダ（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()));
  }
}