Text Files on Cloud Storage to Pub/Sub (Stream) 模板

此模板可以创建一种流处理流水线，该流水线可持续轮询新上传到 Cloud Storage 的文本文件，并逐行读取每个文件，然后将字符串发布到 Pub/Sub 主题。此外，此模板还能以包含 JSON 记录的换行符分隔文件或 CSV 文件形式，将记录发布到 Pub/Sub 主题进行实时处理。您可以使用此模板将数据重放到 Pub/Sub。

流水线无限期运行，需要通过“取消”而非“排空”手动终止，原因是其使用“Watch”转换，该转换是一个不支持排空的“SplittableDoFn”。

目前，轮询间隔固定为 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 基于模板创建作业页面。
转到“基于模板创建作业”
2. 在作业名称字段中，输入唯一的作业名称。
3. 可选：对于区域性端点，从下拉菜单中选择一个值。默认区域为 us-central1。

   如需查看可以在其中运行 Dataflow 作业的区域列表，请参阅 Dataflow 位置。

4. 从 Dataflow 模板下拉菜单中，选择 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 或更高版本。

在 shell 或终端中，运行模板：

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：要从 Cloud Storage 存储桶中读取的文件格式 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/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 项目的 ID
• JOB_NAME：您选择的唯一性作业名称
• LOCATION：要在其中部署 Dataflow 作业的区域，例如 us-central1
• STAGING_LOCATION：暂存本地文件的位置（例如 gs://your-bucket/staging）
• TOPIC_NAME：您的 Pub/Sub 主题名称
• BUCKET_NAME：Cloud Storage 存储桶的名称
• FILE_PATTERN：要从 Cloud Storage 存储桶中读取的文件格式 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 提供的模板列表。