AWS Step Functions如何将状态机输入与Lambda输出同时传入Map状态

解决方案 #

你需要通过Map状态的Parameters配置，将全局的run_id注入到每一个迭代子项的输入中，同时通过ItemsPath指定需要遍历的样本数组，具体修改如下：

1. 核心逻辑说明 #

进入Map状态时的完整数据已经同时包含原始输入的metadata.run_info样本数组和Lambda返回的populate_efs_result.Payload（即run_id），你只需要在Map层将两者合并后传给Iterator即可：

• 使用$$.Map.Item.Value引用当前遍历的单条样本数据
• 无需修改Iterator内部的Task配置，原有环境变量的引用逻辑可以直接复用

2. 修改后的Map状态定义 #

"MapState": {
  "Type": "Map",
  "End": true,
  "ResultPath": "$.metadata.run_result", // 建议改个路径避免覆盖原始样本数组，不需要输出的话可以设为null
  "ItemsPath": "$.metadata.run_info", // 指定遍历的样本数组路径
  "Parameters": {
    "run_id.$": "$.populate_efs_result.Payload", // 注入全局run_id
    "sample_name.$": "$$.Map.Item.Value.sample_name", // 取当前样本的名称
    "sequencer.$": "$$.Map.Item.Value.sequencer" // 取当前样本的测序仪信息
  },
  "Iterator": {
    "StartAt": "TaskState",
    "States": {
      "TaskState": {
        "Type": "Task",
        "End": true,
        "InputPath": "$",
        "ResultPath": null,
        "Resource": "arn:aws:states:::batch:submitJob.sync",
        "Parameters": {
          "JobDefinition": "arn:aws:batch:{ACCOUNT_INFO}:job-definition/BatchJobDefCfn:1",
          "JobName": "test",
          "JobQueue": "arn:aws:batch:{ACCOUNT_INFO}:job-queue/BatchQueue123",
          "ContainerOverrides": {
            "Command": [
              "sh",
              "-c",
              "entrypoint.pl -i /NGS/${sequencer}/${run_id}/ -s ${sample_name}"
            ],
            "Environment": [
              {
                "Name": "run_id",
                "Value.$": "$.run_id"
              },
              {
                "Name": "sample_name",
                "Value.$": "$.sample_name"
              },
              {
                "Name": "sequencer",
                "Value.$": "$.sequencer"
              }
            ]
          }
        }
      }
    }
  }
}

补充说明 #

如果后续需要给样本加更多属性，只需要在Map的Parameters里对应加字段即可，不需要修改Iterator内部逻辑。

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