תבנית לאימות נתונים מ-Cloud Storage בפורמט Avro ל-Spanner

בטא

השימוש בתכונה הזו כפוף לתנאי השימוש במוצרים בגרסת טרום-GA, שמפורטים בחלק "תנאי שירות הכלליים" בתנאים הספציפיים של השירות. השימוש בתכונות בגרסת טרום-GA הוא "כפי שהן" (As is), ויכול להיות שהתמיכה בהן תהיה מוגבלת. מידע נוסף זמין בקטע תיאור שלבי ההשקה.

התבנית Cloud Storage Avro files to Spanner Data Validation (קובצי Avro ב-Cloud Storage לאימות נתונים ב-Spanner) היא צינור נתונים באצווה שקורא נתונים מ-Cloud Storage ומ-Spanner ומשווה אותם כדי לאמת את נכונות ההעברה.

הדרישות לגבי צינורות

• הספרייה ב-Cloud Storage לקובצי Avro צריכה להתקיים לפני הפעלת צינור הנתונים. כדי ליצור את קובצי ה-Avro האלה, מריצים את צינור הנתונים SourceDB to Cloud Spanner עם הפרמטר gcsOutputDirectory.
• מערך הנתונים ב-BigQuery שאליו מיועדים תוצאות האימות צריך להתקיים לפני הפעלת צינור הנתונים.
• הטבלאות ב-Spanner צריכות להתקיים לפני הפעלת צינור הנתונים.
• לכל הטבלאות ב-Spanner צריכה להיות סכימה תואמת (ישירות או באמצעות מיפוי סכימה).
• כדי לאמת את המיגרציה שבוצעה על ידי הצינור SourceDB ל-Cloud Spanner, צריך להעביר ערכים זהים לכל ההגדרות המשותפות (כמו sessionFilePath,‏ schemaOverridesFilePath ו-transformationJarPath).

פרמטרים של תבניות

פרמטרים נדרשים

• ‫gcsInputDirectory: הספרייה הזו משמשת לקריאת קובצי ה-AVRO של הרשומות שנקראו מהמקור. לדוגמה, gs://your-bucket/your-path.
• ‫projectId: השם של פרויקט Cloud Spanner.
• ‫instanceId: מכונת היעד של Cloud Spanner.
• ‫databaseId: מסד הנתונים של Cloud Spanner ביעד.
• ‫bigQueryDataset: מזהה קבוצת הנתונים ב-BigQuery שבה יאוחסנו תוצאות האימות. לדוגמה, validation_report_dataset.

פרמטרים אופציונליים

• ‫spannerHost: נקודת הקצה של Cloud Spanner שאליה מתקשרים בתבנית. לדוגמה, https://batch-spanner.googleapis.com. ערך ברירת המחדל הוא: https://batch-spanner.googleapis.com.
• ‫spannerPriority: העדיפות של הבקשה לקריאות ל-Cloud Spanner. הערך צריך להיות אחד מהערכים הבאים: [HIGH,MEDIUM,LOW]. ברירת המחדל היא HIGH.
• ‫sessionFilePath: נתיב קובץ הסשן ב-Cloud Storage שמכיל מידע על מיפוי מכלי ההעברה של Spanner. ברירת המחדל היא ריקה.
• ‫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.בדוגמה מוצג מיפוי של SingerName ל-TalentName ושל AlbumName ל-RecordName בטבלאות Singers ו-Albums בהתאמה. לדוגמה, [{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]. ברירת המחדל היא ריקה.
• ‫runId: מזהה ייחודי של הרצת האימות. אם לא תציינו שם, המערכת תשתמש בשם של משימת Dataflow. לדוגמה, run_20230101_120000.
• ‫transformationJarPath: מיקום של קובץ jar בהתאמה אישית ב-Cloud Storage שמכיל את לוגיקת ההמרה בהתאמה אישית לעיבוד רשומות. ברירת המחדל היא ריקה.
• ‫transformationClassName: שם המחלקה שמוגדר במלואו עם לוגיקת ההמרה המותאמת אישית. זהו שדה חובה אם מציינים את transformationJarPath. ברירת המחדל היא ריקה.
• ‫transformationCustomParameters: מחרוזת שמכילה פרמטרים מותאמים אישית שיועברו למחלקת ההמרה המותאמת אישית. ברירת המחדל היא ריקה.
• ‫tables: רשימה מופרדת בפסיקים של טבלאות מקור שרוצים לכלול בהרצת האימות. ברירת המחדל היא ריקה.
• ‫tableConfigurationFilePath: נתיב קובץ ב-GCS שמכיל רשימת JSON של טבלאות מקור לאימות. זה חייב להיות קובץ JSON עם המבנה {"tableNames": ["table1", "table2"]}. אופציונלי, optionalConfigurations.<sourceTable>.spannerQuery מגדיר את שאילתת Spanner שמשמשת לקריאת הטבלה הזו, למשל כדי להגביל אותה ל-shardIds שנבחר. מזינים מפתח של טבלה ב-Spanner לפי השם שלה ב-Spanner. כללי השאילתות מפורטים בקטע 'אימות של קבוצת משנה של רסיסים' בקובץ README.md של המודול. ברירת המחדל היא ריקה.
• ‫shardIds: רשימה מופרדת בפסיקים של מזהי שברים לוגיים (המזהים שמשמשים את התבנית sourcedb-to-spanner, שמופיעים כספרייה <table>/<shardId>/ בפלט של Avro). אם מגדירים את המאפיין הזה, רק הרסיסים האלה מאומתים. טבלאות עם קובץ סשן ShardIdColumn מסוננות לפי העמודה הזו. כל טבלה אחרת בהיקף צריכה להכיל spannerQuery ב-tableConfigurationFilePath, אחרת העבודה תיכשל. לדוגמה, shard_001,shard_007. ברירת המחדל היא ריקה.

הרצת התבנית

המסוף

1. עוברים לדף Create job from template (יצירת משימה מתבנית) ב-Dataflow.
כניסה לדף Create job from template
2. בשדה שם המשימה, מזינים שם ייחודי למשימה.
3. אופציונלי: בתפריט הנפתח Regional endpoint (נקודת קצה אזורית), בוחרים ערך. אזור ברירת המחדל הוא us-central1.

   רשימת האזורים שבהם אפשר להריץ משימת Dataflow מופיעה במאמר מיקומי Dataflow.

4. בתפריט הנפתח Dataflow template (תבנית Dataflow), בוחרים בתבנית Cloud Storage Avro files to Spanner Data Validation (קובצי Cloud Storage Avro לאימות נתונים ב-Spanner).
5. בשדות הפרמטרים שמופיעים, מזינים את ערכי הפרמטרים.
6. לוחצים על הפעלת העבודה.

‫CLI של gcloud

הערה: כדי להשתמש ב-Google Cloud CLI להפעלת תבניות flex צריך להשתמש בגרסה 284.0.0 או בגרסה מאוחרת יותר של Google Cloud CLI.

במעטפת או בטרמינל, מריצים את התבנית:

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: הנתיב ב-Cloud Storage לקובצי Avro.
• ‫PROJECT_ID: מזהה הפרויקט ב-Spanner.
• ‫INSTANCE_ID: מזהה מכונת היעד של Spanner.
• ‫DATABASE_ID: מזהה מסד הנתונים של Spanner ביעד.
• ‫BIGQUERY_DATASET: מזהה מערך הנתונים ב-BigQuery של תוצאות האימות.

API

כדי להריץ את התבנית באמצעות API בארכיטקטורת REST, שולחים בקשת 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: מזהה Google Cloud הפרויקט שבו רוצים להריץ את משימת Dataflow
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫VERSION: הגרסה של התבנית שרוצים להשתמש בה

  אפשר להשתמש בערכים הבאים:

  • latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא מקוננת בתיקיית נתונים מתאריך ספציפי המתאימה בדלי – gs://dataflow-templates/
  זהירות: יכול להיות שהתבניות בגרסה העדכנית יעודכנו עם שינויים שיגרמו לבעיות במערכת. כדי למנוע את ההשפעה של שינויי תוכנה שעלולים לגרום לכשל על תהליכי העבודה בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫LOCATION: האזור שבו רוצים לפרוס את משימת Dataflow, לדוגמה: us-central1
• ‫PATH_TO_AVRO_FILES: הנתיב ב-Cloud Storage לקובצי Avro.
• ‫PROJECT_ID: מזהה הפרויקט ב-Spanner.
• ‫INSTANCE_ID: מזהה מכונת היעד של Spanner.
• ‫DATABASE_ID: מזהה מסד הנתונים של Spanner ביעד.
• ‫BIGQUERY_DATASET: מזהה מערך הנתונים ב-BigQuery של תוצאות האימות.

קוד המקור של התבנית

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