תבנית Sourcedb ל-Spanner

בטא

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

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

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

הערה: אם אתם מתכננים לאמת את המיגרציה הזו באמצעות התבנית Cloud Storage Avro files to Spanner Data Validation, אתם צריכים לאכלס את הפרמטר gcsOutputDirectory כדי ליצור את קובצי ה-Avro הנדרשים.

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

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

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

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

• ‫sourceConfigURL: כתובת ה-URL של קובץ ההגדרה של חיבור המקור. פורמט הקובץ תלוי בסוג המקור. ב-Astra, הוא יצביע על קובץ הגדרות חיבור ל-Astra (דוגמה). ב-JDBC, הוא יפנה לקובץ הגדרות של שרדינג JDBC (דוגמה). ב-Cassandra, הוא יצביע על קובץ הגדרות של מנהל התקן של Cassandra (דוגמה). חובה לכלול את הפרמטר הזה. לדוגמה, gs://your-bucket/source-config.json. ברירת המחדל היא ריקה.
• ‫instanceId: מכונת היעד של Cloud Spanner.
• ‫databaseId: מסד הנתונים של היעד ב-Cloud Spanner.
• ‫projectId: השם של פרויקט Cloud Spanner.
• ‫outputDirectory: בספרייה הזו נשמרות רשומות שנכשלו, שדילגו עליהן או שסוננו במהלך המיגרציה.

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

• ‫sourceDbDialect: הערכים האפשריים הם CASSANDRA, ‏ MYSQL, ‏ POSTGRESQL, ‏ ORACLE ו-SQLSERVER. ברירת המחדל היא: MYSQL.
• ‫jdbcDriverJars: רשימת קובצי ה-JAR של מנהלי ההתקנים, מופרדת בפסיקים. לדוגמה, gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar. ברירת המחדל היא ריקה.
• ‫jdbcDriverClassName: שם המחלקה של מנהל ההתקן של JDBC. לדוגמה, com.mysql.jdbc.Driver. ברירת המחדל היא: com.mysql.jdbc.Driver.
• ‫tables: טבלאות להעברה מהמקור. ברירת המחדל היא ריקה.
• ‫numPartitions: מספר המחיצות. הערך הזה, יחד עם הגבול התחתון והגבול העליון, יוצר מחיצות של צעדים לביטויי WHERE שנוצרים ומשמשים לפיצול שווה של עמודת המחיצה. אם הקלט קטן מ-1, המספר מוגדר כ-1. ברירת המחדל היא 0.
• ‫fetchSize: מספר השורות לשליפה לכל קריאת דף עבור מקור JDBC. אם לא מגדירים את הערך הזה, הוא מחושב אוטומטית לפי סוג המכונה של העובד והגודל המשוער של השורה. אם אי אפשר לחשב את הערך (לדוגמה, אם לא מציינים את סוג המכונה של העובד), ברירת המחדל היא 50,000 שורות. אם דיאלקט המקור הוא MySQL, יש לעיין בהערה שלמטה. בסופו של דבר, זה מתורגם לקריאה של Statement.setFetchSize בשכבת ה-JDBC. צריך להשתמש באפשרות הזו רק אם ערך ברירת המחדל גורם לשגיאות זיכרון.הערה לגבי מקור MySql: המחבר Mysql מתעלם מ-FetchSize אלא אם useCursorFetch=true הוא גם חלק ממאפייני החיבור. בניב MySql, צינור הנתונים יוסיף את useCursorFetch=true למאפייני החיבור כברירת מחדל, אלא אם הפרמטר fetchSize מוגדר במפורש ל-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: מחרוזת שמכילה פרמטרים מותאמים אישית שיועברו למחלקת ההמרה המותאמת אישית. ברירת המחדל היא ריקה.
• ‫insertOnlyModeForSpannerMutations: כברירת מחדל, צינור עיבוד הנתונים משתמש בפעולות Upsert כדי לכתוב שורות ל-Spanner. כלומר, השורות הקיימות יוחלפו. אם מפעילים את מצב InsertOnly, המערכת תשתמש בפעולות הוספה במקום בפעולות עדכון והוספה, והשורות הקיימות לא יוחלפו.
• ‫batchSizeForSpannerMutations: גודל המקבץ בבייטים לשינויים ב-Spanner. אם הערך שמוגדר קטן מ-0, נעשה שימוש בערך ברירת המחדל של SpannerIO ב-Apache Beam, שהוא 1MB. כדי להשבית את האפשרות של צירוף מוטציות, צריך להגדיר את הערך כ-0 או כ-10.
• ‫spannerPriority: העדיפות של הבקשה לקריאות ל-Cloud Spanner. הערך צריך להיות אחד מהערכים הבאים: [HIGH,MEDIUM,LOW]. ברירת המחדל היא HIGH.
• ‫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}]. ברירת המחדל היא ריקה.
• ‫schemaOverridesFilePath: קובץ שמציין את הטבלה ואת החלפת שמות העמודות ממקור ל-Spanner. ברירת המחדל היא ריקה.
• ‫uniformizationStageCountHint: רמז למספר שלבי האחידות. נכון לעכשיו, הרלוונטיות היא רק למקורות מבוססי-JDBC כמו MySQL או PostgreSQL. כדי להשבית את האחידות, משאירים את הערך 0 או את ברירת המחדל. מגדירים את הערך -1 כדי לקבל מספר שלבים ששווה ל-log(numPartition). אם מרחב המפתחות הראשיים במקור מפוזר באופן אחיד (לדוגמה, מפתח עם הגדלה אוטומטית עם חורים דלילים), מומלץ להשאיר את ההגדרה הזו מושבתת. אם מרחב המפתחות לא אחיד, יכול להיות שתיתקלו במכונה וירטואלית מפגרת בהרצת זרימת הנתונים. במקרה כזה, אפשר להגדיר את הערך שלו כ-1 כדי להפעיל את האחידות. אם תגדירו אותו ידנית לערכים שונים מ-0 או מ-1-, תוכלו לכוונן את האיזון בין התקורה שנוספת בשלבי האחידות לבין שיפור הביצועים כתוצאה מחלוקת עבודה טובה יותר.
• ‫failureInjectionParameter: פרמטר להזרקת כשלים. משמש רק לבדיקה. ברירת המחדל היא ריקה.
• ‫maxCommitDelay: זמן העיכוב המקסימלי של ביצוע פעולת commit כדי לבצע אופטימיזציה של קצב העברת הנתונים בכתיבה ב-Spanner. אפשר לעיין במאמר https://cloud.google.com/spanner/docs/throughput-optimized-writes.Set כדי להגדיר את הערך ‎-1 ולאפשר ל-Spanner לבחור את ברירת המחדל. הערך צריך להיות חיובי כדי לבטל את ברירת המחדל ולהשיג את האיזון הטוב ביותר בין קצב העברת הנתונים לבין זמן האחזור.ברירת המחדל היא ‎-1.
• ‫gcsOutputDirectory: הספרייה הזו משמשת לכתיבת קובצי ה-AVRO של הרשומות שנקראו מהמקור. לדוגמה, gs://your-bucket/your-path. ברירת המחדל היא ריקה.
• ‫disabledAlgorithms: אלגוריתמים שמופרדים בפסיקים שרוצים להשבית. אם הערך מוגדר כ-none, אף אלגוריתם לא מושבת. חשוב להשתמש בפרמטר הזה בזהירות, כי האלגוריתמים שמושבתים כברירת מחדל עלולים להכיל פגיעויות או בעיות בביצועים. לדוגמה, SSLv3, RC4.
• ‫extraFilesToStage: נתיבים ב-Cloud Storage או סודות ב-Secret Manager של קבצים להעברה זמנית לעובד, מופרדים בפסיקים. הקבצים האלה נשמרים בספרייה ‎ /extra_files בכל עובד. לדוגמה, gs://<BUCKET_NAME>/file.txt,projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<VERSION_ID>.

הרצת התבנית

המסוף

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

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

4. בתפריט הנפתח Dataflow template (תבנית Dataflow), בוחרים בתבנית Sourcedb to Spanner (מ-Sourcedb אל 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/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/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית. אפשר למצוא את הגרסה הזו בתיקיית ההורה המתאימה עם התאריך בדלי – gs://dataflow-templates/
  זהירות: יכול להיות שהתבניות בגרסה העדכנית יעודכנו עם שינויים שיגרמו לבעיות במערכת. כדי למנוע את ההשפעה של שינויים שעלולים לשבור את תהליכי העבודה בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכני ביותר.
• ‫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/VERSION/flex/Sourcedb_to_Spanner_Flex",
     "environment": { "maxWorkers": "10" }
  }
}

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

• ‫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
• ‫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.source.ISrcToSpSourceConnector;
import com.google.cloud.teleport.v2.source.SourceConnectorFactory;
import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard;
import com.google.cloud.teleport.v2.spanner.migrations.source.config.JdbcShardConfig;
import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConnectionConfig;
import com.google.cloud.teleport.v2.spanner.migrations.utils.DataflowWorkerMachineTypeUtils;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.util.Optional;
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;
  }

  /**
   * Validates the provided pipeline options. TODO: move this to source connector.
   *
   * @param options The execution parameters to the pipeline.
   * @param sourceConnectionConfig Parsed source connection config.
   * @throws IllegalArgumentException if the provided options are invalid for the pipeline.
   */
  @VisibleForTesting
  static void validateOptions(
      SourceDbToSpannerOptions options, SourceConnectionConfig sourceConnectionConfig) {
    if (SourceDbToSpannerOptions.PG_SOURCE_DIALECT.equals(options.getSourceDbDialect())) {
      Preconditions.checkArgument(
          (sourceConnectionConfig instanceof JdbcShardConfig),
          "Postgresql dialect should have JDBC source config.");
      for (Shard shard : ((JdbcShardConfig) sourceConnectionConfig).getShardConfigs()) {
        if (StringUtils.isNotBlank(shard.getNamespace())
            && !shard.getNamespace().equals("public")) {
          throw new IllegalArgumentException(
              "Non-public namespaces are currently unsupported for PostgreSQL migrations.");
        }
      }
    }
  }

  /**
   * 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) {
    Pipeline pipeline = Pipeline.create(options);
    String workerMachineType =
        pipeline.getOptions().as(DataflowPipelineWorkerPoolOptions.class).getWorkerMachineType();
    Optional<Integer> resourceHintsMinCpus =
        DataflowWorkerMachineTypeUtils.getMinCpuResourceHint(pipeline.getOptions());
    if (!pipeline.getOptions().getRunner().getSimpleName().equals("DirectRunner")) {
      DataflowWorkerMachineTypeUtils.validateMachineSpecs(
          workerMachineType, 4, resourceHintsMinCpus);
    }

    SpannerConfig spannerConfig = createSpannerConfig(options);
    SourceConnectionConfig sourceConnectionConfig =
        PipelineController.getSourceConnectionConfig(
            options.getSourceDbDialect(), options.getSourceConfigURL());

    validateOptions(options, sourceConnectionConfig);

    // Decide type and source of migration
    ISrcToSpSourceConnector connector = SourceConnectorFactory.getSourceConnectorByDialect(options);
    return connector.executeMigration(options, sourceConnectionConfig, 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;
  }
}