接收PubSub通知触发Dataflow作业的实现方案咨询

完全可以通过Apache Beam内置的Pub/Sub IO实现该需求。你只需要将原管道中作为触发源的beam.Create替换为Pub/Sub读取源，同时将管道切换为流模式运行即可。管道启动后会持续监听指定的Pub/Sub资源，每收到一条消息就自动触发后续的整个处理链路。

实现示例 #

前置依赖 #

不需要额外安装第三方包，Apache Beam的GCP扩展包已经内置了Pub/Sub IO组件，Dataflow运行环境默认支持。

改造后代码 #

import apache_beam as beam
from apache_beam.io.gcp.pubsub import ReadFromPubSub
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions

# 原有业务DoFn保持不变即可
class dofn1(beam.DoFn):
    def process(self, element):
        # 若不需要Pub/Sub消息内容，可直接忽略element参数，仅把消息到达作为触发信号
        # 原有业务逻辑保持不变
        yield element

class dofn2(beam.DoFn):
    def process(self, element):
        # 原有业务逻辑
        yield element

class dofn3(beam.DoFn):
    def process(self, element):
        # 原有业务逻辑
        yield element

def run():
    pipeline_options = PipelineOptions()
    # 开启流模式，该配置为Pub/Sub IO运行的必填项
    pipeline_options.view_as(StandardOptions).streaming = True

    # 替换为你自己的Pub/Sub资源路径
    # 订阅路径格式：projects/你的GCP项目ID/subscriptions/你的订阅名称
    # 若直接使用主题作为源，格式为：projects/你的GCP项目ID/topics/你的主题名称
    PUBSUB_RESOURCE = "projects/your-project-id/subscriptions/your-trigger-subscription"

    with beam.Pipeline(options=pipeline_options) as p:
        _ = (
            p
            # 替换原有beam.Create触发源为Pub/Sub读取源
            | "Read trigger event from PubSub" >> ReadFromPubSub(subscription=PUBSUB_RESOURCE)
            | "job 1" >> beam.ParDo(dofn1())
            | "job 2" >> beam.ParDo(dofn2())
            | "job 3" >> beam.ParDo(dofn3())
        )

if __name__ == "__main__":
    run()

注意事项 #

• 部署到Dataflow运行时，需要保证管道使用的服务账号具备Pub/Sub订阅的读取权限，对应角色为roles/pubsub.subscriber
• 流管道会持续运行直到手动终止，每收到一条Pub/Sub消息就会执行一次完整的后续处理流程，和你之前的人工触发逻辑完全兼容
• 若不需要使用Pub/Sub消息携带的内容，发布触发消息时可以推送空内容，不影响后续逻辑运行

内容的提问来源于stack exchange，提问作者Alex