Modèle de validation des données Cloud Storage Avro vers Spanner

Bêta

Cette fonctionnalité est soumise aux "Conditions des offres de pré-DG" de la section "Conditions générales du service" des Conditions spécifiques du service. Les fonctionnalités pré-DG sont disponibles "en l'état" et peuvent avoir une prise en charge limitée. Pour en savoir plus, consultez les descriptions des étapes de lancement.

Le modèle de validation des données de fichiers Avro Cloud Storage vers Spanner est un pipeline par lots qui lit les données de Cloud Storage et de Spanner, et les compare pour valider la correction de la migration.

Conditions requises pour ce pipeline

• Le répertoire Cloud Storage pour les fichiers Avro doit exister avant l'exécution du pipeline. Générez ces fichiers Avro en exécutant le pipeline SourceDB vers Cloud Spanner avec le paramètre gcsOutputDirectory.
• L'ensemble de données BigQuery cible pour les résultats de la validation doit exister avant l'exécution du pipeline.
• Les tables Spanner doivent exister avant l'exécution du pipeline.
• Les tables Spanner doivent avoir un schéma compatible (directement ou via le mappage de schéma).
• Pour valider une migration effectuée par le pipeline SourceDB vers Cloud Spanner, vous devez transmettre des valeurs identiques pour toutes les configurations partagées (telles que sessionFilePath, schemaOverridesFilePath et transformationJarPath).

Paramètres de modèle

Paramètres obligatoires

• gcsInputDirectory : ce répertoire est utilisé pour lire les fichiers AVRO des enregistrements lus à partir de la source. Exemple :gs://your-bucket/your-path
• projectId : nom du projet Cloud Spanner.
• instanceId : instance Cloud Spanner de destination.
• databaseId : base de données Cloud Spanner de destination.
• bigQueryDataset : ID de l'ensemble de données BigQuery dans lequel les résultats de la validation seront stockés. Exemple :validation_report_dataset

Paramètres facultatifs

• spannerHost : point de terminaison Cloud Spanner à appeler dans le modèle. Exemple :https://batch-spanner.googleapis.com La valeur par défaut est https://spanner.googleapis.com.
• spannerPriority : priorité des requêtes pour les appels Cloud Spanner. La valeur doit être l'une des suivantes : [HIGH,MEDIUM,LOW]. La valeur par défaut est HIGH.
• sessionFilePath : chemin d'accès au fichier de session dans Cloud Storage contenant les informations de mappage de l'outil de migration Spanner. La valeur par défaut est vide.
• schemaOverridesFilePath : fichier qui spécifie les remplacements de noms de tables et de colonnes de la source à Spanner. La valeur par défaut est vide.
• tableOverrides : il s'agit des remplacements de noms de tables de la source à Spanner. Ils sont écrits au format suivant : [{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}]. Cet exemple montre comment mapper la table "Singers" à "Vocalists" et la table "Albums" à "Records". Exemple :[{Singers, Vocalists}, {Albums, Records}] La valeur par défaut est vide.
• columnOverrides : il s'agit des remplacements de noms de colonnes de la source à Spanner. Elles sont écrites au format suivant : [{SourceTableName1.SourceColumnName1, SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1, SourceTableName2.SpannerColumnName1}]. Notez que SourceTableName doit rester identique dans la paire source et Spanner. Pour remplacer les noms de tables, utilisez tableOverrides.L'exemple montre comment mapper SingerName à TalentName et AlbumName à RecordName dans les tables Singers et Albums, respectivement. Exemple :[{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}] La valeur par défaut est vide.
• runId : identifiant unique de l'exécution de la validation. Si vous ne le fournissez pas, le nom du job Dataflow sera utilisé. Exemple :run_20230101_120000
• transformationJarPath : emplacement du fichier JAR personnalisé dans Cloud Storage contenant la logique de transformation personnalisée pour le traitement des enregistrements. La valeur par défaut est vide.
• transformationClassName : nom de classe complet avec une logique de transformation personnalisée. Ce champ est obligatoire si la valeur de transformationJarPath est spécifiée. La valeur par défaut est vide.
• transformationCustomParameters : chaîne contenant les paramètres personnalisés à transmettre à la classe de transformation personnalisée. La valeur par défaut est vide.
• tables : liste des tables sources à inclure dans l'exécution de la validation, séparées par une virgule. La valeur par défaut est vide.
• tableConfigurationFilePath : chemin d'accès à un fichier GCS contenant une liste JSON des tables sources à valider. Il doit s'agir d'un fichier JSON avec la structure {"tableNames": ["table1", "table2"]}. Si vous le souhaitez, optionalConfigurations.<sourceTable>.spannerQuery définit la requête Spanner utilisée pour lire cette table, par exemple pour la limiter à la shardIds sélectionnée. Identifiez une table Spanner uniquement par son nom Spanner. Consultez la section "Valider un sous-ensemble de partitions" du fichier README.md du module pour connaître les règles de requête. La valeur par défaut est vide.
• shardIds : liste d'ID de partition logique séparés par une virgule (ID utilisés par le modèle sourcedb-to-spanner, qui apparaissent sous la forme du répertoire <table>/<shardId>/ dans la sortie Avro). Si cette option est définie, seuls ces shards sont validés. Les tables avec un fichier de session ShardIdColumn sont filtrées sur cette colonne. Toutes les autres tables concernées doivent comporter un spannerQuery dans tableConfigurationFilePath, sinon le job échoue. Par exemple, shard_001,shard_007. La valeur par défaut est vide.

Exécuter le modèle

Console

1. Accédez à la page Dataflow Créer un job à partir d'un modèle.
Accéder à la page Créer un job à partir d'un modèle
2. Dans le champ Nom du job, saisissez un nom de job unique.
3. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. La région par défaut est us-central1.

   Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter un job Dataflow, consultez la page Emplacements Dataflow.

4. Dans le menu déroulant Modèle Dataflow, sélectionnez le modèle Validation des données Cloud Storage Avro vers Spanner.
5. Dans les champs fournis, saisissez vos valeurs de paramètres.
6. Cliquez sur Run Job (Exécuter la tâche).

gcloud CLI

Remarque : Pour utiliser Google Cloud CLI afin d'exécuter les modèles flex, vous devez avoir Google Cloud CLI version 284.0.0 ou ultérieure.

Dans le shell ou le terminal, exécutez le modèle :

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

Remplacez les éléments suivants :

• JOB_NAME : nom de job unique de votre choix
• VERSION : version du modèle que vous souhaitez utiliser

  Vous pouvez utiliser les valeurs suivantes :

  • latest pour utiliser la dernière version du modèle, disponible dans le dossier parent non daté du bucket gs://dataflow-templates/latest/
  • Le nom de la version, par exemple : 2023-09-12-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  Attention : La dernière version des modèles peut être mise à jour avec des modifications destructives. Vos environnements de production devraient utiliser des modèles conservés dans le dernier dossier parent daté afin d'empêcher que ces modifications destructives n'affectent vos workflows de production.
• REGION_NAME : région dans laquelle vous souhaitez déployer votre job Dataflow, par exemple us-central1
• PATH_TO_AVRO_FILES : chemin d'accès Cloud Storage aux fichiers Avro.
• PROJECT_ID : ID du projet Spanner.
• INSTANCE_ID : ID de l'instance Spanner de destination.
• DATABASE_ID : ID de la base de données Spanner de destination.
• BIGQUERY_DATASET : ID de l'ensemble de données BigQuery pour les résultats de la validation.

API

Pour exécuter le modèle à l'aide de l'API REST, envoyez une requête HTTP POST. Pour en savoir plus sur l'API, ses autorisations et leurs champs d'application, consultez la section 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"
  }
}

Remplacez les éléments suivants :

• PROJECT_ID : ID du projet Google Cloud dans lequel vous souhaitez exécuter le job Dataflow
• JOB_NAME : nom de job unique de votre choix
• VERSION : version du modèle que vous souhaitez utiliser

  Vous pouvez utiliser les valeurs suivantes :

  • latest pour utiliser la dernière version du modèle, disponible dans le dossier parent non daté du bucket gs://dataflow-templates/latest/
  • Le nom de la version, par exemple : 2023-09-12-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  Attention : La dernière version des modèles peut être mise à jour avec des modifications destructives. Vos environnements de production devraient utiliser des modèles conservés dans le dernier dossier parent daté afin d'empêcher que ces modifications destructives n'affectent vos workflows de production.
• LOCATION : région dans laquelle vous souhaitez déployer votre job Dataflow, par exemple us-central1
• PATH_TO_AVRO_FILES : chemin d'accès Cloud Storage aux fichiers Avro.
• PROJECT_ID : ID du projet Spanner.
• INSTANCE_ID : ID de l'instance Spanner de destination.
• DATABASE_ID : ID de la base de données Spanner de destination.
• BIGQUERY_DATASET : ID de l'ensemble de données BigQuery pour les résultats de la validation.

Code source du modèle

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