在 CI 管道中部署数据流答案

【问题标题】：Deploying Dataflow in a CI pipeline在 CI 管道中部署数据流
【发布时间】：2018-11-02 05:31:52
【问题描述】：

我使用 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 管道部署的方式在数据流上打包和运行这个模块？

【问题讨论】：

不应该部署为模板吗？

标签： python google-cloud-dataflow

【解决方案1】：

所以是的，这很愚蠢。

基本上当你使用

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

语法，在后台调用wait_until_finish。所以在我没有意识到的情况下调用了等待，导致它永远挂起。重构删除上下文管理器解决了这个问题。

【讨论】：

【解决方案2】：

扩展jamielennox's answer。

在本地开发环境的直接运行器上，您希望看到管道无限期地运行；也许只是在一段时间后用 Ctrl-C 手动取消。

在部署管道以在 GCP 的数据流上运行时，您希望脚本部署作业并结束。

runner_name = pipeline_options.get_all_options().get('runner')

if runner_name == 'DirectRunner':
    with beam.Pipeline(options=pipeline_options) as pipeline:
        _my_setup_pipeline(config, pipeline, subscription_full_name)

elif runner_name == 'DataflowRunner':
    pipeline = beam.Pipeline(options=pipeline_options)
    _my_setup_pipeline(config, pipeline, subscription_full_name)
    pipeline.run()

else:
    raise Exception(f'Unknown runner: {runner_name}')

【讨论】：