如何在Dataflow Pipeline结束时触发事件？Cloud Function场景咨询

完全理解你的痛点——Cloud Function的超时限制确实卡死了同步等待Dataflow作业的路子，官方其实有现成的异步事件驱动方案，核心就是利用Dataflow的作业状态通知机制结合Pub/Sub，不用在CF里死等waitUntilFinish()。下面分两种官方认可的方案细说：

方案一：利用Dataflow原生作业状态通知（无需修改模板） #

这是最省心的官方推荐方案，Dataflow支持在提交作业时指定一个Pub/Sub主题，当作业进入终态（成功完成、失败、取消等）时，会自动发送一条状态消息到这个主题，你只需要写一个监听该主题的Cloud Function来执行后续操作。

步骤1：提交Dataflow模板时配置通知主题 #

不管是用gcloud命令还是API提交作业，添加--notificationPubsubTopic参数（gcloud里是--notification-pubsub-topic）指定你的Pub/Sub主题路径：

gcloud dataflow jobs run my-dataflow-job \
  --gcs-location gs://my-bucket/templates/my-custom-template \
  --param input=gs://my-bucket/input \
  --param output=gs://my-bucket/output \
  --notification-pubsub-topic projects/my-project/topics/dataflow-job-status

⚠️ 注意：要确保Dataflow的默认服务账号（service-<PROJECT_NUMBER>@dataflow-service-producer-prod.iam.gserviceaccount.com）拥有Pub/Sub的roles/pubsub.publisher权限，否则无法发送通知消息。

步骤2：编写监听Pub/Sub的Cloud Function #

创建一个触发类型为Pub/Sub的Cloud Function，监听上面指定的主题，当收到JOB_STATE_DONE的消息时执行后续逻辑。消息是JSON格式的，核心字段包括jobId、state、projectId等：

import base64
import json

def on_dataflow_job_completed(event, context):
    # 解析Pub/Sub消息
    message_data = base64.b64decode(event['data']).decode('utf-8')
    job_status = json.loads(message_data)
    
    # 只处理成功完成的作业
    if job_status['state'] == 'JOB_STATE_DONE':
        job_id = job_status['jobId']
        print(f"Dataflow作业 {job_id} 已成功完成，开始执行后续操作...")
        # 这里写你的后续逻辑：比如数据校验、触发另一个流水线、发送Slack通知等
    elif job_status['state'] == 'JOB_STATE_FAILED':
        # 可选：处理作业失败的告警逻辑
        print(f"Dataflow作业 {job_status['jobId']} 执行失败，触发告警...")

方案二：在自定义Dataflow模板中添加终端消息发送（更灵活） #

如果你需要在作业完成时传递自定义数据（比如处理的记录数、输出文件路径等），可以修改你的Dataflow模板代码，在Pipeline的最后一个步骤添加一个发送Pub/Sub消息的Transform。

示例（Python版Dataflow代码） #

在Pipeline的末尾添加一个DoFn来发送完成消息：

import apache_beam as beam
from apache_beam.io import WriteToText
from apache_beam.io.gcp.pubsub import PublishToPubSub
import json

class SendCompletionMessage(beam.DoFn):
    def process(self, element):
        # 自定义消息内容，比如带上处理结果统计
        completion_msg = {
            "processed_records": element,
            "status": "SUCCESS"
        }
        yield json.dumps(completion_msg).encode('utf-8')

with beam.Pipeline() as p:
    # 你的主Pipeline逻辑
    processed_count = (
        p
        | "读取输入数据" >> beam.io.ReadFromText("gs://input/path")
        | "处理数据" >> beam.Map(your_processing_function)
        | "写入输出" >> WriteToText("gs://output/path")
        | "统计记录数" >> beam.combiners.Count.Globally()
    )
    
    # 发送完成消息到Pub/Sub
    processed_count | "发送完成通知" >> PublishToPubSub(topic="projects/my-project/topics/dataflow-completion")

这种方案的优势是能把作业的自定义数据传递给后续流程，但需要修改你的Dataflow模板代码。

为什么这两种方案能避开Cloud Function时长限制？ #

两种方案都是异步事件驱动：第一个Cloud Function只负责提交Dataflow作业，提交完成就结束；后续操作由第二个Cloud Function在收到Pub/Sub消息时触发，完全不受第一个CF的超时限制。

内容的提问来源于stack exchange，提问作者Rafaël