תבנית של סנכרון שינויים בזרמי נתונים ב-Spanner ל-Pub/Sub

תבנית סנכרון שינויים בזרמי נתונים של Spanner ל-Pub/Sub היא צינור עיבוד נתונים בסטרימינג שמעביר בסטרימינג רשומות של שינויים בנתוני Spanner וכותב אותן לנושאים ב-Pub/Sub באמצעות Dataflow Runner V2.

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

כדי להוציא את הנתונים לנושא חדש ב-Pub/Sub, צריך קודם ליצור את הנושא. אחרי היצירה, Pub/Sub יוצר ומצרף מינוי לנושא החדש באופן אוטומטי. אם מנסים להוציא נתונים לנושא ב-Pub/Sub שלא קיים, צינור עיבוד הנתונים של Dataflow זורק חריגה, והצינור נתקע כי הוא מנסה ליצור חיבור באופן רציף.

אם נושא Pub/Sub הנדרש כבר קיים, אפשר להוציא נתונים לנושא הזה.

מידע נוסף זמין במאמרים מידע על סנכרון שינויים בזרמי נתונים, יצירת חיבורים לסנכרון שינויים בזרמי נתונים באמצעות Dataflow ושיטות מומלצות לשימוש בסנכרון שינויים בזרמי נתונים.

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

• מופע Spanner צריך להתקיים לפני שמריצים את צינור הנתונים.
• מסד הנתונים של Spanner צריך להתקיים לפני שמריצים את צינור הנתונים.
• מופע המטא-נתונים של Spanner צריך להתקיים לפני שמריצים את צינור העיבוד.
• מסד הנתונים של המטא-נתונים של Spanner צריך להתקיים לפני שמריצים את צינור העיבוד.
• סנכרון שינויים בזרמי נתונים של Spanner חייב להתקיים לפני שמריצים את צינור הנתונים.
• הנושא ב-Pub/Sub צריך להתקיים לפני שמריצים את צינור הנתונים.

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

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

• ‫spannerInstanceId: מכונת Spanner שממנה יתבצע קריאה של סנכרון שינויים בזרמי נתונים.
• ‫spannerDatabase: מסד הנתונים של Spanner שממנו ייקראו סנכרון שינויים בזרמי נתונים.
• ‫spannerMetadataInstanceId: מופע Spanner לשימוש בטבלת המטא-נתונים של מחבר הנתונים של סנכרון שינויים בזרמי נתונים.
• ‫spannerMetadataDatabase: מסד הנתונים של Spanner שבו יש להשתמש עבור טבלת המטא-נתונים של מחבר סנכרון שינויים בזרמי נתונים.
• ‫spannerChangeStreamName: השם של סנכרון שינויים בזרמי נתונים ב-Spanner שממנו רוצים לקרוא.
• ‫pubsubTopic: נושא Pub/Sub לפלט של סנכרון שינויים בזרמי נתונים.

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

• ‫spannerProjectId: הפרויקט שממנו יקראו את נתוני סנכרון שינויים בזרמי נתונים. בפרויקט הזה נוצר גם טבלת המטא-נתונים של המחבר של סנכרון שינויים בזרמי נתונים. ברירת המחדל של הפרמטר הזה היא הפרויקט שבו צינור Dataflow פועל.
• ‫spannerDatabaseRole: תפקיד מסד הנתונים ב-Spanner שבו יש להשתמש כשמריצים את התבנית. הפרמטר הזה נדרש רק אם משתמש ה-IAM שמריץ את התבנית הוא משתמש עם בקרת גישה ברמת דיוק גבוהה. לתפקיד במסד הנתונים צריכה להיות הרשאת SELECT בסנכרון שינויים בזרמי נתונים והרשאת EXECUTE בפונקציית הקריאה של סנכרון שינויים בזרמי נתונים. מידע נוסף זמין במאמר בנושא בקרת גישה ברמת גרנולריות גבוהה לסנכרון שינויים בזרמי נתונים (https://cloud.google.com/spanner/docs/fgac-change-streams).
• ‫spannerMetadataTableName: שם טבלת המטא-נתונים של מחבר Spanner לשימוש בסנכרון שינויים בזרמי נתונים. אם לא מספקים את הטבלה, Spanner יוצר באופן אוטומטי את טבלת המטא-נתונים של מחבר הסטרימינג במהלך השינוי של צינור הנתונים. כשמעדכנים צינור קיים, צריך לציין את הפרמטר הזה. אין להשתמש בפרמטר הזה במקרים אחרים.
• ‫startTimestamp: תאריך ושעת ההתחלה (https://tools.ietf.org/html/rfc3339), כולל, לשימוש בקריאת סנכרון שינויים בזרמי נתונים. לדוגמה, ex- 2021-10-12T07:20:50.52Z. ברירת המחדל היא חותמת הזמן של תחילת הצינור, כלומר השעה הנוכחית.
• ‫endTimestamp: תאריך ושעת הסיום (https://tools.ietf.org/html/rfc3339), כולל, לשימוש בקריאת סנכרון שינויים בזרמי נתונים. לדוגמה, ex- 2021-10-12T07:20:50.52Z. ברירת המחדל היא זמן אינסופי בעתיד.
• ‫spannerHost: נקודת הקצה של Cloud Spanner שאליה מתבצעת קריאה בתבנית. היא משמשת לבדיקה בלבד. לדוגמה, https://spanner.googleapis.com. ברירת המחדל היא: https://spanner.googleapis.com.
• ‫outputDataFormat: הפורמט של הפלט. הפלט עטוף בהרבה PubsubMessages ונשלח לנושא Pub/Sub. הפורמטים המותרים הם JSON ו-AVRO. ברירת המחדל היא JSON.
• ‫pubsubAPI: Pub/Sub API שמשמש להטמעה של צינור עיבוד הנתונים. ממשקי ה-API המותרים הם pubsubio ו-native_client. במספר קטן של שאילתות לשנייה (QPS), ל-native_client יש זמן אחזור קצר יותר. במספר גדול של QPS, ‏ pubsubio מספק ביצועים טובים ויציבים יותר. ערך ברירת המחדל הוא pubsubio.
• ‫pubsubProjectId: הפרויקט של נושא Pub/Sub. ברירת המחדל של הפרמטר הזה היא הפרויקט שבו צינור Dataflow פועל.
• ‫rpcPriority: עדיפות הבקשה לשיחות Spanner. הערכים המותרים הם HIGH,‏ MEDIUM ו-LOW. ברירת המחדל היא: HIGH).
• ‫includeSpannerSource: קובע אם לכלול את מזהה מסד הנתונים ומזהה המופע של Spanner כדי לקרוא את סנכרון שינויים בזרמי נתונים בהודעת הפלט. ברירת המחדל היא: false.
• ‫outputMessageMetadata: ערך המחרוזת של השדה המותאם אישית outputMessageMetadata בהודעת הפלט של pub/sub. ברירת המחדל היא ריק, והשדה outputMessageMetadata מאוכלס רק אם הערך הזה לא ריק. כשמזינים את הערך כאן, צריך להוסיף תו בריחה (escape) לכל התווים המיוחדים (למשל: מירכאות כפולות).

הפעלת התבנית

המסוף

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

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

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

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/Spanner_Change_Streams_to_PubSub \
        --region REGION_NAME \
        --parameters \
    spannerInstanceId=SPANNER_INSTANCE_ID,\
    spannerDatabase=SPANNER_DATABASE,\
    spannerMetadataInstanceId=SPANNER_METADATA_INSTANCE_ID,\
    spannerMetadataDatabase=SPANNER_METADATA_DATABASE,\
    spannerChangeStreamName=SPANNER_CHANGE_STREAM,\
    pubsubTopic=PUBSUB_TOPIC
    

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

• ‫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: מזהה מכונת Spanner
• ‫SPANNER_DATABASE: מסד נתונים של Spanner
• ‫SPANNER_METADATA_INSTANCE_ID: מזהה מכונת מטא-נתונים של Spanner
• ‫SPANNER_METADATA_DATABASE: מסד נתונים של מטא-נתונים ב-Spanner
• ‫SPANNER_CHANGE_STREAM: Spanner change stream
• ‫PUBSUB_TOPIC: נושא ה-Pub/Sub לפלט של סנכרון שינויים בזרמי נתונים

API

כדי להריץ את התבנית באמצעות API בארכיטקטורת REST, שולחים בקשת 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",
            "pubsubTopic": "PUBSUB_TOPIC"
        },
        "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Spanner_Change_Streams_to_PubSub",
    }
  }
  

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

• ‫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: מזהה מכונת Spanner
• ‫SPANNER_DATABASE: מסד נתונים של Spanner
• ‫SPANNER_METADATA_INSTANCE_ID: מזהה מכונת מטא-נתונים של Spanner
• ‫SPANNER_METADATA_DATABASE: מסד נתונים של מטא-נתונים ב-Spanner
• ‫SPANNER_CHANGE_STREAM: Spanner change stream
• ‫PUBSUB_TOPIC: נושא ה-Pub/Sub לפלט של סנכרון שינויים בזרמי נתונים

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

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.SpannerChangeStreamsToPubSubOptions;
import com.google.cloud.teleport.v2.transforms.FileFormatFactorySpannerChangeStreamsToPubSub;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link SpannerChangeStreamsToPubSub} pipeline streams change stream record(s) and stores to
 * pubsub topic in user specified format. The sink data can be stored in a JSON Text or Avro data
 * format.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-googlecloud/README_Spanner_Change_Streams_to_PubSub.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Spanner_Change_Streams_to_PubSub",
    category = TemplateCategory.STREAMING,
    displayName = "Cloud Spanner change streams to Pub/Sub",
    description = {
      "The Cloud Spanner change streams to the Pub/Sub template is a streaming pipeline that streams Cloud Spanner data change records and writes them into Pub/Sub topics using Dataflow Runner V2.\n",
      "To output your data to a new Pub/Sub topic, you need to first create the topic. After creation, Pub/Sub automatically generates and attaches a subscription to the new topic. "
          + "If you try to output data to a Pub/Sub topic that doesn't exist, the dataflow pipeline throws an exception, and the pipeline gets stuck as it continuously tries to make a connection.\n",
      "If the necessary Pub/Sub topic already exists, you can output data to that topic.",
      "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 = SpannerChangeStreamsToPubSubOptions.class,
    flexContainerName = "googlecloud-to-googlecloud",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-spanner-change-streams-to-pubsub",
    contactInformation = "https://cloud.google.com/support",
    requirements = {
      "The Cloud Spanner instance must exist before 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 Pub/Sub topic must exist prior to running the pipeline."
    },
    streaming = true,
    supportsAtLeastOnce = true)
public class SpannerChangeStreamsToPubSub {
  private static final Logger LOG = LoggerFactory.getLogger(SpannerChangeStreamsToPubSub.class);
  private static final String USE_RUNNER_V2_EXPERIMENT = "use_runner_v2";

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

    LOG.info("Starting Input Messages to Pub/Sub");

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

    run(options);
  }

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

  private static String getPubsubProjectId(SpannerChangeStreamsToPubSubOptions options) {
    return options.getPubsubProjectId().isEmpty()
        ? options.getProject()
        : options.getPubsubProjectId();
  }

  public static boolean isValidAsciiString(String outputMessageMetadata) {
    if (outputMessageMetadata != null
        && !StandardCharsets.US_ASCII.newEncoder().canEncode(outputMessageMetadata)) {
      return false;
    }
    return true;
  }

  public static PipelineResult run(SpannerChangeStreamsToPubSubOptions options) {
    LOG.info("Requested Message Format is " + options.getOutputDataFormat());
    options.setStreaming(true);
    options.setEnableStreamingEngine(true);

    final Pipeline pipeline = Pipeline.create(options);
    // Get the Spanner project, instance, database, metadata instance, metadata database
    // change stream, pubsub topic, and pubsub api parameters.
    String spannerProjectId = getSpannerProjectId(options);
    String instanceId = options.getSpannerInstanceId();
    String databaseId = options.getSpannerDatabase();
    String metadataInstanceId = options.getSpannerMetadataInstanceId();
    String metadataDatabaseId = options.getSpannerMetadataDatabase();
    String changeStreamName = options.getSpannerChangeStreamName();
    String pubsubProjectId = getPubsubProjectId(options);
    String pubsubTopicName = options.getPubsubTopic();
    String pubsubAPI = options.getPubsubAPI();
    Boolean includeSpannerSource = options.getIncludeSpannerSource();
    String outputMessageMetadata = options.getOutputMessageMetadata();

    // Ensure outputMessageMetadata only contains valid ascii characters
    if (!isValidAsciiString(outputMessageMetadata)) {
      throw new RuntimeException("outputMessageMetadata contains non ascii characters.");
    }

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

    final RpcPriority rpcPriority = options.getRpcPriority();
    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()))
            .withProjectId(spannerProjectId)
            .withInstanceId(instanceId)
            .withDatabaseId(databaseId);
    // Propagate database role for fine-grained access control on change stream.
    if (options.getSpannerDatabaseRole() != null) {
      spannerConfig =
          spannerConfig.withDatabaseRole(
              ValueProvider.StaticValueProvider.of(options.getSpannerDatabaseRole()));
    }
    pipeline
        .apply(
            SpannerIO.readChangeStream()
                .withSpannerConfig(spannerConfig)
                .withMetadataInstance(metadataInstanceId)
                .withMetadataDatabase(metadataDatabaseId)
                .withChangeStreamName(changeStreamName)
                .withInclusiveStartAt(startTimestamp)
                .withInclusiveEndAt(endTimestamp)
                .withRpcPriority(rpcPriority)
                .withMetadataTable(metadataTableName))
        .apply(
            "Convert each record to a PubsubMessage",
            FileFormatFactorySpannerChangeStreamsToPubSub.newBuilder()
                .setOutputDataFormat(options.getOutputDataFormat())
                .setProjectId(pubsubProjectId)
                .setPubsubAPI(pubsubAPI)
                .setPubsubTopicName(pubsubTopicName)
                .setIncludeSpannerSource(includeSpannerSource)
                .setSpannerDatabaseId(databaseId)
                .setSpannerInstanceId(instanceId)
                .setOutputMessageMetadata(outputMessageMetadata)
                .build());
    return pipeline.run();
  }
}

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

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