Modello di convalida dei dati da Avro Cloud Storage a Spanner

Beta

Questa funzionalità è soggetta ai "Termini dell'offerta pre-GA" nella sezione dei Termini di servizio generali dei Termini specifici dei servizi. Le funzionalità pre-GA sono disponibili "così come sono" e potrebbero avere un supporto limitato. Per ulteriori informazioni, consulta le descrizioni della fase di lancio.

Il modello Cloud Storage Avro files to Spanner Data Validation è una pipeline batch che legge i dati da Cloud Storage e Spanner e li confronta per convalidare la correttezza della migrazione.

Requisiti della pipeline

• La directory Cloud Storage per i file Avro deve esistere prima dell'esecuzione della pipeline. Genera questi file Avro eseguendo la pipeline SourceDB to Cloud Spanner con il parametro gcsOutputDirectory.
• Il set di dati BigQuery di destinazione per i risultati della convalida deve esistere prima dell'esecuzione della pipeline.
• Le tabelle Spanner devono esistere prima dell'esecuzione della pipeline.
• Le tabelle Spanner devono avere uno schema compatibile (direttamente o tramite la mappatura dello schema).
• Per convalidare una migrazione eseguita dalla pipeline SourceDB a Cloud Spanner, devi trasmettere valori identici per qualsiasi configurazione condivisa (ad esempio sessionFilePath, schemaOverridesFilePath e transformationJarPath).

Parametri del modello

Parametri obbligatori

• gcsInputDirectory: questa directory viene utilizzata per leggere i file AVRO dei record letti dall'origine. Ad esempio, gs://your-bucket/your-path.
• projectId: il nome del progetto Cloud Spanner.
• instanceId: l'istanza Cloud Spanner di destinazione.
• databaseId: il database Cloud Spanner di destinazione.
• bigQueryDataset: l'ID set di dati BigQuery in cui verranno archiviati i risultati della convalida. Ad esempio, validation_report_dataset.

Parametri facoltativi

• spannerHost: l'endpoint Cloud Spanner da chiamare nel modello. Ad esempio, https://batch-spanner.googleapis.com. Il valore predefinito è https://batch-spanner.googleapis.com.
• spannerPriority: la priorità della richiesta per le chiamate Cloud Spanner. Il valore deve essere uno dei seguenti: [HIGH,MEDIUM,LOW]. Il valore predefinito è HIGH.
• sessionFilePath: percorso del file di sessione in Cloud Storage che contiene le informazioni di mappatura di Spanner Migration Tool. Il valore predefinito è vuoto.
• schemaOverridesFilePath: un file che specifica le sostituzioni del nome della tabella e della colonna dall'origine a Spanner. Il valore predefinito è vuoto.
• tableOverrides: questi sono gli override del nome della tabella dall'origine a Spanner. Sono scritte nel seguente formato: [{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}]. Questo esempio mostra la mappatura della tabella Singers a Vocalists e della tabella Albums a Records. Ad esempio, [{Singers, Vocalists}, {Albums, Records}]. Il valore predefinito è vuoto.
• columnOverrides: questi sono gli override dei nomi delle colonne dall'origine a Spanner. Sono scritti nel seguente formato: [{SourceTableName1.SourceColumnName1, SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1, SourceTableName2.SpannerColumnName1}]. Tieni presente che SourceTableName deve rimanere invariato sia nell'origine sia nella coppia Spanner. Per ignorare i nomi delle tabelle, utilizza tableOverrides.L'esempio mostra la mappatura di SingerName a TalentName e di AlbumName a RecordName rispettivamente nelle tabelle Singers e Albums. Ad esempio, [{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]. Il valore predefinito è vuoto.
• runId: un identificatore univoco per l'esecuzione della convalida. Se non viene fornito, verrà utilizzato il nome del job Dataflow. Ad esempio, run_20230101_120000.
• transformationJarPath: percorso jar personalizzato in Cloud Storage che contiene la logica di trasformazione personalizzata per l'elaborazione dei record. Il valore predefinito è vuoto.
• transformationClassName: nome completo della classe con la logica di trasformazione personalizzata. È un campo obbligatorio se viene specificato transformationJarPath. Il valore predefinito è vuoto.
• transformationCustomParameters: stringa contenente eventuali parametri personalizzati da passare alla classe di trasformazione personalizzata. Il valore predefinito è vuoto.
• tables: un elenco separato da virgole delle tabelle di origine da includere nell'esecuzione della convalida. Il valore predefinito è vuoto.
• tableConfigurationFilePath: un percorso del file GCS contenente un elenco JSON delle tabelle di origine da convalidare. Deve essere un file JSON con la struttura {"tableNames": ["table1", "table2"]}. (Facoltativo) optionalConfigurations.<sourceTable>.spannerQuery imposta la query Spanner utilizzata per leggere la tabella, ad esempio per limitarla al shardIds selezionato. Chiave di una tabella solo Spanner in base al nome Spanner. Per le regole di query, consulta la sezione "Validating a subset of shards" (Convalida di un sottoinsieme di shard) del file README.md del modulo. Il valore predefinito è vuoto.
• shardIds: un elenco separato da virgole di ID shard logici (gli ID utilizzati dal modello sourcedb-to-spanner, che vengono visualizzati come directory <table>/<shardId>/ nell'output Avro). Se impostato, vengono convalidati solo questi shard. Le tabelle con un file di sessione ShardIdColumn vengono filtrate in base a questa colonna; ogni altra tabella inclusa nell'ambito ha bisogno di un spannerQuery in tableConfigurationFilePath, altrimenti il job non va a buon fine. Ad esempio, shard_001,shard_007. Il valore predefinito è vuoto.

Esegui il modello

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. La regione predefinita è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, vedi Località Dataflow.

4. Nel menu a discesa Modello Dataflow, seleziona il modello File Avro di Cloud Storage per la convalida dei dati di Spanner.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud CLI

Nota:per utilizzare Google Cloud CLI per eseguire modelli flex, devi avere la versione Google Cloud CLI 284.0.0 o successive.

Nella shell o nel terminale, esegui il modello:

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

Sostituisci quanto segue:

• JOB_NAME: un nome univoco del job a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare l'ultima versione del modello, disponibile nella cartella principale senza data nel bucket: gs://dataflow-templates/latest/
  • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella principale con data nel bucket: gs://dataflow-templates/
  Attenzione:l'ultima versione dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. I tuoi ambienti di produzione devono utilizzare i modelli conservati nella cartella principale con data più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi workflow di produzione.
• REGION_NAME: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• PATH_TO_AVRO_FILES: il percorso Cloud Storage dei file Avro.
• PROJECT_ID: l'ID progetto Spanner.
• INSTANCE_ID: l'ID dell'istanza Spanner di destinazione.
• DATABASE_ID: l'ID del database Spanner di destinazione.
• BIGQUERY_DATASET: l'ID set di dati BigQuery per i risultati della convalida.

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per saperne di più sull'API e sui relativi ambiti di autorizzazione, consulta 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"
  }
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto Google Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome univoco del job a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare l'ultima versione del modello, disponibile nella cartella principale senza data nel bucket: gs://dataflow-templates/latest/
  • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella principale con data nel bucket: gs://dataflow-templates/
  Attenzione:l'ultima versione dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. I tuoi ambienti di produzione devono utilizzare i modelli conservati nella cartella principale con data più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi workflow di produzione.
• LOCATION: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• PATH_TO_AVRO_FILES: il percorso Cloud Storage dei file Avro.
• PROJECT_ID: l'ID progetto Spanner.
• INSTANCE_ID: l'ID dell'istanza Spanner di destinazione.
• DATABASE_ID: l'ID del database Spanner di destinazione.
• BIGQUERY_DATASET: l'ID set di dati BigQuery per i risultati della convalida.

Codice sorgente del modello

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