在 Google DataFlow 上完成管道后运行函数/管道

Question

我正在尝试在 Beam 管道完成后在 Google DataFlow 上运行一个函数（或管道）。

目前我已经构建了一个 hack 来运行该函数，方法是将上一个 Pipeline 的结果写入 null with

_ = existing_pipeline | "do next task" >> beam.Map(func)

...函数在哪里：

def func(_):
    # do some work, and ignore `_`

但是有更好的方法吗？

Answer 1

假设您希望函数在您的机器上而不是在云端运行，您应该执行以下操作：

result = existing_pipeline.run()
result.wait_until_finish()
# do some work