如何将Step Functions Map状态的输入传递至内部Lambda函数？

问题分析 #

当使用Map状态的ItemReader读取S3 CSV文件时，默认每个迭代的输入仅为CSV的单行数据，Map状态的原始输入（如bucket_name、processable_key_name）不会自动传递给内部Lambda函数。你之前尝试的$$.Map.Item.Value仅指向单行数据，自然拿不到原始参数；直接在Lambda中获取map_input的方式未利用Step Functions上下文变量机制，所以也无效。

解决方案 #

通过Step Functions的上下文变量和参数映射，将Map状态的原始输入与CSV单行数据合并后传递给内部Lambda，以下是两种可行修改方案：

方案1：在Lambda调用时直接合并参数 #

修改FirstMapTask的Parameters配置，构建包含单行数据和原始参数的Payload：

"FirstMapTask": {
  "Type": "Task",
  "Resource": "arn:aws:states:::lambda:invoke",
  "ResultSelector": {
    "name.$": "$.Payload.name",
    "email.$": "$.Payload.email"
  },
  "ResultPath": "$.output",
  "Parameters": {
    "Payload": {
      "row_data.$": "$",
      "bucket_name.$": "$$.Map.Input.bucket_name",
      "processable_key_name.$": "$$.Map.Input.processable_key_name"
    },
    "FunctionName": "arn:aws:lambda:us-east-1:1234566632:function:stepfunctions-serverless-framework-dev-secondFunction"
  },
  "End": true
}

• $$.Map.Input是Step Functions上下文变量，指向Map状态收到的原始输入（即InitialState的输出）。
• 修改后第二个Lambda的Payload中将同时包含row_data（CSV单行数据）和两个原始参数。

方案2：在Map状态层面合并输入参数 #

在mapped_task中添加Parameters配置，将原始输入与单行数据合并，确保每个迭代的输入包含所有所需信息：

"mapped_task": {
  "Comment": "This state will be handling the payout of each event",
  "Type": "Map",
  "Parameters": {
    "row_data.$": "$$.Map.Item.Value",
    "bucket_name.$": "$.bucket_name",
    "processable_key_name.$": "$.processable_key_name"
  },
  "ItemReader": {
    "Resource": "arn:aws:states:::s3:getObject",
    "ReaderConfig": {
      "InputType": "CSV",
      "CSVHeaderLocation": "FIRST_ROW"
    },
    "Parameters": {
      "Bucket.$": "$.bucket_name",
      "Key.$": "$.processable_key_name"
    }
  },
  "ItemProcessor": {
    "ProcessorConfig": {
      "Mode": "DISTRIBUTED",
      "ExecutionType": "STANDARD"
    },
    "StartAt": "FirstMapTask",
    "States": {
      "FirstMapTask": {
        "Type": "Task",
        "Resource": "arn:aws:states:::lambda:invoke",
        "ResultSelector": {
          "name.$": "$.Payload.name",
          "email.$": "$.Payload.email"
        },
        "ResultPath": "$.output",
        "Parameters": {
          "Payload.$": "$",
          "FunctionName": "arn:aws:lambda:us-east-1:1234566632:function:stepfunctions-serverless-framework-dev-secondFunction"
        },
        "End": true
      }
    }
  },
  "Label": "Processabledata",
  "MaxConcurrency": 40,
  "ToleratedFailurePercentage": 100,
  "End": true
}

• 此方案中，Map状态会先将原始参数和单行数据合并为一个对象，再传递给ItemProcessor，后续Lambda读取整个Payload即可获取所有数据。

验证 #

修改状态机配置后启动执行，第二个Lambda的输入Payload中应包含bucket_name、processable_key_name和CSV单行数据，即可解决问题。

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