Vorlage „Cloud Storage Avro für Spanner-Datenvalidierung“

Beta

Diese Funktion unterliegt den „Bedingungen für Pre-GA-Angebote“ im Abschnitt „Allgemeine Dienstbedingungen“ der dienstspezifischen Nutzungsbedingungen. Pre-GA‑Funktionen stehen in der vorliegenden Form zur Verfügung und bieten möglicherweise nur eingeschränkten Support. Weitere Informationen finden Sie unter den Beschreibungen der Startphase.

Die Vorlage „Cloud Storage Avro für Spanner-Datenvalidierung“ ist eine Batchpipeline, die Daten aus Cloud Storage und Spanner liest und vergleicht, um die Richtigkeit der Migration zu validieren.

Pipelineanforderungen

• Das Cloud Storage-Verzeichnis für Avro-Dateien muss vor der Ausführung der Pipeline vorhanden sein. Generieren Sie diese Avro-Dateien, indem Sie die Pipeline SourceDB zu Cloud Spanner mit dem Parameter gcsOutputDirectory ausführen.
• Das BigQuery-Ziel-Dataset für die Validierungsergebnisse muss vor der Pipelineausführung vorhanden sein.
• Die Spanner-Tabellen müssen vor der Pipelineausführung vorhanden sein.
• Die Spanner-Tabellen müssen ein kompatibles Schema haben (entweder direkt oder über die Schemazuordnung).
• Wenn Sie eine Migration validieren möchten, die von der SourceDB to Cloud Spanner-Pipeline ausgeführt wurde, müssen Sie identische Werte für alle freigegebenen Konfigurationen (z. B. sessionFilePath, schemaOverridesFilePath und transformationJarPath) übergeben.

Vorlagenparameter

Erforderliche Parameter

• gcsInputDirectory: Dieses Verzeichnis wird verwendet, um die AVRO-Dateien der Datensätze zu lesen, die aus der Quelle gelesen werden. Beispiel: gs://your-bucket/your-path.
• projectId: Dies ist der Name des Cloud Spanner-Projekts.
• instanceId: Die Cloud Spanner-Zielinstanz.
• databaseId: Die Ziel-Cloud Spanner-Datenbank.
• bigQueryDataset: Die BigQuery-Dataset-ID, in der die Validierungsergebnisse gespeichert werden. Beispiel: validation_report_dataset.

Optionale Parameter

• spannerHost: Der Cloud Spanner-Endpunkt, der in der Vorlage aufgerufen werden soll. Beispiel: https://batch-spanner.googleapis.com. Die Standardeinstellung ist https://batch-spanner.googleapis.com.
• spannerPriority: Die Anfragepriorität für Cloud Spanner-Aufrufe. Der Wert muss einer der folgenden sein: [HIGH,MEDIUM,LOW]. Der Standardwert ist HIGH.
• sessionFilePath: Pfad der Sitzungsdatei in Cloud Storage, die Zuordnungsinformationen aus dem Spanner Migration Tool enthält. Die Standardeinstellung ist leer.
• schemaOverridesFilePath: Eine Datei, in der die Überschreibungen für Tabellen- und Spaltennamen von der Quelle zu Spanner angegeben werden. Die Standardeinstellung ist leer.
• tableOverrides: Dies sind die Überschreibungen des Tabellennamens von der Quelle zu Spanner. Sie werden im folgenden Format geschrieben: [{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}]. In diesem Beispiel wird die Tabelle „Singers“ der Tabelle „Vocalists“ und die Tabelle „Albums“ der Tabelle „Records“ zugeordnet. Beispiel: [{Singers, Vocalists}, {Albums, Records}]. Die Standardeinstellung ist leer.
• columnOverrides: Dies sind die Überschreibungen für Spaltennamen von der Quelle zu Spanner. Sie werden im folgenden Format geschrieben: [{SourceTableName1.SourceColumnName1, SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1, SourceTableName2.SpannerColumnName1}]. Der SourceTableName sollte sowohl in der Quelle als auch im Spanner-Paar gleich bleiben. Verwenden Sie tableOverrides, um Tabellennamen zu überschreiben.Im Beispiel wird SingerName in der Tabelle „Singers“ (Sänger) in TalentName und AlbumName in der Tabelle „Albums“ (Alben) in RecordName umgewandelt. Beispiel: [{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]. Die Standardeinstellung ist leer.
• runId: Eine eindeutige Kennung für den Validierungslauf. Wenn nicht angegeben, wird der Dataflow-Jobname verwendet. Beispiel: run_20230101_120000.
• transformationJarPath: Speicherort der benutzerdefinierten JAR-Datei in Cloud Storage, die die benutzerdefinierte Transformationslogik für die Verarbeitung von Datensätzen enthält. Die Standardeinstellung ist leer.
• transformationClassName: Voll qualifizierter Klassenname mit der benutzerdefinierten Transformationslogik. Es ist ein Pflichtfeld, wenn „transformationJarPath“ angegeben ist. Die Standardeinstellung ist leer.
• transformationCustomParameters: String mit allen benutzerdefinierten Parametern, die an die benutzerdefinierte Transformationsklasse übergeben werden sollen. Die Standardeinstellung ist leer.
• tables: Eine durch Kommas getrennte Liste von Quelltabellen, die in den Validierungslauf aufgenommen werden sollen. Die Standardeinstellung ist leer.
• tableConfigurationFilePath: Ein GCS-Dateipfad, der eine JSON-Liste der zu validierenden Quelltabelle enthält. Dabei muss es sich um eine JSON-Datei mit der Struktur {"tableNames": ["table1", "table2"]} handeln. Die Standardeinstellung ist leer.

Führen Sie die Vorlage aus.

Console

1. Rufen Sie die Dataflow-Seite Job aus Vorlage erstellen auf.
Zur Seite "Job aus Vorlage erstellen“
2. Geben Sie im Feld Jobname einen eindeutigen Jobnamen ein.
3. Optional: Wählen Sie für Regionaler Endpunkt einen Wert aus dem Drop-down-Menü aus. Die Standardregion ist us-central1.

   Eine Liste der Regionen, in denen Sie einen Dataflow-Job ausführen können, finden Sie unter Dataflow-Standorte.

4. Wählen Sie im Drop-down-Menü Dataflow-Vorlage die Vorlage Cloud Storage Avro files to Spanner Data Validation (Cloud Storage-Avro-Dateien für die Spanner-Datenvalidierung) aus.
5. Geben Sie Ihre Parameterwerte in die Parameterfelder ein.
6. Klicken Sie auf Job ausführen.

gcloud-CLI

Hinweis: Um flex-Vorlagen mit dem Google Cloud CLI auszuführen, benötigen Sie das Google Cloud CLI in der Version 284.0.0 oder höher.

Führen Sie die Vorlage in der Shell oder im Terminal aus:

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

Ersetzen Sie Folgendes:

• JOB_NAME: Ein eindeutiger Jobname Ihrer Wahl
• VERSION: Die Version der Vorlage, die Sie verwenden möchten

  Sie können die folgenden Werte verwenden:

  • latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates/latest/.
  • Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates/.
  Achtung: Die neueste Version der Vorlagen wird möglicherweise mit funktionsgefährdenden Änderungen aktualisiert. In Ihren Produktionsumgebungen sollten Vorlagen verwendet werden, die sich im aktuellsten datierten übergeordneten Ordner befinden, um zu verhindern, dass diese funktionsgefährdenden Änderungen Ihre Produktionsworkflows beeinträchtigen.
• REGION_NAME: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
• PATH_TO_AVRO_FILES: Der Cloud Storage-Pfad zu den Avro-Dateien.
• PROJECT_ID: die Spanner-Projekt-ID.
• INSTANCE_ID: die ID der Ziel-Spanner-Instanz.
• DATABASE_ID: die ID der Ziel-Spanner-Datenbank.
• BIGQUERY_DATASET: Die BigQuery-Dataset-ID für die Validierungsergebnisse.

API

Senden Sie eine HTTP-POST-Anfrage, um die Vorlage mithilfe der REST API auszuführen. Weitere Informationen zur API und ihren Autorisierungsbereichen finden Sie unter 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"
  }
}

Ersetzen Sie Folgendes:

• PROJECT_ID: Die Google Cloud Projekt-ID, in der Sie den Dataflow-Job ausführen möchten
• JOB_NAME: Ein eindeutiger Jobname Ihrer Wahl
• VERSION: Die Version der Vorlage, die Sie verwenden möchten

  Sie können die folgenden Werte verwenden:

  • latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates/latest/.
  • Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates/.
  Achtung: Die neueste Version der Vorlagen wird möglicherweise mit funktionsgefährdenden Änderungen aktualisiert. In Ihren Produktionsumgebungen sollten Vorlagen verwendet werden, die sich im aktuellsten datierten übergeordneten Ordner befinden, um zu verhindern, dass diese funktionsgefährdenden Änderungen Ihre Produktionsworkflows beeinträchtigen.
• LOCATION: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
• PATH_TO_AVRO_FILES: Der Cloud Storage-Pfad zu den Avro-Dateien.
• PROJECT_ID: die Spanner-Projekt-ID.
• INSTANCE_ID: die ID der Ziel-Spanner-Instanz.
• DATABASE_ID: die ID der Ziel-Spanner-Datenbank.
• BIGQUERY_DATASET: Die BigQuery-Dataset-ID für die Validierungsergebnisse.

Quellcode der Vorlage

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