תבנית Sourcedb ל-Spanner

בטא

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

התבנית SourceDB to Spanner היא צינור להעברת נתונים באצווה שמעתיק נתונים ממסד נתונים רלציוני למסד נתונים קיים ב-Spanner. הצינור הזה משתמש ב-JDBC כדי להתחבר למסד הנתונים הרלציוני. אתם יכולים להשתמש בתבנית הזו כדי להעתיק נתונים מכל מסד נתונים רלציוני עם מנהלי התקנים זמינים של JDBC אל Spanner. הכלי הזה תומך רק בקבוצה מוגבלת של סוגים של MySQL

כדי להוסיף עוד שכבת הגנה, אפשר גם להעביר מפתח Cloud KMS יחד עם פרמטרים של שם משתמש, סיסמה ומחרוזת חיבור בהצפנת Base64, שמוצפנים באמצעות מפתח Cloud KMS. פרטים נוספים על הצפנת שם המשתמש, הסיסמה ופרמטרים של מחרוזת החיבור מופיעים בנקודת הקצה Cloud KMS API encryption.

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

• מנהלי ההתקנים של JDBC למסד הנתונים הרלציוני צריכים להיות זמינים.
• טבלאות Spanner צריכות להתקיים לפני הפעלת צינור הנתונים.
• לטבלאות Spanner צריכה להיות סכימה תואמת.
• צריכה להיות גישה למסד הנתונים הרלציוני מרשת המשנה שבה פועל Dataflow.

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

פרמטר	תיאור
sourceConfigURL	מחרוזת כתובת ה-URL של חיבור JDBC. לדוגמה, jdbc:mysql://127.4.5.30:3306/my-db?autoReconnect=true&maxReconnects=10&unicode=true&characterEncoding=UTF-8 או הגדרת השארד.
instanceId	מופע היעד של Cloud Spanner.
databaseId	מסד הנתונים של Cloud Spanner שמשמש כיעד.
projectId	זהו השם של הפרויקט ב-Cloud Spanner.
outputDirectory	הספרייה הזו משמשת להעברה של רשומות שנכשלו, שדילגו עליהן או שסוננו במהלך ההעברה.
jdbcDriverJars	אופציונלי: רשימת קובצי ה-JAR של מנהלי ההתקנים, מופרדת בפסיקים. לדוגמה: gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar. ברירת המחדל היא ריק.
jdbcDriverClassName	אופציונלי: שם המחלקה של מנהל ההתקן של JDBC. לדוגמה: com.mysql.jdbc.Driver. ברירת המחדל היא: com.mysql.jdbc.Driver.
username	אופציונלי: שם המשתמש שישמש לחיבור JDBC. ברירת המחדל היא ריק.
password	אופציונלי: הסיסמה שתשמש לחיבור JDBC. ברירת המחדל היא ריק.
tables	אופציונלי: טבלאות להעברה מהמקור. ברירת המחדל היא ריק.
numPartitions	אופציונלי: מספר המחיצות. הערך הזה, יחד עם הגבול התחתון והגבול העליון, יוצר מחיצות של צעדים לביטויים שנוצרים של משפט WHERE שמשמשים לפיצול שווה של עמודת המחיצה. אם הקלט קטן מ-1, המספר מוגדר כ-1. ברירת המחדל: 0.
spannerHost	אופציונלי: נקודת הקצה של Cloud Spanner שאליה מתקשרים בתבנית. לדוגמה: https://batch-spanner.googleapis.com. ברירת המחדל היא: https://batch-spanner.googleapis.com.
maxConnections	אופציונלי: מגדיר את מאגר חיבורי ה-JDBC בכל עובד עם מספר מקסימלי של חיבורים. כדי לא להגביל את מספר המשתמשים, צריך להזין מספר שלילי. לדוגמה: -1. ברירת המחדל: 0.
sessionFilePath	אופציונלי: נתיב הסשן ב-Cloud Storage שמכיל פרטי מיפוי מכלי ההעברה של Spanner. ברירת המחדל היא ריק.
transformationJarPath	אופציונלי: מיקום מותאם אישית של קובץ jar ב-Cloud Storage שמכיל את הלוגיקה של טרנספורמציה מותאמת אישית לעיבוד רשומות. ברירת המחדל היא ריק.
transformationClassName	אופציונלי: שם מחלקה שמוגדר במלואו ומכיל את הלוגיקה של השינוי המותאם אישית. זהו שדה חובה אם מציינים את transformationJarPath. ברירת המחדל היא ריק.
transformationCustomParameters	אופציונלי: מחרוזת שמכילה פרמטרים מותאמים אישית שיועברו למחלקת ההמרה המותאמת אישית. ברירת המחדל היא ריק.
disabledAlgorithms	אופציונלי: אלגוריתמים להשבתה, מופרדים בפסיקים. אם הערך הוא none, אף אלגוריתם לא מושבת. חשוב להשתמש בפרמטר הזה בזהירות, כי יכול להיות שיש אלגוריתמים שמושבתים כברירת מחדל בגלל פגיעויות או בעיות בביצועים. לדוגמה: SSLv3, RC4.
extraFilesToStage	אופציונלי: נתיבים של Cloud Storage או סודות של Secret Manager שמופרדים בפסיקים, לקבצים שיועברו לשלב ההכנה בעובד. הקבצים האלה נשמרים בספרייה ‎ /extra_files בכל עובד. לדוגמה: gs://<BUCKET>/file.txt,projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<VERSION_ID>.

הפעלת התבנית

המסוף

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

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

4. בתפריט הנפתח Dataflow template (תבנית של העברת נתונים), בוחרים באפשרות the Sourcedb to Spanner template.
5. בשדות הפרמטרים שמופיעים, מזינים את ערכי הפרמטרים.
6. לוחצים על הפעלת העבודה.

‫CLI של gcloud

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

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

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Sourcedb_to_Spanner_Flex \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --parameters \
       sourceConfigURL=SOURCE_CONFIG_URL,\
       instanceId=INSTANCE_ID,\
       databaseId=DATABASE_ID,\
       projectId=PROJECT_ID,\
       outputDirectory=OUTPUT_DIRECTORY,\

מחליפים את מה שכתוב בשדות הבאים:

• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫VERSION: הגרסה של התבנית שבה רוצים להשתמש

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

  • ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שגרסת התבניות העדכנית תתעדכן עם שינויים שעלולים לשבור את התאימות לאחור. כדי למנוע מהשינויים האלה לשבש את תהליכי העבודה בסביבת הייצור, צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר בסביבות הייצור.
• ‫REGION_NAME: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫SOURCE_CONFIG_URL: כתובת ה-URL להתחברות למארח של מסד הנתונים של המקור. הערך יכול להיות 1. כתובת ה-URL של חיבור ה-JDBC – חייבת להכיל את המארח, היציאה ושם מסד הנתונים של המקור, ויכולה להכיל מאפיינים כמו autoReconnect,‏ maxReconnects וכו'. פורמט: ‎`jdbc:mysql://{host}:{port}/{dbName}?{parameters}`‎2. הנתיב להגדרת השארד
• ‫INSTANCE_ID: מזהה המופע של Cloud Spanner.
• ‫DATABASE_ID: מזהה מסד הנתונים ב-Cloud Spanner.
• ‫PROJECT_ID: מזהה הפרויקט ב-Cloud Spanner.
• ‫OUTPUT_DIRECTORY: ספריית הפלט לאירועים שנכשלו, דולגו או סוננו

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": {
       "sourceConfigURL": "SOURCE_CONFIG_URL",
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "projectId": "PROJECT_ID",
       "outputDirectory": "OUTPUT_DIRECTORY",
     },
     "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Sourcedb_to_Spanner_Flex",
     "environment": { "maxWorkers": "10" }
  }
}

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: מזהה הפרויקט שבו רוצים להריץ את משימת Dataflow Google Cloud
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫VERSION: הגרסה של התבנית שבה רוצים להשתמש

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

  • ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שגרסת התבניות העדכנית תתעדכן עם שינויים שעלולים לשבור את התאימות לאחור. כדי למנוע מהשינויים האלה לשבש את תהליכי העבודה בסביבת הייצור, צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר בסביבות הייצור.
• ‫LOCATION: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫SOURCE_CONFIG_URL: כתובת ה-URL להתחברות למארח של מסד הנתונים של המקור. הערך יכול להיות 1. כתובת ה-URL של חיבור ה-JDBC – חייבת להכיל את המארח, היציאה ושם מסד הנתונים של המקור, ויכולה להכיל מאפיינים כמו autoReconnect,‏ maxReconnects וכו'. פורמט: ‎`jdbc:mysql://{host}:{port}/{dbName}?{parameters}`‎2. הנתיב להגדרת השארד
• ‫INSTANCE_ID: מזהה המופע של Cloud Spanner.
• ‫DATABASE_ID: מזהה מסד הנתונים ב-Cloud Spanner.
• ‫PROJECT_ID: מזהה הפרויקט ב-Cloud Spanner.
• ‫OUTPUT_DIRECTORY: ספריית הפלט לאירועים שנכשלו, דולגו או סוננו

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

Java

/*
 * Copyright (C) 2024 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 com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.CommonTemplateJvmInitializer;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.options.SourceDbToSpannerOptions;
import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard;
import com.google.cloud.teleport.v2.spanner.migrations.utils.DataflowWorkerMachineTypeUtils;
import com.google.cloud.teleport.v2.spanner.migrations.utils.SecretManagerAccessorImpl;
import com.google.cloud.teleport.v2.spanner.migrations.utils.ShardFileReader;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.util.List;
import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions;
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.commons.lang3.StringUtils;

/**
 * A template that copies data from a relational database using JDBC to an existing Spanner
 * database.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/sourcedb-to-spanner/README_Sourcedb_to_Spanner_Flex.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Sourcedb_to_Spanner_Flex",
    category = TemplateCategory.BATCH,
    displayName = "Sourcedb to Spanner",
    description = {
      "The SourceDB to Spanner template is a batch pipeline that copies data from a relational"
          + " database into an existing Spanner database. This pipeline uses JDBC to connect to"
          + " the relational database. You can use this template to copy data from any relational"
          + " database with available JDBC drivers into Spanner. This currently only supports a limited set of types of MySQL",
    },
    optionsClass = SourceDbToSpannerOptions.class,
    flexContainerName = "source-db-to-spanner",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/sourcedb-to-spanner",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The JDBC drivers for the relational database must be available.",
      "The Spanner tables must exist before pipeline execution.",
      "The Spanner tables must have a compatible schema.",
      "The relational database must be accessible from the subnet where Dataflow runs."
    })
public class SourceDbToSpanner {

  /**
   * Main entry point for executing the pipeline. This will run the pipeline asynchronously. If
   * blocking execution is required, use the {@link SourceDbToSpanner#run} method to start the
   * pipeline and invoke {@code result.waitUntilFinish()} on the {@link PipelineResult}.
   *
   * @param args The command-line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    SourceDbToSpannerOptions options = getSourceDbToSpannerOptions(args);
    run(options);
  }

  @VisibleForTesting
  protected static SourceDbToSpannerOptions getSourceDbToSpannerOptions(String[] args) {
    // Parse the user options passed from the command-line
    SourceDbToSpannerOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(SourceDbToSpannerOptions.class);
    // Stage SSL certificates to extraFiles if required as per the pipeline options.
    // Ref https://cloud.google.com/dataflow/docs/guides/templates/ssl-certificates
    new CommonTemplateJvmInitializer().beforeProcessing(options);
    return options;
  }

  /**
   * Create the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  @VisibleForTesting
  static PipelineResult run(SourceDbToSpannerOptions options) {
    // TODO - Validate if options are as expected
    Pipeline pipeline = Pipeline.create(options);
    String workerMachineType =
        pipeline.getOptions().as(DataflowPipelineWorkerPoolOptions.class).getWorkerMachineType();
    DataflowWorkerMachineTypeUtils.validateMachineSpecs(workerMachineType, 4);

    SpannerConfig spannerConfig = createSpannerConfig(options);

    // Decide type and source of migration
    // TODO(vardhanvthigle): Move this within pipelineController.
    switch (options.getSourceDbDialect()) {
      case SourceDbToSpannerOptions.CASSANDRA_SOURCE_DIALECT:
        Preconditions.checkArgument(
            StringUtils.isNotEmpty(options.getSourceConfigURL()),
            "Cassandra Dialect needs sourceConfigURL to be set.");
        return PipelineController.executeCassandraMigration(options, pipeline, spannerConfig);
      case SourceDbToSpannerOptions.ASTRA_DB_SOURCE_DIALECT:
        return PipelineController.executeCassandraMigration(options, pipeline, spannerConfig);

      default:
        /* Implementation detail, not having a default leads to failure in compile time checks enforced here */
        /* Making jdbc as default case which includes MYSQL and PG. */
        Preconditions.checkArgument(
            StringUtils.isNotEmpty(options.getSourceConfigURL()),
            "JDBC based source needs sourceConfigURL to be set.");
        return executeJdbcMigration(options, pipeline, spannerConfig);
    }
  }

  // TODO(vardhanvthigle): Move this within pipelineController.
  private static PipelineResult executeJdbcMigration(
      SourceDbToSpannerOptions options, Pipeline pipeline, SpannerConfig spannerConfig) {
    if (options.getSourceConfigURL().startsWith("gs://")) {
      List<Shard> shards =
          new ShardFileReader(new SecretManagerAccessorImpl())
              .readForwardMigrationShardingConfig(options.getSourceConfigURL());
      return PipelineController.executeJdbcShardedMigration(
          options, pipeline, shards, spannerConfig);
    } else {
      return PipelineController.executeJdbcSingleInstanceMigration(
          options, pipeline, spannerConfig);
    }
  }

  @VisibleForTesting
  static SpannerConfig createSpannerConfig(SourceDbToSpannerOptions options) {
    SpannerConfig spannerConfig =
        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()));
    if (options.getMaxCommitDelay() >= 0) {
      spannerConfig = spannerConfig.withMaxCommitDelay(options.getMaxCommitDelay());
    }
    return spannerConfig;
  }
}