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

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

הערה: מחבר Spanner change streams לא תומך בהפסקת העבודה של משימה. כדי להפסיק את צינור הנתונים, מבטלים את העבודה במקום לנקז אותו. הפסקת העבודה של משימה באמצעות מחבר של Spanner לשינוי נתונים יכולה לגרום לכך שהמשימה תיתקע במצב של הפסקת עבודה. מידע נוסף זמין במאמר בנושא Draining במסמכי 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 change stream שממנו ייקראו הנתונים.
• ‫pubsubTopic: נושא Pub/Sub לפלט של זרמי שינויים.

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

• ‫spannerProjectId: הפרויקט שממנו ייקראו נתוני ה-Change Streams. בפרויקט הזה נוצר גם טבלת המטא-נתונים של המחבר של זרמי השינויים. ברירת המחדל של הפרמטר הזה היא הפרויקט שבו צינור ה-Dataflow פועל.
• ‫spannerDatabaseRole: תפקיד מסד הנתונים ב-Spanner שבו יש להשתמש כשמריצים את התבנית. הפרמטר הזה נדרש רק אם משתמש ה-IAM שמריץ את התבנית הוא משתמש עם בקרת גישה פרטנית. לתפקיד במסד הנתונים צריכה להיות הרשאת SELECT בפיד השינויים והרשאת EXECUTE בפונקציית הקריאה של פיד השינויים. מידע נוסף זמין במאמר בנושא בקרת גישה ברמת גרנולריות גבוהה לזרמי שינויים (https://cloud.google.com/spanner/docs/fgac-change-streams).
• ‫spannerMetadataTableName: השם של טבלת המטא-נתונים של מחבר Spanner change streams שבה רוצים להשתמש. אם לא מספקים את הטבלה, מערכת Spanner יוצרת אותה באופן אוטומטי במהלך השינוי של צינור עיבוד הנתונים. כשמעדכנים צינור קיים, צריך לציין את הפרמטר הזה. אין להשתמש בפרמטר הזה במקרים אחרים.
• ‫startTimestamp: תאריך ושעת ההתחלה (https://tools.ietf.org/html/rfc3339), כולל, לשימוש בקריאת זרמי שינויים. לדוגמה, ex- 2021-10-12T07:20:50.52Z. ברירת המחדל היא חותמת הזמן של תחילת הצינור, כלומר השעה הנוכחית.
• ‫endTimestamp: חותמת הזמן של סיום התקופה (בפורמט DateTime‏ 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) לכל התווים המיוחדים (למשל: מירכאות כפולות).
• ‫useSpannerEmulatorHost: האם להשתמש במארח Spanner שהוגדר ב-getSpannerHost() כמארח האמולטור. ברירת המחדל היא: false.
• ‫spannerChangeStreamTvfNameList: רשימה של שמות של Spanner Change Stream TVF, מופרדים באמצעות נקודה-פסיק, שאותם רוצים לשאול ולאחד. ברירת המחדל היא ריק.

הרצת התבנית

המסוף

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 Pub/Sub (העברת נתוני שינויים מ-Cloud Spanner ל-Pub/Sub).
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,\
    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

כדי להריץ את התבנית באמצעות 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",
            "pubsubTopic": "PUBSUB_TOPIC"
        },
        "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: מזהה מכונת 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.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.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();

    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()
            .withProjectId(spannerProjectId)
            .withInstanceId(instanceId)
            .withDatabaseId(databaseId);

    if (options.getUseSpannerEmulatorHost() != null && options.getUseSpannerEmulatorHost()) {
      spannerConfig =
          spannerConfig.withEmulatorHost(
              ValueProvider.StaticValueProvider.of(options.getSpannerHost()));
    } else {
      spannerConfig =
          spannerConfig.withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()));
    }

    // 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)
                .withTvfNameList(tvfNameList))
        .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 סיפקה.