תבנית של סנכרון שינויים בזרמי נתונים של Spanner ל-Cloud Storage

תבנית Spanner change streams to Cloud Storage היא צינור להזרמת נתונים (pipeline) שמעביר בסטרימינג רשומות של שינויים בנתוני Spanner וכותב אותן לקטגוריה של Cloud Storage באמצעות Dataflow Portable Runner.

הערה: מחבר Spanner change streams לא תומך בהפסקת העבודה של משימה. כדי להפסיק את צינור הנתונים, מבטלים את העבודה במקום לנקז אותו. הפסקת העבודה של משימה באמצעות מחבר של Spanner לשינוי נתונים יכולה לגרום לכך שהמשימה תיתקע במצב של הפסקת עבודה. מידע נוסף זמין במאמר בנושא Draining במסמכי Spanner.

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

שימו לב שאפשר לצמצם את זמן האחזור ברשת ואת עלויות התעבורה ברשת על ידי הפעלת משימת Dataflow מאותו אזור שבו נמצאת מכונת Spanner או קטגוריית Cloud Storage. אם אתם משתמשים במקורות, ביעדים, במיקומים של קבצים זמניים או במיקומים של קבצים להעברה שנמצאים מחוץ לאזור של העבודה, יכול להיות שהנתונים שלכם יישלחו בין אזורים. מידע נוסף על אזורי Dataflow

מידע נוסף על מקורות לשינויים, איך יוצרים צינורות עיבוד נתונים של Dataflow למקורות לשינויים ושיטות מומלצות

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

• מופע Spanner חייב להתקיים לפני שמריצים את צינור הנתונים.
• מסד הנתונים של Spanner צריך להתקיים לפני שמריצים את צינור הנתונים.
• מופע המטא-נתונים של Spanner צריך להתקיים לפני שמריצים את צינור העיבוד.
• מסד הנתונים של המטא-נתונים של Spanner צריך להתקיים לפני שמריצים את צינור הנתונים.
• השינוי ב-Spanner צריך להתבצע לפני שמריצים את צינור הנתונים.
• קטגוריית הפלט של Cloud Storage צריכה להתקיים לפני שמריצים את צינור הנתונים.

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

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

• ‫spannerInstanceId: מזהה מופע Spanner שממנו ייקראו נתונים של מקורות לשינויים.
• ‫spannerDatabase: מסד הנתונים של Spanner שממנו יקראו נתונים של מקורות נתונים.
• ‫spannerMetadataInstanceId: מזהה מופע Spanner לשימוש בטבלת המטא-נתונים של מחבר מקורות הנתונים.
• ‫spannerMetadataDatabase: מסד הנתונים של Spanner שבו יש להשתמש עבור טבלת המטא-נתונים של מחבר הנתונים של שינוי הזרמים.
• ‫spannerChangeStreamName: השם של Spanner change stream שממנו ייקראו הנתונים.
• ‫gcsOutputDirectory: הנתיב והקידומת של שם הקובץ לכתיבת קובצי פלט. חייב להסתיים בלוכסן. הפורמט של DateTime משמש לניתוח נתיב הספרייה עבור מעצבי תאריכים ושעות. לדוגמה, gs://your-bucket/your-path.

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

• ‫spannerProjectId: המזהה של פרויקט Google Cloud שמכיל את מסד הנתונים של Spanner שממנו רוצים לקרוא את נתוני ה-Change Streams. בפרויקט הזה נוצר גם טבלת המטא-נתונים של המחבר של זרמי השינויים. ברירת המחדל של הפרמטר הזה היא הפרויקט שבו צינור ה-Dataflow פועל.
• ‫spannerDatabaseRole: תפקיד מסד הנתונים ב-Spanner שבו יש להשתמש כשמריצים את התבנית. הפרמטר הזה נדרש רק אם משתמש ה-IAM שמריץ את התבנית הוא משתמש עם בקרת גישה פרטנית. לתפקיד במסד הנתונים צריכה להיות הרשאת SELECT בפיד השינויים והרשאת EXECUTE בפונקציית הקריאה של פיד השינויים. מידע נוסף זמין במאמר בנושא בקרת גישה ברמת גרנולריות גבוהה לזרמי שינויים (https://cloud.google.com/spanner/docs/fgac-change-streams).
• ‫spannerMetadataTableName: השם של טבלת המטא-נתונים של מחבר Spanner change streams שבה רוצים להשתמש. אם לא מספקים את המידע הזה, נוצרת באופן אוטומטי טבלת מטא-נתונים של זרמי שינויים ב-Spanner במהלך הביצוע של צינור הנתונים. כשמעדכנים צינור קיים, צריך לציין ערך לפרמטר הזה. אחרת, אל תשתמשו בפרמטר הזה.
• ‫startTimestamp: תאריך ושעת ההתחלה, כולל, לשימוש בקריאת נתוני שינויים, בפורמט Ex-2021-10-12T07:20:50.52Z. ברירת המחדל היא חותמת הזמן של תחילת הצינור, כלומר השעה הנוכחית.
• ‫endTimestamp: תאריך ושעת הסיום, כולל, לשימוש בקריאת שינויים בנתונים. לדוגמה, Ex-2021-10-12T07:20:50.52Z. ברירת המחדל היא זמן אינסופי בעתיד.
• ‫spannerHost: נקודת הקצה של Cloud Spanner שאליה מתקשרים בתבנית. משמש רק לבדיקה. לדוגמה, https://spanner.googleapis.com. ברירת המחדל היא: https://spanner.googleapis.com.
• ‫outputFileFormat: הפורמט של קובץ הפלט ב-Cloud Storage. הפורמטים המותרים הם TEXT ו-AVRO. ברירת המחדל היא AVRO.
• ‫windowDuration: משך החלון הוא המרווח שבו הנתונים נכתבים לספריית הפלט. מגדירים את משך הזמן על סמך קצב העברת הנתונים של צינור עיבוד הנתונים. לדוגמה, כדי להשיג תפוקה גבוהה יותר, יכול להיות שיהיה צורך להקטין את גודל החלון כדי שהנתונים יתאימו לזיכרון. ברירת המחדל היא 5m (חמש דקות), והמינימום הוא 1s (שנייה אחת). הפורמטים המותרים הם: [int]s (לשניות, לדוגמה: 5s),‏ [int]m (לדקות, לדוגמה: 12m),‏ [int]h (לשעות, לדוגמה: 2h). לדוגמה, 5m.
• ‫rpcPriority: העדיפות של הבקשה לשיחות Spanner. הערך חייב להיות HIGH,‏ MEDIUM או LOW. ברירת המחדל היא HIGH.
• ‫spannerChangeStreamTvfNameList: רשימה של שמות של Spanner Change Stream TVF, מופרדים באמצעות נקודה-פסיק, שאותם רוצים לשאול ולאחד. ברירת המחדל היא ריק.
• ‫outputFilenamePrefix: הקידומת שתוצב בכל קובץ עם חלון. לדוגמה, output-. ברירת המחדל היא: output.
• ‫numShards: המספר המקסימלי של רסיסי פלט שנוצרים במהלך הכתיבה. מספר גבוה יותר של רסיסים פירושו תפוקה גבוהה יותר לכתיבה ב-Cloud Storage, אבל יכול להיות שעלות צבירת הנתונים תהיה גבוהה יותר ברסיסים כשמעבדים קבצים של פלט Cloud Storage. ברירת המחדל היא 20.

הרצת התבנית

המסוף

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

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

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

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-REGION_NAME/VERSION/flex/ \
    --region REGION_NAME \
    --parameters \
spannerInstanceId=SPANNER_INSTANCE_ID,\
spannerDatabase=SPANNER_DATABASE,\
spannerMetadataInstanceId=SPANNER_METADATA_INSTANCE_ID,\
spannerMetadataDatabase=SPANNER_METADATA_DATABASE,\
spannerChangeStreamName=SPANNER_CHANGE_STREAM,\
gcsOutputDirectory=GCS_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
• ‫SPANNER_INSTANCE_ID: מזהה המכונה של Cloud Spanner
• ‫SPANNER_DATABASE: מסד נתונים של Cloud Spanner
• ‫SPANNER_METADATA_INSTANCE_ID: מזהה מופע של מטא נתונים ב-Cloud Spanner
• ‫SPANNER_METADATA_DATABASE: מסד נתונים של מטא נתונים ב-Cloud Spanner
• ‫SPANNER_CHANGE_STREAM: Cloud Spanner change stream
• ‫GCS_OUTPUT_DIRECTORY: מיקום הקובץ של פלט נתוני השינויים

API

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "spannerInstanceId": "SPANNER_INSTANCE_ID",
          "spannerDatabase": "SPANNER_DATABASE",
          "spannerMetadataInstanceId": "SPANNER_METADATA_INSTANCE_ID",
          "spannerMetadataDatabase": "SPANNER_METADATA_DATABASE",
          "spannerChangeStreamName": "SPANNER_CHANGE_STREAM",
          "gcsOutputDirectory": "GCS_OUTPUT_DIRECTORY"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/",
   }
}

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

• ‫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
• ‫SPANNER_INSTANCE_ID: מזהה המכונה של Cloud Spanner
• ‫SPANNER_DATABASE: מסד נתונים של Cloud Spanner
• ‫SPANNER_METADATA_INSTANCE_ID: מזהה מופע של מטא נתונים ב-Cloud Spanner
• ‫SPANNER_METADATA_DATABASE: מסד נתונים של מטא נתונים ב-Cloud Spanner
• ‫SPANNER_CHANGE_STREAM: Cloud Spanner change stream
• ‫GCS_OUTPUT_DIRECTORY: מיקום הקובץ של פלט נתוני השינויים

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

Java

/*
 * Copyright (C) 2022 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.Timestamp;
import com.google.cloud.spanner.Options.RpcPriority;
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.options.SpannerChangeStreamsToGcsOptions;
import com.google.cloud.teleport.v2.transforms.FileFormatFactorySpannerChangeStreams;
import com.google.cloud.teleport.v2.utils.DurationUtils;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
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.io.gcp.spanner.SpannerIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link SpannerChangeStreamsToGcs} pipeline streams change stream record(s) and stores to
 * Google Cloud Storage bucket in user specified format. The sink data can be stored in a Text or
 * Avro file format.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-googlecloud/README_Spanner_Change_Streams_to_Google_Cloud_Storage.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Spanner_Change_Streams_to_Google_Cloud_Storage",
    category = TemplateCategory.STREAMING,
    displayName = "Cloud Spanner change streams to Cloud Storage",
    description = {
      "The Cloud Spanner change streams to Cloud Storage template is a streaming pipeline that streams Spanner data change records and writes them into a Cloud Storage bucket using Dataflow Runner V2.\n",
      "The pipeline groups Spanner change stream records into windows based on their timestamp, with each window representing a time duration whose length you can configure with this template. "
          + "All records with timestamps belonging to the window are guaranteed to be in the window; there can be no late arrivals. "
          + "You can also define a number of output shards; the pipeline creates one Cloud Storage output file per window per shard. "
          + "Within an output file, records are unordered. Output files can be written in either JSON or AVRO format, depending on the user configuration.\n",
      "Note that you can minimize network latency and network transport costs by running the Dataflow job from the same region as your Cloud Spanner instance or Cloud Storage bucket. "
          + "If you use sources, sinks, staging file locations, or temporary file locations that are located outside of your job's region, your data might be sent across regions. "
          + "See more about <a href=\"https://cloud.google.com/dataflow/docs/concepts/regional-endpoints\">Dataflow regional endpoints</a>.\n",
      "Learn more about <a href=\"https://cloud.google.com/spanner/docs/change-streams\">change streams</a>, <a href=\"https://cloud.google.com/spanner/docs/change-streams/use-dataflow\">how to build change streams Dataflow pipelines</a>, and <a href=\"https://cloud.google.com/spanner/docs/change-streams/use-dataflow#best_practices\">best practices</a>."
    },
    optionsClass = SpannerChangeStreamsToGcsOptions.class,
    flexContainerName = "googlecloud-to-googlecloud",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-spanner-change-streams-to-cloud-storage",
    contactInformation = "https://cloud.google.com/support",
    requirements = {
      "The Cloud Spanner instance must exist prior to running the pipeline.",
      "The Cloud Spanner database must exist prior to running the pipeline.",
      "The Cloud Spanner metadata instance must exist prior to running the pipeline.",
      "The Cloud Spanner metadata database must exist prior to running the pipeline.",
      "The Cloud Spanner change stream must exist prior to running the pipeline.",
      "The Cloud Storage output bucket must exist prior to running the pipeline."
    },
    streaming = true,
    supportsAtLeastOnce = true)
public class SpannerChangeStreamsToGcs {
  private static final Logger LOG = LoggerFactory.getLogger(SpannerChangeStreamsToGcs.class);
  private static final String USE_RUNNER_V2_EXPERIMENT = "use_runner_v2";

  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    LOG.info("Starting Input Files to GCS");

    SpannerChangeStreamsToGcsOptions options =
        PipelineOptionsFactory.fromArgs(args).as(SpannerChangeStreamsToGcsOptions.class);

    run(options);
  }

  private static String getProjectId(SpannerChangeStreamsToGcsOptions options) {
    return options.getSpannerProjectId().isEmpty()
        ? options.getProject()
        : options.getSpannerProjectId();
  }

  public static PipelineResult run(SpannerChangeStreamsToGcsOptions options) {
    LOG.info("Requested File Format is " + options.getOutputFileFormat());
    options.setStreaming(true);
    options.setEnableStreamingEngine(true);

    final Pipeline pipeline = Pipeline.create(options);

    // Get the Spanner project, instance, database, and change stream parameters.
    String projectId = getProjectId(options);
    String instanceId = options.getSpannerInstanceId();
    String databaseId = options.getSpannerDatabase();
    String metadataInstanceId = options.getSpannerMetadataInstanceId();
    String metadataDatabaseId = options.getSpannerMetadataDatabase();
    String changeStreamName = options.getSpannerChangeStreamName();

    // Retrieve and parse the start / end timestamps.
    Timestamp startTimestamp =
        options.getStartTimestamp().isEmpty()
            ? Timestamp.now()
            : Timestamp.parseTimestamp(options.getStartTimestamp());
    Timestamp endTimestamp =
        options.getEndTimestamp().isEmpty()
            ? Timestamp.MAX_VALUE
            : Timestamp.parseTimestamp(options.getEndTimestamp());

    // Add use_runner_v2 to the experiments option, since Change Streams connector is only supported
    // on Dataflow runner v2.
    List<String> experiments = options.getExperiments();
    if (experiments == null) {
      experiments = new ArrayList<>();
    }
    if (!experiments.contains(USE_RUNNER_V2_EXPERIMENT)) {
      experiments.add(USE_RUNNER_V2_EXPERIMENT);
    }
    options.setExperiments(experiments);

    String metadataTableName =
        options.getSpannerMetadataTableName() == null
            ? null
            : options.getSpannerMetadataTableName();

    String tvfNameListString = options.getSpannerChangeStreamTvfNameList();
    List<String> tvfNameList = null;
    if (tvfNameListString != null && !tvfNameListString.isEmpty()) {
      tvfNameList =
          Arrays.stream(tvfNameListString.split(";"))
              .filter(name -> !name.trim().isEmpty())
              .collect(Collectors.toList());
    }

    final RpcPriority rpcPriority = options.getRpcPriority();
    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()))
            .withProjectId(projectId)
            .withInstanceId(instanceId)
            .withDatabaseId(databaseId);
    // Propagate database role for fine-grained access control on change stream.
    if (options.getSpannerDatabaseRole() != null) {
      LOG.info("Setting database role on SpannerConfig: " + options.getSpannerDatabaseRole());
      spannerConfig =
          spannerConfig.withDatabaseRole(
              ValueProvider.StaticValueProvider.of(options.getSpannerDatabaseRole()));
    }
    LOG.info("Created SpannerConfig: " + spannerConfig);
    pipeline
        .apply(
            SpannerIO.readChangeStream()
                .withSpannerConfig(spannerConfig)
                .withMetadataInstance(metadataInstanceId)
                .withMetadataDatabase(metadataDatabaseId)
                .withChangeStreamName(changeStreamName)
                .withInclusiveStartAt(startTimestamp)
                .withInclusiveEndAt(endTimestamp)
                .withRpcPriority(rpcPriority)
                .withMetadataTable(metadataTableName)
                .withTvfNameList(tvfNameList))
        .apply(
            "Creating " + options.getWindowDuration() + " Window",
            Window.into(FixedWindows.of(DurationUtils.parseDuration(options.getWindowDuration()))))
        .apply(
            "Write To GCS",
            FileFormatFactorySpannerChangeStreams.newBuilder().setOptions(options).build());

    return pipeline.run();
  }
}

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

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