在 CI 管道中部署 Dataflow

Question

我使用 beam SDK 在 python 中编写了一个流式 Google Dataflow 管道。有文档介绍了如何在本地运行它并设置标志-runner以在数据流上运行它。

我现在正在尝试将其自动部署到 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。因此，在我没有意识到的情况下，等待被调用，导致它永远徘徊。重构以删除上下文管理器可以解决该问题。