在CI管道中部署数据流

Question

我使用beam SDK用python编写了一个Google数据流管道。有一些文档介绍了我如何在本地运行它，并设置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

扩展Jamielennox的答案。

在本地开发环境的直接运行器上时，您希望看到管道无限期运行；可能只是在一段时间后使用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}')

Answer 2

是的，这是一件愚蠢的事情。

基本上当你使用

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

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