如何在Google Cloud Functions工作流中仅触发Function3一次

方案一：用Firestore/Datastore做状态追踪 #

通过轻量级数据库记录每个文件的加载状态，在所有状态完成后触发Function3，适合需要自定义状态逻辑的场景。

• 步骤：

  1. 给文件打批次标识：Function1上传文件时，为每个文件添加自定义元数据（比如batch_id），或者在文件名中包含唯一的批次ID（比如batch_123_file1.csv），用来关联同批次的文件。
  2. 更新状态记录：Function2完成BigQuery加载后，在Firestore中更新对应批次下该文件的状态为completed。
  3. 检查全量完成状态：每次更新状态后，查询当前批次下所有文件的状态，如果3个文件都标记为完成，触发Function3，并标记该批次为已处理（避免重复触发）。

• 代码示例（Python）：

import firebase_admin
from firebase_admin import firestore
import requests

# 初始化Firestore
firebase_admin.initialize_app()
db = firestore.client()

def function2(event, context):
    # 处理文件加载到BigQuery的逻辑（此处省略核心加载代码）
    # ...

    # 从事件中获取批次ID和文件名
    file_metadata = event.get('metadata', {})
    batch_id = file_metadata.get('batch_id')
    file_name = event['name']

    if not batch_id:
        # 从文件名解析批次ID（示例格式：batch_123_file1.csv）
        batch_id = file_name.split('_')[1]

    # 使用事务确保状态更新的原子性
    with db.transaction() as transaction:
        batch_ref = db.collection('load_tracking').document(batch_id)
        doc = transaction.get(batch_ref)
        
        if not doc.exists:
            # 首次写入时初始化文件状态、总数量和处理标记
            transaction.set(batch_ref, {
                'files': {file_name: 'completed'},
                'total_files': 3,
                'processed': False
            })
        else:
            current_files = doc.to_dict().get('files', {})
            current_files[file_name] = 'completed'
            transaction.update(batch_ref, {'files': current_files})
        
        # 检查是否所有文件完成且未处理过
        updated_doc = transaction.get(batch_ref)
        doc_data = updated_doc.to_dict()
        completed_count = sum(1 for s in doc_data['files'].values() if s == 'completed')
        
        if completed_count == doc_data['total_files'] and not doc_data['processed']:
            # 触发Function3
            requests.post('https://REGION-PROJECT_ID.cloudfunctions.net/function3')
            # 标记批次已处理，防止重复触发
            transaction.update(batch_ref, {'processed': True})

方案二：用Cloud Workflows编排全流程 #

直接用Cloud Workflows替代零散的触发逻辑，由Workflow负责编排整个流程的顺序和等待，适合标准化的流程场景，无需自己维护状态。

• 步骤：

  1. 创建Workflow定义，按顺序执行以下步骤：
     • 调用Function1完成文件上传
     • 等待GCS存储桶中3个文件全部存在
     • 并行调用3个Function2（每个对应一个文件），等待全部执行完成
     • 调用Function3执行后续查询逻辑
  2. 将原来的Cloud Scheduler任务改为触发这个Workflow，而非直接调用Function1。

• Workflow YAML示例：

main:
  steps:
    # 调用Function1上传文件
    - invoke_function1:
        call: http.post
        args:
          url: https://REGION-PROJECT_ID.cloudfunctions.net/function1
          auth:
            type: OIDC
    
    # 等待3个文件全部出现在GCS桶中，重试机制确保文件完全上传
    - wait_for_all_files:
        call: sys.sleep
        args:
          seconds: 5
        retry:
          predicate: ${
            not(
              exists(google.cloud.storage.object.get(bucket="YOUR_BUCKET_NAME", object="file1.csv")) and
              exists(google.cloud.storage.object.get(bucket="YOUR_BUCKET_NAME", object="file2.csv")) and
              exists(google.cloud.storage.object.get(bucket="YOUR_BUCKET_NAME", object="file3.csv"))
            )
          }
          max_retries: 6
          backoff:
            initial_delay: 5
            multiplier: 1.5
    
    # 并行执行3个Function2，提高处理效率
    - run_all_function2:
        parallel:
          - process_file1:
              call: http.post
              args:
                url: https://REGION-PROJECT_ID.cloudfunctions.net/function2
                auth:
                  type: OIDC
                body:
                  file_name: "file1.csv"
          - process_file2:
              call: http.post
              args:
                url: https://REGION-PROJECT_ID.cloudfunctions.net/function2
                auth:
                  type: OIDC
                body:
                  file_name: "file2.csv"
          - process_file3:
              call: http.post
              args:
                url: https://REGION-PROJECT_ID.cloudfunctions.net/function2
                auth:
                  type: OIDC
                body:
                  file_name: "file3.csv"
    
    # 触发Function3执行后续查询
    - invoke_function3:
        call: http.post
        args:
          url: https://REGION-PROJECT_ID.cloudfunctions.net/function3
          auth:
            type: OIDC

方案三：基于BigQuery事件的状态跟踪 #

利用BigQuery的审计日志事件，跟踪三个目标表的加载完成状态，当所有表都完成加载后触发Function3，适合依赖BigQuery加载状态的场景。

• 步骤：
  1. 为BigQuery配置Cloud Audit Logs，将jobs.jobCompleted事件导出到一个Pub/Sub主题。
  2. 创建Pub/Sub订阅，添加过滤条件，只接收目标三个表的LOAD类型作业完成事件（过滤表达式示例）：

     protoPayload.methodName="jobs.insert" AND protoPayload.serviceData.jobCompletedEvent.job.jobType="LOAD_JOB" AND (protoPayload.resourceName:"projects/YOUR_PROJECT/datasets/YOUR_DATASET/tables/TABLE_1" OR protoPayload.resourceName:"projects/YOUR_PROJECT/datasets/YOUR_DATASET/tables/TABLE_2" OR protoPayload.resourceName:"projects/YOUR_PROJECT/datasets/YOUR_DATASET/tables/TABLE_3")
     

  3. 编写一个状态跟踪函数（订阅该Pub/Sub主题），每次收到事件就记录对应表的完成状态，当三个表都完成时触发Function3。

通用注意事项 #

• 无论采用哪种方案，都要给Function3做幂等设计，确保即使因网络波动等原因重复触发，也不会产生重复数据或业务错误。
• 状态跟踪类方案要定期清理过期的状态记录，避免数据库或存储资源浪费。

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