如何在 GCS 上自定义从 PubSub 到文本文件的 GCP 数据流答案

【问题标题】：How to customize GCP Dataflow from PubSub to Text file on GCS如何在 GCS 上自定义从 PubSub 到文本文件的 GCP 数据流
【发布时间】：2020-09-16 14:17:31
【问题描述】：

我们有一个模板 GCP 数据流从 PubSub 到 GCS 上的文本文件。

在写入 GCS 之前，我想在写入 GCS 之前处理来自 pubsub 的每条消息。

我在 GitHub 上看到了这个 code，但我不确定在哪里更新代码，有没有办法打印来自 PubSub 的每条消息？

【问题讨论】：

您使用什么语言来构建您的管道？

标签： google-cloud-dataflow

【解决方案1】：

来自 Github 的代码是一个数据流管道。您可以将管道作为非模板作业运行，通常是为了测试管道是否正常工作。

如果你想将它作为模板运行，你必须create and stage the template

要创建模板，您必须对其进行编译，建议使用maven 进行编译。

 mvn compile exec:java \
 -Dexec.mainClass=com.example.myclass \
 -Dexec.args="--runner=DataflowRunner \
              --project=YOUR_PROJECT_ID \
              --stagingLocation=gs://YOUR_BUCKET_NAME/staging \
              --templateLocation=gs://YOUR_BUCKET_NAME/templates/YOUR_TEMPLATE_NAME"

【讨论】：

【解决方案2】：

您可以通过在该代码中的PubSub Read 之后应用DoFn 来修改代码，例如

pipeline
    .apply("Read PubSub Events", PubsubIO.readStrings().fromTopic(options.getInputTopic()))
    .apply(ParDo.of([YourMessageProcessingDoFnHere]))
    .apply(
        options.getWindowDuration() + " Window",

【讨论】：