Template Validasi Data Cloud Storage Avro ke Spanner

Beta

Fitur ini tunduk pada "Persyaratan Penawaran Pra-GA" di bagian Persyaratan Layanan Umum dari Persyaratan Khusus Layanan. Fitur pra-GA tersedia "sebagaimana adanya" dan mungkin memiliki dukungan terbatas. Untuk mengetahui informasi selengkapnya, lihat deskripsi tahap peluncuran.

Template Validasi Data file Avro Cloud Storage ke Spanner adalah pipeline batch yang membaca data dari Cloud Storage dan Spanner, lalu membandingkannya untuk memvalidasi kebenaran migrasi.

Persyaratan pipeline

• Direktori Cloud Storage untuk file Avro harus ada sebelum eksekusi pipeline. Buat file Avro ini dengan menjalankan pipeline SourceDB to Cloud Spanner dengan parameter gcsOutputDirectory.
• Set data BigQuery target untuk hasil validasi harus ada sebelum eksekusi pipeline.
• Tabel Spanner harus ada sebelum eksekusi pipeline.
• Tabel Spanner harus memiliki skema yang kompatibel (baik secara langsung maupun melalui pemetaan skema).
• Untuk memvalidasi migrasi yang dilakukan oleh pipeline SourceDB ke Cloud Spanner, Anda harus meneruskan nilai yang identik untuk konfigurasi bersama (seperti sessionFilePath, schemaOverridesFilePath, dan transformationJarPath).

Parameter template

Parameter yang diperlukan

• gcsInputDirectory: Direktori ini digunakan untuk membaca file AVRO dari data yang dibaca dari sumber. Contoh, gs://your-bucket/your-path.
• projectId: Ini adalah nama project Cloud Spanner.
• instanceId: Instance Cloud Spanner tujuan.
• databaseId: Database Cloud Spanner tujuan.
• bigQueryDataset: ID set data BigQuery tempat hasil validasi akan disimpan. Contoh, validation_report_dataset.

Parameter opsional

• spannerHost: Endpoint Cloud Spanner yang akan dipanggil dalam template. Contoh, https://batch-spanner.googleapis.com. Secara default: https://batch-spanner.googleapis.com.
• spannerPriority: Prioritas permintaan untuk panggilan Cloud Spanner. Nilai harus berupa salah satu dari: [HIGH,MEDIUM,LOW]. Defaultnya adalah HIGH.
• sessionFilePath: Jalur file sesi di Cloud Storage yang berisi informasi pemetaan dari Spanner Migration Tool. Nilai defaultnya adalah kosong.
• schemaOverridesFilePath: File yang menentukan penggantian nama tabel dan kolom dari sumber ke Spanner. Nilai defaultnya adalah kosong.
• tableOverrides: Ini adalah penggantian nama tabel dari sumber ke Spanner. Tabel ini ditulis dalam format berikut: [{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}] Contoh ini menunjukkan pemetaan tabel Singers ke Vocalists dan tabel Albums ke Records. Contoh, [{Singers, Vocalists}, {Albums, Records}]. Nilai defaultnya adalah kosong.
• columnOverrides: Ini adalah penggantian nama kolom dari sumber ke spanner. Kolom tersebut ditulis dalam format berikut: [{SourceTableName1.SourceColumnName1, SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1, SourceTableName2.SpannerColumnName1}]. Perhatikan bahwa SourceTableName harus tetap sama di pasangan sumber dan spanner. Untuk mengganti nama tabel, gunakan tableOverrides.Contoh ini menunjukkan pemetaan SingerName ke TalentName dan AlbumName ke RecordName dalam tabel Singers dan Albums. Contoh, [{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]. Nilai defaultnya adalah kosong.
• runId: ID unik untuk proses validasi. Jika tidak disediakan, Nama Tugas Dataflow akan digunakan. Contoh, run_20230101_120000.
• transformationJarPath: Lokasi jar kustom di Cloud Storage yang berisi logika transformasi kustom untuk memproses data. Nilai defaultnya adalah kosong.
• transformationClassName: Nama class yang sepenuhnya memenuhi syarat yang memiliki logika transformasi kustom. Kolom ini wajib diisi jika transformationJarPath ditentukan. Nilai defaultnya adalah kosong.
• transformationCustomParameters: String yang berisi parameter kustom yang akan diteruskan ke class transformasi kustom. Nilai defaultnya adalah kosong.
• tables: Daftar tabel sumber yang dipisahkan koma untuk disertakan dalam proses validasi. Nilai defaultnya adalah kosong.
• tableConfigurationFilePath: Jalur file GCS yang berisi daftar JSON tabel sumber untuk divalidasi. File ini harus berupa file JSON dengan struktur {"tableNames": ["table1", "table2"]}. Secara opsional, optionalConfigurations.<sourceTable>.spannerQuery menetapkan kueri Spanner yang digunakan untuk membaca tabel tersebut, misalnya untuk membatasinya ke shardIds yang dipilih. Beri kunci pada tabel khusus Spanner berdasarkan nama Spanner-nya. Lihat bagian 'Memvalidasi subset shard' di README.md modul untuk mengetahui aturan kueri. Nilai defaultnya adalah kosong.
• shardIds: Daftar ID shard logis yang dipisahkan koma (ID yang digunakan oleh template sourcedb-to-spanner, yang muncul sebagai direktori <table>/<shardId>/ dalam output Avro). Jika disetel, hanya shard tersebut yang divalidasi. Tabel dengan file sesi ShardIdColumn difilter pada kolom tersebut; setiap tabel dalam cakupan lainnya memerlukan spannerQuery di tableConfigurationFilePath, jika tidak, tugas akan gagal. Misalnya, shard_001,shard_007. Nilai defaultnya adalah kosong.

Menjalankan template

Konsol

1. Buka halaman Dataflow Membuat tugas dari template.
Buka Membuat tugas dari template
2. Di kolom Nama tugas, masukkan nama tugas yang unik.
3. Opsional: Untuk Endpoint regional, pilih nilai dari menu drop-down. Region default-nya adalah us-central1.

   Untuk mengetahui daftar region tempat Anda dapat menjalankan tugas Dataflow, lihat Lokasi Dataflow.

4. Dari menu drop-down Dataflow template, pilih template Cloud Storage Avro files to Spanner Data Validation.
5. Di kolom parameter yang disediakan, masukkan nilai parameter Anda.
6. Klik Jalankan tugas.

gcloud CLI

Catatan: Untuk menggunakan Google Cloud CLI guna menjalankan template flex, Anda harus memiliki Google Cloud CLI versi 284.0.0 atau yang lebih baru.

Di shell atau terminal Anda, jalankan template:

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

Ganti kode berikut:

• JOB_NAME: nama tugas unik pilihan Anda
• VERSION: versi template yang ingin Anda gunakan

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan template versi terbaru, yang tersedia di folder induk tanpa tanggal di bucket— gs://dataflow-templates/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan template versi tertentu, yang dapat ditemukan bertingkat di masing-masing folder induk yang diberi tanggal dalam bucket— gs://dataflow-templates/
  Perhatian: Template versi terbaru dapat diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk berisi tanggal terbaru untuk mencegah perubahan yang dapat menyebabkan gangguan ini memengaruhi alur kerja produksi Anda.
• REGION_NAME: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
• PATH_TO_AVRO_FILES: jalur Cloud Storage ke file Avro.
• PROJECT_ID: project ID Spanner.
• INSTANCE_ID: ID instance Spanner tujuan.
• DATABASE_ID: ID database Spanner tujuan.
• BIGQUERY_DATASET: ID set data BigQuery untuk hasil validasi.

API

Untuk menjalankan template menggunakan REST API, kirim permintaan POST HTTP. Untuk mengetahui informasi selengkapnya tentang API dan cakupan otorisasinya, lihat 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"
  }
}

Ganti kode berikut:

• PROJECT_ID: Google Cloud Project ID tempat Anda ingin menjalankan tugas Dataflow
• JOB_NAME: nama tugas unik pilihan Anda
• VERSION: versi template yang ingin Anda gunakan

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan template versi terbaru, yang tersedia di folder induk tanpa tanggal di bucket— gs://dataflow-templates/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan template versi tertentu, yang dapat ditemukan bertingkat di masing-masing folder induk yang diberi tanggal dalam bucket— gs://dataflow-templates/
  Perhatian: Template versi terbaru dapat diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk berisi tanggal terbaru untuk mencegah perubahan yang dapat menyebabkan gangguan ini memengaruhi alur kerja produksi Anda.
• LOCATION: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
• PATH_TO_AVRO_FILES: jalur Cloud Storage ke file Avro.
• PROJECT_ID: project ID Spanner.
• INSTANCE_ID: ID instance Spanner tujuan.
• DATABASE_ID: ID database Spanner tujuan.
• BIGQUERY_DATASET: ID set data BigQuery untuk hasil validasi.

Kode sumber template

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