并行执行工作流步骤

并行步骤可以通过同时执行多个阻塞调用来缩短工作流的总执行时间。

sleep、HTTP 调用和callbacks等阻塞调用可能需要一些时间，从几毫秒到几天不等。并行步骤旨在帮助执行此类并发长时间运行的操作。如果工作流必须执行多个彼此独立的阻塞调用，则使用并行分支可同时启动调用并等待所有调用完成，从而缩短总执行时间。

例如，如果您的工作流必须从几个独立系统中检索客户数据才能继续操作，则并行分支允许并发 API 请求。如果有五个系统，并且每个系统都需要两秒钟的时间做出响应，那么按顺序执行工作流中的步骤至少需要 10 秒钟；而并行执行这些步骤可能只需两秒钟。

注意 ：并行分支对非阻塞步骤（例如赋值、条件或非阻塞调用）没有性能优势。由于系统开销，使用不包含阻塞调用的并行分支可能会增加总执行时间。

创建并行步骤

创建 parallel 步骤，用于定义可以同时执行两个或多个步骤的工作流的一部分。

YAML

  - PARALLEL_STEP_NAME:
      parallel:
        exception_policy: POLICY
        shared: [VARIABLE_A, VARIABLE_B, ...]
        concurrency_limit: CONCURRENCY_LIMIT
        BRANCHES_OR_FOR:
          ...

JSON

  [
    {
      "PARALLEL_STEP_NAME": {
        "parallel": {
          "exception_policy": "POLICY",
          "shared": [
            "VARIABLE_A",
            "VARIABLE_B",
            ...
          ],
          "concurrency_limit": "CONCURRENCY_LIMIT",
          "BRANCHES_OR_FOR":
          ...
        }
      }
    }
  ]

替换以下内容：

• PARALLEL_STEP_NAME：并行步骤的名称。
• POLICY（可选）：确定发生未处理的异常时其他分支将执行的操作。默认政策 continueAll 不会执行进一步操作，所有其他分支都将尝试运行。请注意，continueAll 是目前唯一支持的政策。
• VARIABLE_A、VARIABLE_B 等：具有父作用域的可写变量列表，允许在并行步骤中进行赋值。如需了解详情，请参阅共享变量。
• CONCURRENCY_LIMIT（可选）：在后续分支和迭代排队等待等待之前，在单个工作流执行中可以并发执行的分支和迭代的最大数量。这仅适用于单个 parallel 步骤，且不会级联。必须是正整数，可以是字面量值或表达式。如需了解详情，请参阅并发限制。
• BRANCHES_OR_FOR：使用 branches 或 for 指示以下其中一项：
  • 可以并发运行的分支。
  • 迭代可以并发运行的循环。

请注意以下几点：

• 并行分支和迭代可以按任何顺序运行，并且在每次执行时可能以不同的顺序运行。
• 并行步骤可包含其他嵌套并行步骤，直至达到深度上限。请参阅配额和限制。
• 如需了解详情，请参阅并行步骤的语法参考页面。

提示：Cloud Functions (第 2 代) 可以通过底层 Cloud Run 服务 利用并发功能，让您可以更好地控制价格。在并行 for 步骤中调用 Cloud Functions 函数时，您可以通过设置并发设置让同一 Cloud Function 实例处理多个项来节省费用。详细了解如何在 Cloud Run 中配置并发。

将实验性函数替换为并行步骤

如果您使用 experimental.executions.map 支持并行工作，则可以迁移工作流以改用并行步骤，从而并行执行普通的 for 循环。如需查看示例，请参阅将实验性函数替换为并行步骤。

示例

这些示例演示了语法。

并行执行操作（使用分支）

如果您的工作流有多组可同时执行的不同步骤，那么将它们放在并行分支中可以减少完成这些步骤所需的总时间。

在以下示例中，用户 ID 会作为参数传递给工作流，并且从两项不同的服务并行检索数据。共享变量允许将值写入分支中，并在分支完成后读取：

YAML

main:
  params: [input]
  steps:
    - init:
        assign:
          - userProfile: {}
          - recentItems: []
    - enrichUserData:
        parallel:
          shared: [userProfile, recentItems]  # userProfile and recentItems are shared to make them writable in the branches
          branches:
            - getUserProfileBranch:
                steps:
                  - getUserProfile:
                      call: http.get
                      args:
                        url: '${"https://example.com/users/" + input.userId}'
                      result: userProfile
            - getRecentItemsBranch:
                steps:
                  - getRecentItems:
                      try:
                        call: http.get
                        args:
                          url: '${"https://example.com/items?userId=" + input.userId}'
                        result: recentItems
                      except:
                        as: e
                        steps:
                          - ignoreError:
                              assign:  # continue with an empty list if this call fails
                                - recentItems: []

JSON

{
  "main": {
    "params": [
      "input"
    ],
    "steps": [
      {
        "init": {
          "assign": [
            {
              "userProfile": {}
            },
            {
              "recentItems": []
            }
          ]
        }
      },
      {
        "enrichUserData": {
          "parallel": {
            "shared": [
              "userProfile",
              "recentItems"
            ],
            "branches": [
              {
                "getUserProfileBranch": {
                  "steps": [
                    {
                      "getUserProfile": {
                        "call": "http.get",
                        "args": {
                          "url": "${\"https://example.com/users/\" + input.userId}"
                        },
                        "result": "userProfile"
                      }
                    }
                  ]
                }
              },
              {
                "getRecentItemsBranch": {
                  "steps": [
                    {
                      "getRecentItems": {
                        "try": {
                          "call": "http.get",
                          "args": {
                            "url": "${\"https://example.com/items?userId=\" + input.userId}"
                          },
                          "result": "recentItems"
                        },
                        "except": {
                          "as": "e",
                          "steps": [
                            {
                              "ignoreError": {
                                "assign": [
                                  {
                                    "recentItems": []
                                  }
                                ]
                              }
                            }
                          ]
                        }
                      }
                    }
                  ]
                }
              }
            ]
          }
        }
      }
    ]
  }
}

并行处理多项内容（使用并行循环）

如果您需要对列表中的每一项执行相同的操作，则可以使用并行循环更快地完成执行。并行循环允许并行执行多次循环迭代。请注意，与常规的 for 循环不同，迭代可以按任何顺序执行。

在以下示例中，一组用户通知在并行 for 循环中进行处理：

YAML

main:
  params: [input]
  steps:
    - sendNotifications:
        parallel:
          for:
            value: notification
            in: ${input.notifications}
            steps:
              - notify:
                  call: http.post
                  args:
                    url: https://example.com/sendNotification
                    body:
                      notification: ${notification}

JSON

{
  "main": {
    "params": [
      "input"
    ],
    "steps": [
      {
        "sendNotifications": {
          "parallel": {
            "for": {
              "value": "notification",
              "in": "${input.notifications}",
              "steps": [
                {
                  "notify": {
                    "call": "http.post",
                    "args": {
                      "url": "https://example.com/sendNotification",
                      "body": {
                        "notification": "${notification}"
                      }
                    }
                  }
                }
              ]
            }
          }
        }
      }
    ]
  }
}

汇总数据（使用并行循环）

您可以处理一组项，同时从对每一项执行的操作收集数据。例如，您可能想要跟踪已创建项的 ID，或维护存在错误的项的列表。

在以下示例中，对公共 BigQuery 数据集执行 10 个单独的查询，每个查询会返回一个文档或一组文档中的单词数。共享变量允许单词计数累积，并在所有迭代完成后读取。计算所有文档的字数之后，工作流会返回总数。

YAML

main:
  params: [input]
  steps:
    - init:
        assign:
          - numWords: 0
          - corpuses:
              - sonnets
              - various
              - 1kinghenryvi
              - 2kinghenryvi
              - 3kinghenryvi
              - comedyoferrors
              - kingrichardiii
              - titusandronicus
              - tamingoftheshrew
              - loveslabourslost
    - runQueries:
        parallel:  # numWords is shared so it can be written within the parallel loop
          shared: [numWords]
          for:
            value: corpus
            in: ${corpuses}
            steps:
              - runQuery:
                  call: googleapis.bigquery.v2.jobs.query
                  args:
                    projectId: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
                    body:
                      useLegacySql: false
                      query: ${"SELECT COUNT(DISTINCT word) FROM `bigquery-public-data.samples.shakespeare` " + " WHERE corpus='" + corpus + "' "}
                  result: query
              - add:
                  assign:
                    - numWords: ${numWords + int(query.rows[0].f[0].v)}  # first result is the count
    - done:
        return: ${numWords}

JSON

{
  "main": {
    "params": [
      "input"
    ],
    "steps": [
      {
        "init": {
          "assign": [
            {
              "numWords": 0
            },
            {
              "corpuses": [
                "sonnets",
                "various",
                "1kinghenryvi",
                "2kinghenryvi",
                "3kinghenryvi",
                "comedyoferrors",
                "kingrichardiii",
                "titusandronicus",
                "tamingoftheshrew",
                "loveslabourslost"
              ]
            }
          ]
        }
      },
      {
        "runQueries": {
          "parallel": {
            "shared": [
              "numWords"
            ],
            "for": {
              "value": "corpus",
              "in": "${corpuses}",
              "steps": [
                {
                  "runQuery": {
                    "call": "googleapis.bigquery.v2.jobs.query",
                    "args": {
                      "projectId": "${sys.get_env(\"GOOGLE_CLOUD_PROJECT_ID\")}",
                      "body": {
                        "useLegacySql": false,
                        "query": "${\"SELECT COUNT(DISTINCT word) FROM `bigquery-public-data.samples.shakespeare` \" + \" WHERE corpus='\" + corpus + \"' \"}"
                      }
                    },
                    "result": "query"
                  }
                },
                {
                  "add": {
                    "assign": [
                      {
                        "numWords": "${numWords + int(query.rows[0].f[0].v)}"
                      }
                    ]
                  }
                }
              ]
            }
          }
        }
      },
      {
        "done": {
          "return": "${numWords}"
        }
      }
    ]
  }
}

后续步骤

• 语法参考文档：并行步骤
• 教程：运行并行执行其他工作流的工作流
• 教程：并行运行多个 BigQuery 作业