תבנית של קובצי טקסט ב-Cloud Storage ל-Pub/Sub (סטרימינג)

התבנית הזו יוצרת צינור להזרמת נתונים שבודק באופן רציף אם יש קובצי טקסט חדשים שהועלו ל-Cloud Storage, קורא כל קובץ שורה אחר שורה ומפרסם מחרוזות בנושא Pub/Sub. התבנית מפרסמת רשומות בקובץ מופרד בתו שורה חדשה שמכיל רשומות JSON או קובץ CSV בנושא Pub/Sub לעיבוד בזמן אמת. אפשר להשתמש בתבנית הזו כדי להפעיל מחדש נתונים ב-Pub/Sub.

הצינור פועל ללא הגבלת זמן, וצריך להפסיק אותו ידנית באמצעות הפקודה 'ביטול' ולא באמצעות הפקודה 'התרוקנות', כי הוא משתמש בטרנספורמציה 'Watch', שהיא 'SplittableDoFn' שלא תומכת בהתרוקנות.

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

התבנית הזו מושפעת מבעיה מספר 18459 ב-Beam, וסביר להניח שיהיו בעיות בזיכרון בדליים גדולים (הבעיה נצפתה בדליים עם 5 מיליון קבצים). ריכזנו כאן כמה פתרונות עקיפים ידועים:

1. מפצלים את הקבצים לשני מאגרי מידע או יותר ומשתמשים בתבנית הזו.
2. יוצרים תבנית בהתאמה אישית, שאולי מבוססת על התבנית הזו, שמעבירה שמות של קבצים דרך Pub/Sub ומעבדת את הקבצים האלה.

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

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

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

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

• ‫inputFilePattern: דפוס קובץ הקלט לקריאה. לדוגמה, gs://bucket-name/files/*.json.
• ‫outputTopic: נושא הקלט של Pub/Sub שאליו ייכתבו הנתונים. השם צריך להיות בפורמט projects/<PROJECT_ID>/topics/<TOPIC_NAME>. לדוגמה, projects/your-project-id/topics/your-topic-name.

הפעלת התבנית

המסוף

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

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

4. בתפריט הנפתח Dataflow template (תבנית של העברת נתונים), בוחרים באפשרות the Text Files on Cloud Storage to Pub/Sub (Stream) template.
5. בשדות הפרמטרים שמופיעים, מזינים את ערכי הפרמטרים.
6. אופציונלי: כדי לעבור מעיבוד של כל נתון בדיוק פעם אחת אל מצב סטרימינג של כל נתון לפחות פעם אחת, בוחרים באפשרות לפחות פעם אחת.
7. לוחצים על הפעלת העבודה.

gcloud

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

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

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-REGION_NAME/VERSION/Stream_GCS_Text_to_Cloud_PubSub \
    --region REGION_NAME\
    --staging-location STAGING_LOCATION\
    --parameters \
inputFilePattern=gs://BUCKET_NAME/FILE_PATTERN,\
outputTopic=projects/PROJECT_ID/topics/TOPIC_NAME

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

• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫REGION_NAME: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫STAGING_LOCATION: המיקום של קבצים מקומיים להעברה (לדוגמה, gs://your-bucket/staging)
• ‫TOPIC_NAME: השם של נושא ה-Pub/Sub
• ‫BUCKET_NAME: שם הקטגוריה של Cloud Storage
• ‫FILE_PATTERN: תבנית ה-glob של הקובץ לקריאה מקטגוריה של Cloud Storage (לדוגמה, path/*.csv)

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/Stream_GCS_Text_to_Cloud_PubSub
{
   "jobName": "JOB_NAME",
   "environment": {
       "ipConfiguration": "WORKER_IP_UNSPECIFIED",
       "additionalExperiments": []
    },
   "parameters": {
       "inputFilePattern": "gs://BUCKET_NAME/FILE_PATTERN",
       "outputTopic": "projects/PROJECT_ID/topics/TOPIC_NAME"
   }
}

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

• ‫PROJECT_ID: מזהה הפרויקט שבו רוצים להריץ את משימת Dataflow Google Cloud
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫LOCATION: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫STAGING_LOCATION: המיקום של קבצים מקומיים להעברה (לדוגמה, gs://your-bucket/staging)
• ‫TOPIC_NAME: השם של נושא ה-Pub/Sub
• ‫BUCKET_NAME: שם הקטגוריה של Cloud Storage
• ‫FILE_PATTERN: תבנית ה-glob של הקובץ לקריאה מקטגוריה של Cloud Storage (לדוגמה, path/*.csv)

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

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.templates;

import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.templates.TextToPubsub.Options;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Watch;
import org.joda.time.Duration;

/**
 * The {@code TextToPubsubStream} is a streaming version of {@code TextToPubsub} pipeline that
 * publishes records to Cloud Pub/Sub from a set of files. The pipeline continuously polls for new
 * files, reads them row-by-row and publishes each record as a string message. The polling interval
 * is fixed and equals to 10 seconds. At the moment, publishing messages with attributes is
 * unsupported.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v1/README_Stream_GCS_Text_to_Cloud_PubSub.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Stream_GCS_Text_to_Cloud_PubSub",
    category = TemplateCategory.STREAMING,
    displayName = "Text Files on Cloud Storage to Pub/Sub",
    description = {
      "This template creates a streaming pipeline that continuously polls for new text files uploaded to Cloud Storage, reads each file line by line, and publishes strings to a Pub/Sub topic. "
          + "The template publishes records in a newline-delimited file containing JSON records or CSV file to a Pub/Sub topic for real-time processing. "
          + "You can use this template to replay data to Pub/Sub.\n",
      "The pipeline runs indefinitely and needs to be terminated manually via a <a href=\"https://cloud.google.com/dataflow/docs/guides/stopping-a-pipeline#cancel\">cancel</a> and not a <a href=\"https://cloud.google.com/dataflow/docs/guides/stopping-a-pipeline#drain\">drain</a>, due to its use of the <code>Watch</code> transform, which is a <code>SplittableDoFn</code> that does not support draining.\n",
      "Currently, the polling interval is fixed and set to 10 seconds. This template does not set any timestamp on the individual records, so the event time is equal to the publishing time during execution. "
          + "If your pipeline relies on an accurate event time for processing, you should not use this pipeline."
    },
    optionsClass = Options.class,
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/text-to-pubsub-stream",
    contactInformation = "https://cloud.google.com/support",
    requirements = {
      "Input files must be in newline-delimited JSON or CSV format. Records that span multiple lines in the source files can cause issues downstream, because each line within the files is published as a message to Pub/Sub.",
      "The Pub/Sub topic must exist prior to execution.",
      "The pipeline runs indefinitely and needs to be terminated manually.",
    },
    streaming = true,
    supportsAtLeastOnce = true,
    supportsExactlyOnce = true)
public class TextToPubsubStream {
  private static final Duration DEFAULT_POLL_INTERVAL = Duration.standardSeconds(10);

  /**
   * Main entry-point for the pipeline. Reads in the command-line arguments, parses them, and
   * executes the pipeline.
   *
   * @param args Arguments passed in from the command-line.
   */
  public static void main(String[] args) {

    // Parse the user options passed from the command-line
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);

    run(options);
  }

  /**
   * Executes the pipeline with the provided execution parameters.
   *
   * @param options The execution parameters.
   */
  public static PipelineResult run(Options options) {
    // Create the pipeline.
    Pipeline pipeline = Pipeline.create(options);

    /*
     * Steps:
     *  1) Read from the text source.
     *  2) Write each text record to Pub/Sub
     */
    pipeline
        .apply(
            "Read Text Data",
            TextIO.read()
                .from(options.getInputFilePattern())
                .watchForNewFiles(DEFAULT_POLL_INTERVAL, Watch.Growth.never()))
        .apply("Write to PubSub", PubsubIO.writeStrings().to(options.getOutputTopic()));

    return pipeline.run();
  }
}

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

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