תבנית Cloud Storage Avro to Spanner

התבנית Cloud Storage Avro files to Spanner היא צינור להעברת נתונים באצווה שקורא קובצי Avro שיוצאו מ-Spanner ואוחסנו ב-Cloud Storage, ומייבא אותם למסד נתונים של Spanner.

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

• מסד הנתונים של Spanner שמוגדר כיעד חייב להתקיים ולא להכיל נתונים.
• צריכות להיות לכם הרשאות קריאה לקטגוריה ב-Cloud Storage והרשאות כתיבה למסד הנתונים של Spanner שאליו אתם מייצאים.
• הנתיב של Cloud Storage שמוגדר כקלט חייב להתקיים, והוא צריך לכלול קובץ spanner-export.json שמכיל תיאור JSON של הקבצים לייבוא.
• אם קובץ ה-Avro של המקור לא מכיל מפתח ראשי, צריך ליצור טבלת Spanner ריקה עם מפתח ראשי לפני שמריצים את התבנית. השלב הזה לא נדרש אם קובץ ה-Avro מגדיר את המפתח הראשי.

אזהרה: אם מסד הנתונים של Spanner לא ריק, יכול להיות שנתונים חדשים יחליפו נתונים קיימים, או שהייבוא ייכשל.

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

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

• ‫instanceId: מזהה המכונה של מסד הנתונים של Spanner.
• ‫databaseId: מזהה מסד הנתונים של Spanner.
• ‫inputDir: הנתיב ב-Cloud Storage שממנו מייבאים את קובצי Avro.

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

• ‫spannerHost: נקודת הקצה של Cloud Spanner שאליה מתבצעת קריאה בתבנית. משמש רק לבדיקה. לדוגמה, https://batch-spanner.googleapis.com. ברירת המחדל היא: https://batch-spanner.googleapis.com.
• ‫waitForIndexes: אם הערך הוא true, צינור העיבוד ממתין ליצירת האינדקסים. אם false, יכול להיות שהעבודה תסתיים בזמן שיצירת האינדקסים עדיין מתבצעת ברקע. ערך ברירת המחדל הוא false.
• ‫waitForForeignKeys: אם הערך הוא true, צינור הנתונים ממתין ליצירת מפתחות זרים. אם false, יכול להיות שהעבודה תסתיים בזמן שמפתחות זרים עדיין נוצרים ברקע. ערך ברירת המחדל הוא false.
• ‫waitForChangeStreams: אם הערך הוא true, פייפליין ממתין ליצירת סנכרון שינויים בזרמי נתונים. אם false, יכול להיות שהעבודה תושלם בזמן שיצירת סנכרון שינויים בזרמי נתונים עדיין מתבצעת ברקע. ערך ברירת המחדל הוא true.
• ‫waitForSequences: כברירת מחדל, צינור הייבוא נחסם בזמן יצירת רצף. אם התוצאה היא false, יכול להיות שצינור הייבוא יסתיים כשברקע עדיין ייווצרו רצפים.
• ‫earlyIndexCreateFlag: מציין אם יצירת אינדקס מוקדמת מופעלת. אם התבנית מפעילה מספר גדול של הצהרות DDL, יעיל יותר ליצור אינדקסים לפני טעינת הנתונים. לכן, התנהגות ברירת המחדל היא ליצור את האינדקסים קודם אם מספר הצהרות ה-DDL חורג מסף מסוים. כדי להשבית את התכונה הזו, צריך להגדיר לפרמטר earlyIndexCreateFlag את הערך false. ערך ברירת המחדל הוא true.
• ‫spannerProjectId: המזהה של פרויקט Google Cloud שמכיל את מסד הנתונים של Spanner. אם לא מגדירים את הפרויקט, נעשה שימוש בפרויקט ברירת המחדל ב-Google Cloud.
• ‫ddlCreationTimeoutInMinutes: הזמן הקצוב בדקות להצהרות DDL שהתבנית מבצעת. ערך ברירת המחדל הוא 30 דקות.
• ‫spannerPriority: העדיפות של הבקשה לקריאות ל-Spanner. הערכים האפשריים הם HIGH, MEDIUM ו-LOW. ערך ברירת המחדל הוא MEDIUM.
• ‫earlyIndexCreateThreshold: הסף למספר האינדקסים והמפתחות הזרים שקובע אם ליצור אינדקסים לפני טעינת הנתונים. אם המספר הכולל של האינדקסים והמפתחות החיצוניים גדול יותר מהסף הזה, הם ייווצרו לפני ייבוא הנתונים כדי לשפר את הביצועים. ערך ברירת המחדל הוא 40.
• ‫runIndexDdlInParallel: אם הערך הוא true, צינור עיבוד הנתונים מריץ הצהרות DDL ליצירת אינדקס במקביל. ערך ברירת המחדל הוא false.

הרצת התבנית

המסוף

1. עוברים לדף Create job from template (יצירת משימה מתבנית) ב-Dataflow.
כניסה לדף Create job from template
2. בשדה שם המשימה, מזינים שם ייחודי למשימה.

   כדי שהעבודה תופיע בדף Instances של מסוף Google Cloud Spanner, שם העבודה צריך להיות בפורמט הבא:

   cloud-spanner-import-SPANNER_INSTANCE_ID-SPANNER_DATABASE_NAME

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

   • ‫SPANNER_INSTANCE_ID: מזהה מכונת Spanner
   • ‫SPANNER_DATABASE_NAME: השם של מסד הנתונים שלכם ב-Spanner
3. אופציונלי: בשדה Regional endpoint (נקודת קצה אזורית), בוחרים ערך מהתפריט הנפתח. אזור ברירת המחדל הוא us-central1.

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

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

gcloud

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

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

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-REGION_NAME/VERSION/GCS_Avro_to_Cloud_Spanner \
    --region REGION_NAME \
    --staging-location GCS_STAGING_LOCATION \
    --parameters \
instanceId=INSTANCE_ID,\
databaseId=DATABASE_ID,\
inputDir=GCS_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
• ‫INSTANCE_ID: המזהה של מופע Spanner שמכיל את מסד הנתונים
• ‫DATABASE_ID: המזהה של מסד הנתונים של Spanner שאליו רוצים לייבא
• ‫GCS_DIRECTORY: הנתיב ב-Cloud Storage שממנו מייבאים את קובצי ה-Avro, לדוגמה, gs://mybucket/somefolder

API

כדי להפעיל את התבנית באמצעות API בארכיטקטורת REST, שולחים בקשת HTTP POST. מידע נוסף על ה-API ועל היקפי ההרשאות שלו זמין במאמר projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates-LOCATION/VERSION/GCS_Avro_to_Cloud_Spanner
{
   "jobName": "JOB_NAME",
   "parameters": {
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "inputDir": "gs://GCS_DIRECTORY"
   },
   "environment": {
       "machineType": "n1-standard-2"
   }
}

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

• ‫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
• ‫INSTANCE_ID: המזהה של מופע Spanner שמכיל את מסד הנתונים
• ‫DATABASE_ID: המזהה של מסד הנתונים של Spanner שאליו רוצים לייבא
• ‫GCS_DIRECTORY: הנתיב ב-Cloud Storage שממנו מייבאים את קובצי ה-Avro, לדוגמה, gs://mybucket/somefolder

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

Java

/*
 * Copyright (C) 2018 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.spanner;

import com.google.cloud.spanner.Options.RpcPriority;
import com.google.cloud.spanner.SpannerOptions;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateCreationParameter;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.metadata.TemplateParameter.TemplateEnumOption;
import com.google.cloud.teleport.spanner.ImportPipeline.Options;
import com.google.cloud.teleport.spanner.spannerio.SpannerConfig;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.transforms.SerializableFunction;

/**
 * Avro to Cloud Spanner Import pipeline.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v1/README_GCS_Avro_to_Cloud_Spanner.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "GCS_Avro_to_Cloud_Spanner",
    category = TemplateCategory.BATCH,
    displayName = "Avro Files on Cloud Storage to Cloud Spanner",
    description =
        "The Cloud Storage Avro files to Cloud Spanner template is a batch pipeline that reads Avro files exported from "
            + "Cloud Spanner stored in Cloud Storage and imports them to a Cloud Spanner database.",
    optionsClass = Options.class,
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/avro-to-cloud-spanner",
    contactInformation = "https://cloud.google.com/support",
    requirements = {
      "The target Cloud Spanner database must exist and must be empty.",
      "You must have read permissions for the Cloud Storage bucket and write permissions for the target Cloud Spanner database.",
      "The input Cloud Storage path must exist, and it must include a <a href=\"https://cloud.google.com/spanner/docs/import-non-spanner#create-export-json\">spanner-export.json</a> file that contains a JSON description of files to import."
    })
public class ImportPipeline {

  /** Options for {@link ImportPipeline}. */
  public interface Options extends PipelineOptions {

    @TemplateParameter.Text(
        order = 1,
        groupName = "Target",
        regexes = {"^[a-z0-9\\-]+$"},
        description = "Cloud Spanner instance ID",
        helpText = "The instance ID of the Spanner database.")
    ValueProvider<String> getInstanceId();

    void setInstanceId(ValueProvider<String> value);

    @TemplateParameter.Text(
        order = 2,
        groupName = "Target",
        regexes = {"^[a-z_0-9\\-]+$"},
        description = "Cloud Spanner database ID",
        helpText = "The database ID of the Spanner database.")
    ValueProvider<String> getDatabaseId();

    void setDatabaseId(ValueProvider<String> value);

    @TemplateParameter.GcsReadFolder(
        order = 3,
        groupName = "Source",
        description = "Cloud storage input directory",
        helpText = "The Cloud Storage path where the Avro files are imported from.")
    ValueProvider<String> getInputDir();

    void setInputDir(ValueProvider<String> value);

    @TemplateParameter.Text(
        order = 4,
        groupName = "Target",
        optional = true,
        description = "Cloud Spanner Endpoint to call",
        helpText = "The Cloud Spanner endpoint to call in the template. Only used for testing.",
        example = "https://batch-spanner.googleapis.com")
    @Default.String("https://batch-spanner.googleapis.com")
    ValueProvider<String> getSpannerHost();

    void setSpannerHost(ValueProvider<String> value);

    @TemplateParameter.Boolean(
        order = 5,
        optional = true,
        description = "Wait for Indexes",
        helpText =
            "If `true`, the pipeline waits for indexes to be created. If `false`, the job might complete while indexes are still being created in the background. The default value is `false`.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getWaitForIndexes();

    void setWaitForIndexes(ValueProvider<Boolean> value);

    @TemplateParameter.Boolean(
        order = 6,
        optional = true,
        description = "Wait for Foreign Keys",
        helpText =
            "If `true`, the pipeline waits for foreign keys to be created. If `false`, the job might complete while foreign keys are still being created in the background. The default value is `false`.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getWaitForForeignKeys();

    void setWaitForForeignKeys(ValueProvider<Boolean> value);

    @TemplateParameter.Boolean(
        order = 7,
        optional = true,
        description = "Wait for Change Streams",
        helpText =
            "If `true`, the pipeline waits for change streams to be created. If `false`, the job might complete while change streams are still being created in the background. The default value is `true`.")
    @Default.Boolean(true)
    ValueProvider<Boolean> getWaitForChangeStreams();

    void setWaitForChangeStreams(ValueProvider<Boolean> value);

    @TemplateParameter.Boolean(
        order = 7,
        optional = true,
        description = "Wait for Sequences",
        helpText =
            "By default, the import pipeline is blocked on sequence creation. If `false`, the import pipeline might"
                + " complete with sequences still being created in the background.")
    @Default.Boolean(true)
    ValueProvider<Boolean> getWaitForSequences();

    void setWaitForSequences(ValueProvider<Boolean> value);

    @TemplateParameter.Boolean(
        order = 8,
        optional = true,
        description = "Create Indexes early",
        helpText =
            "Specifies whether early index creation is enabled. If the template runs a large number of DDL statements, it's more efficient to create indexes before loading data. Therefore, the default behavior is to create the indexes first when the number of DDL statements exceeds a threshold. To disable this feature, set `earlyIndexCreateFlag` to `false`. The default value is `true`.")
    @Default.Boolean(true)
    ValueProvider<Boolean> getEarlyIndexCreateFlag();

    void setEarlyIndexCreateFlag(ValueProvider<Boolean> value);

    @TemplateCreationParameter(value = "false")
    @Description("If true, wait for job finish")
    @Default.Boolean(true)
    boolean getWaitUntilFinish();

    @TemplateParameter.ProjectId(
        order = 9,
        groupName = "Target",
        optional = true,
        description = "Cloud Spanner Project Id",
        helpText =
            "The ID of the Google Cloud project that contains the Spanner database. If not set, the default Google Cloud project is used.")
    ValueProvider<String> getSpannerProjectId();

    void setSpannerProjectId(ValueProvider<String> value);

    void setWaitUntilFinish(boolean value);

    @TemplateParameter.Integer(
        order = 10,
        optional = true,
        description = "DDL Creation timeout in minutes",
        helpText =
            "The timeout in minutes for DDL statements performed by the template. The default value is 30 minutes.")
    @Default.Integer(30)
    ValueProvider<Integer> getDdlCreationTimeoutInMinutes();

    void setDdlCreationTimeoutInMinutes(ValueProvider<Integer> value);

    @TemplateParameter.Enum(
        order = 11,
        groupName = "Target",
        enumOptions = {
          @TemplateEnumOption("LOW"),
          @TemplateEnumOption("MEDIUM"),
          @TemplateEnumOption("HIGH")
        },
        optional = true,
        description = "Priority for Spanner RPC invocations",
        helpText =
            "The request priority for Spanner calls. Possible values are `HIGH`, `MEDIUM`, and `LOW`. The default value is `MEDIUM`.")
    ValueProvider<RpcPriority> getSpannerPriority();

    void setSpannerPriority(ValueProvider<RpcPriority> value);

    @TemplateParameter.Integer(
        order = 12,
        optional = true,
        description = "Early Index Create Threshold",
        helpText =
            "The threshold for the number of indexes and foreign keys that determines whether to create indexes before data loading. If the total number of indexes and foreign keys is larger than this threshold, they will be created before data import for better performance. The default value is 40.")
    @Default.Integer(40)
    ValueProvider<Integer> getEarlyIndexCreateThreshold();

    void setEarlyIndexCreateThreshold(ValueProvider<Integer> value);

    @TemplateParameter.Boolean(
        order = 13,
        optional = true,
        description = "Run Index DDL in parallel",
        helpText =
            "If `true`, the pipeline executes index creation DDL statements in parallel. The default value is `false`.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getRunIndexDdlInParallel();

    void setRunIndexDdlInParallel(ValueProvider<Boolean> value);
  }

  public static void main(String[] args) {

    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);

    Pipeline p = Pipeline.create(options);

    SpannerConfig spannerConfig =
        SpannerConfig.create()
            // Temporary fix explicitly setting SpannerConfig.projectId to the default project
            // if spannerProjectId is not provided as a parameter. Required as of Beam 2.38,
            // which no longer accepts null label values on metrics, and SpannerIO#setup() has
            // a bug resulting in the label value being set to the original parameter value,
            // with no fallback to the default project.
            // TODO: remove NestedValueProvider when this is fixed in Beam.
            .withProjectId(
                NestedValueProvider.of(
                    options.getSpannerProjectId(),
                    (SerializableFunction<String, String>)
                        input -> input != null ? input : SpannerOptions.getDefaultProjectId()))
            .withHost(options.getSpannerHost())
            .withInstanceId(options.getInstanceId())
            .withDatabaseId(options.getDatabaseId())
            .withRpcPriority(options.getSpannerPriority());

    p.apply(
        new ImportTransform(
            spannerConfig,
            options.getInputDir(),
            options.getWaitForIndexes(),
            options.getWaitForForeignKeys(),
            options.getWaitForChangeStreams(),
            options.getWaitForSequences(),
            options.getEarlyIndexCreateFlag(),
            options.getDdlCreationTimeoutInMinutes(),
            options.getEarlyIndexCreateThreshold(),
            options.getRunIndexDdlInParallel()));

    PipelineResult result = p.run();

    if (options.getWaitUntilFinish()
        &&
        /* Only if template location is null, there is a dataflow job to wait for. Else it's
         * template generation which doesn't start a dataflow job.
         */
        options.as(DataflowPipelineOptions.class).getTemplateLocation() == null) {
      result.waitUntilFinish();
    }
  }
}

המאמרים הבאים

• מידע נוסף על תבניות Dataflow
• כאן אפשר לעיין ברשימת התבניות ש-Google סיפקה.