python - 在 CI 管道中部署 Dataflow

Question

我使用 Beam SDK 在 Python 中编写了流式 Google Dataflow 管道。有关于我如何在本地运行它并设置 -runner 标志以在 Dataflow 上运行它的文档。

我现在正在尝试将其自动部署到 CI 管道(bitbucket 管道，但并不真正相关)。有关于如何“运行”管道的文档，但没有真正“部署”它的文档。我测试过的命令如下所示:

python -m dataflow --runner "DataflowRunner" \
                   --jobName "<jobName>" \
                   --topic "<pub-sub-topic"" \
                   --project "<project>" \
                   --dataset "<dataset>" \
                   --worker_machine_type "n1-standard-2" \
                   --temp_location "gs://<bucket-name>/tmp/"

这将运行该作业，但因为它正在流式传输，所以它永远不会返回。它还在内部管理打包和推送到桶中。我知道如果我终止该进程它会继续运行，但是在 CI 服务器上设置它的方式我可以检测进程是否真的成功或者我只是在超时后终止它是很困难的。

这看起来很荒谬，好像我遗漏了一些明显的东西，但是我如何以一种我可以可靠地知道它是从 CI 管道部署的方式在数据流上打包和运行这个模块？

Answer 1

所以是的，这很愚蠢。

基本上当你使用

with beam.Pipeline(options=options) as p: 

语法，在底层调用wait_until_finish .所以等待在我没有意识到的情况下被调用，导致它永远徘徊。重构以删除上下文管理器可以解决问题。