在Dataflow流水线处理完所有数据后执行操作

批处理Dataflow作业处理完所有数据后是否可以执行操作？具体来说，我想将刚刚处理的流水线的文本文件移动到不同的GCS存储桶中。我不确定将数据放入管道中的位置，以确保在数据处理完成后执行一次。在Dataflow流水线处理完所有数据后执行操作

来源

2017-06-01 user01380121

我不明白为什么你需要做这个后管道执行。您可以使用侧面输出将文件写入多个存储桶，并在管道完成后保存自己的副本。

如果这不适合你（无论什么原因），那么你可以简单地在blocking execution模式下运行你的管道，即使用pipeline.run().waitUntilFinish()，然后在那之后写下你的代码的其余部分。

[..] 
/do some stuff before the pipeline runs 
Pipeline pipeline = ... 
pipeline.run().waitUntilFinish(); 
//do something after the pipeline finishes here 
[..]

来源

2017-06-02 12:21:26

谢谢。使用'BlockingDataflowPipelineRunner'运行这个工作就可以实现。 'waitUntilFinish（）'似乎在1.x Java API中不可用。 – user01380121

正确，不是。您在1.x中使用Blocking runner和wait/poll –

-1

我觉得两个选项可以帮助你在这里：

1）使用TextIO写入桶或你想要的，指定的确切GCS路径（如GS文件夹：//沙箱/其他桶）

2）使用Object Change Notifications结合Cloud Functions。你可以在这个here和JS的GCS SDK中找到一个很好的入门书。在这个选项中你将做的事情基本上是在某物桶落下时设置一个触发器，并使用自己写的云功能将其移动到另一个。

来源

2017-06-02 09:50:54

在Dataflow流水线处理完所有数据后执行操作

回答

相关问题