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

Question

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

在向GCS写信之前，我想在向Gazli压气站写信之前处理来自pubsub的每条消息。

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

Answer 1

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

如果你想将其作为模板运行，你必须创建并暂存模板

要创建模板，您必须对其进行编译，建议使用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"

Answer 2

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

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