在 Step Functions 中使用 Lambda 函数处理各个项目

• 使用以下定义通过 Step Functions 控制台创建状态机。有关创建状态机的信息，请参阅开始使用分布式 Map 状态教程中的第 1 步：创建工作流原型。

  在此状态机中，您将定义一个分布式 Map 状态，该状态接受 10 个整数的数组作为输入，并将这些数组项分批传递给子工作流执行。每个子工作流程执行都会接收一批三个项目作为输入，并运行一个内联 Map 状态。内联 Map 状态的每次迭代都会调用 Lambda 函数，并将批次中的一个项目传递给该函数。然后，此函数将该项与系数 7 相乘并返回结果。

  每个子工作流程执行的输出都是一个 JSON 数组，其中包含每个所传递项目的乘法结果。

  重要

  确保将以下代码中 Lambda 函数的 Amazon 资源名称 (ARN) 替换为您将在第 2 步中创建的函数的 ARN。

  {
    "StartAt": "Pass",
    "States": {
      "Pass": {
        "Type": "Pass",
        "Next": "Map",
        "Result": {
          "MyMultiplicationFactor": 7,
          "MyItems": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
        }
      },
      "Map": {
        "Type": "Map",
        "ItemProcessor": {
          "ProcessorConfig": {
            "Mode": "DISTRIBUTED",
            "ExecutionType": "STANDARD"
          },
          "StartAt": "InnerMap",
          "States": {
            "InnerMap": {
              "Type": "Map",
              "ItemProcessor": {
                "ProcessorConfig": {
                  "Mode": "INLINE"
                },
                "StartAt": "Lambda Invoke",
                "States": {
                  "Lambda Invoke": {
                    "Type": "Task",
                    "Resource": "arn:aws:states:::lambda:invoke",
                    "OutputPath": "$.Payload",
                    "Parameters": {
                      "Payload.$": "$",
                      "FunctionName": "arn:aws:lambda:us-east-1:123456789012:function:functionName"
                    },
                    "Retry": [
                      {
                        "ErrorEquals": [
                          "Lambda.ServiceException",
                          "Lambda.AWSLambdaException",
                          "Lambda.SdkClientException",
                          "Lambda.TooManyRequestsException"
                        ],
                        "IntervalSeconds": 2,
                        "MaxAttempts": 6,
                        "BackoffRate": 2
                      }
                    ],
                    "End": true
                  }
                }
              },
              "End": true,
              "ItemsPath": "$.Items",
              "ItemSelector": {
                "MyMultiplicationFactor.$": "$.BatchInput.MyMultiplicationFactor",
                "MyItem.$": "$$.Map.Item.Value"
              }
            }
          }
        },
        "End": true,
        "Label": "Map",
        "MaxConcurrency": 1000,
        "ItemsPath": "$.MyItems",
        "ItemBatcher": {
          "MaxItemsPerBatch": 3,
          "BatchInput": {
            "MyMultiplicationFactor.$": "$.MyMultiplicationFactor"
          }
        }
      }
    }
  }

创建 Lambda 函数

1. 使用 Lambda 控制台创建一个 Python Lambda 函数，命名为 ProcessSingleItem。有关创建 Lambda 函数的信息，请参阅开始使用分布式 Map 状态教程中的第 4 步：配置 Lambda 函数。

2. 复制以下 Lambda 函数代码，并将其粘贴到 Lambda 函数的代码源部分。

   import json
   
   def lambda_handler(event, context):
       
       multiplication_factor = event['MyMultiplicationFactor']
       item = event['MyItem']
       
       result = multiplication_factor * item
       
       return {
           'statusCode': 200,
           'multiplied': result
       }

3. 创建 Lambda 函数后，复制显示在页面右上角的函数 ARN。以下是一个 ARN 示例，其中 function-name 是 Lambda 函数的名称（在本例中为 ProcessSingleItem）：

   arn:aws:lambda:us-east-1:123456789012:function:function-name

   您需要在第 1 步中创建的状态机中提供函数 ARN。

4. 选择部署，以部署更改。