如何并行运行Google Workflows批量处理Cloud Firestore文档

完全可以实现所有文档转换操作的并行执行，替换现有串行模式。Google Workflows 原生提供并行步骤能力，刚好适配你这种无依赖的批量任务场景。

具体实现方式 #

你不需要调整现有单文档转换的核心逻辑，只需要把当前串行遍历文档的for循环，替换为并行迭代配置即可：

• 保留原有拉取Cloud Firestore目标集合全量文档的步骤，先拿到所有待处理的文档列表
• 移除原有的串行for循环块，改用parallel包裹的并行for迭代结构，Workflows会为每个文档自动分配独立的执行分支，分支之间互不阻塞
• 在并行迭代的步骤内，直接调用你已经写好的单文档转换逻辑即可

对应的最小YAML参考配置如下：

main:
  steps:
    - fetchDocs:
        assign:
          # 替换为你实际查询Firestore集合、获取待处理文档列表的逻辑
          - targetDocs: ${firestore.list("projects/你的项目ID/databases/(default)/documents/你的集合名").documents}
    - parallelTransform:
        parallel:
          # 建议根据下游服务配额设置合理并发上限，避免触发限流
          concurrency: 40
          # 如果需要汇总所有文档的处理结果，把接收结果的变量名加到shared数组里
          shared: []
          for:
            value: currentDoc
            in: ${targetDocs}
            steps:
              - runTransform:
                  # 替换为你原有单文档转换的调用逻辑，比如调用Cloud Function、执行HTTP请求等
                  call: yourSingleDocTransformLogic
                  args:
                    doc: ${currentDoc}

注意事项 #

• 不要设置无上限并发：如果你的转换逻辑需要调用其他Google Cloud服务API、第三方接口，一定要根据对应服务的QPS配额设置concurrency值，避免大量并发请求触发限流报错
• 并行分支的执行上下文默认互相隔离，如果需要统计处理成功/失败的文档数、汇总转换结果，必须提前把对应的结果变量声明在shared列表中，否则分支内对变量的修改不会同步到主流程
• 并行模式不会额外增加Workflows的计费成本：Workflows按实际执行的步骤数计费，并行执行只是把多步骤的执行从串行改为并行，总步骤数和串行一致，还能大幅缩短整个任务的总运行时长
• 如果单批次待处理文档量超过1000份，建议先按文档ID做哈希分片，再用并行块处理不同分片，避免触达单工作流的执行资源上限

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