AWS Step Function - 动态并行 MaxConcurrency 字段

Question

我们使用带有 Map 状态的 Step 函数动态并行来实现并发。 是否可以在 map state stepfunctions 中从上游任务（lambda 或从文件读取）将值传递给“MaxConcurrency”字段。

当前代码：

"Type": "Map",
"InputPath": "$.detail",
"ItemsPath": "$.shipped",
"MaxConcurrency": 3,
"ResultPath": "$.detail.shipped",

期望（将输入从 lambda 任务或读取文件任务传递给 MaxConcurrency）：

 "Type": "Map",
 "InputPath": "$.detail",
 "ItemsPath": "$.shipped",
 "MaxConcurrency": "$.input",
 "ResultPath": "$.detail.shipped"

因为它只支持整数而出现错误。

Answer 1

您可以在状态机定义时设置maxConcurrency，但不能在执行时设置。如您所见，Map 的maxConcurrency 需要number，但状态机语言使用strings 动态传递变量。

(注意：Step Functions 默认是并发的。Docs:maxConcurrency's "默认值为 0，这对并行性没有配额，并且迭代被同时调用尽可能”。）

选项1：选择+地图（难度：低）

限制在执行时动态的解决方法是Choice state，它根据输入变量分支到离散映射状态。每个分支的 Map 都有一个不同的maxConcurrency，但在其他方面是相同的。添加您需要的任何离散maxConcurrency 选项。 Choice 还接受默认情况以捕获不匹配的选择输入。

// execution input
{
  "concurrency": 5,
  "jobs": [ { "jobId": 1 }, { "jobId": 2 }, { "jobId": 3 }, { "jobId": 4}]
}

// state machine definition (partial)
"States": {
  "Max-Concurrency-Choice": {
    "Type": "Choice",
    "Choices": [
      {
        "Variable": "$.concurrency",
        "NumericEquals": 5,
        "Next": "MapState-MaxConcurrency-5"  // maxConcurrency in this branch is set at 5
      },
      {
        "Variable": "$.concurrency",
        "NumericEquals": 10,
        "Next": "MapState-MaxConcurrency-10" // maxConcurrency in this branch is set at 10
      }
    ],
    "Default": "MapState-MaxConcurrency-1" // maxConcurrency in the default branch is set at 1
  },

选项 2：嵌套 Sfn + API 调用（难度：高）

将您的 Sfn 嵌套在新的 Sfn 中。新的父 Sfn 在输入中采用 maxConcurrency。它有两个任务：

1. 在 Lambda 任务中，调用 UpdateStateMachine API，并为您当前的子 Sfn 使用新的字符串化 JSON 状态机定义。
2. 调用您当前的状态机。 Sfn 将拥有新的maxConurrency。