Cloud Storage 中的文字檔案到 Pub/Sub (串流) 範本

這個範本可以建立串流管道，持續輪詢新上傳到 Cloud Storage 的文字檔案，並逐行讀取每個檔案，然後將字串發布至 Pub/Sub 主題。此外，這個範本還能以內含 JSON 記錄的換行符號分隔檔案或 CSV 檔案格式，將記錄發布到 Pub/Sub 主題進行即時處理；您也可以針對 Pub/Sub 重送資料。

管道會無限期持續執行，由於使用「Watch」轉換 (不支援排空的「SplittableDoFn」)，因此必須透過「cancel」手動終止，而非「drain」。

目前的輪詢間隔固定為 10 秒一次。這個範本不會對個別記錄設定任何時間戳記，因此在執行期間，事件時間會等於發布時間。假如您的管道需要準確的時間才能處理作業，則請勿使用這個管道。

這個範本受到 Beam 問題 18459 的影響，可能會在大型值區發生記憶體問題 (在含有 500 萬個檔案的值區中觀察到)。以下提供一些已知解決方法：

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」(依據範本建立工作) 頁面。
前往「依範本建立工作」
2. 在「Job name」(工作名稱) 欄位中，輸入專屬工作名稱。
3. 選用：如要使用區域端點，請從下拉式選單中選取值。預設區域為 us-central1。

   如需可執行 Dataflow 工作的區域清單，請參閱「Dataflow 位置」。

4. 從「Dataflow template」(Dataflow 範本) 下拉式選單中，選取「Text Files on Cloud Storage to Pub/Sub」(從 Cloud Storage 匯入文字檔到 Pub/Sub) (串流) 範本。
5. 在提供的參數欄位中輸入參數值。
6. 選用：如要從「僅需處理一次」切換至「至少一次串流模式」，請選取「At Least Once」(至少一次)。
7. 按一下「Run Job」(執行工作)。

gcloud

注意：如要使用 the Google Cloud CLI 執行 classic範本，您必須安裝 Google Cloud CLI 138.0.0 以上版本。

在殼層或終端機中執行範本：

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-REGION_NAME/VERSION/ \
    --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 bucket 的名稱
• FILE_PATTERN：要從 Cloud Storage bucket 讀取資料的檔案模式 glob (例如 path/*.csv)

API

如要使用 REST API 執行範本，請傳送 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/
{
   "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 專案 ID
• JOB_NAME：您選擇的不重複工作名稱
• LOCATION：您要部署 Dataflow 工作的區域，例如 us-central1
• STAGING_LOCATION：用於暫存本機檔案的位置 (例如 gs://your-bucket/staging)
• TOPIC_NAME：Pub/Sub 主題名稱
• BUCKET_NAME：Cloud Storage bucket 的名稱
• FILE_PATTERN：要從 Cloud Storage bucket 讀取資料的檔案模式 glob (例如 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 提供的範本清單。