本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

# 在 Step Functions 中使用 Lambda 函數處理個別項目
<a name="tutorial-itembatcher-single-item-process"></a>

在本教學課程中，您可以使用*分散式地圖狀態*[ItemBatcher （地圖）](input-output-itembatcher.md)的欄位，使用 Lambda 函數反覆查看批次中存在的個別項目。*分散式映射狀態*會啟動四個子工作流程執行。每個子工作流程都會執行*內嵌映射狀態*。對於其每次反覆運算，*內嵌映射狀態*會叫用 Lambda 函數，並將單一項目從批次傳遞至函數。然後，Lambda 函數會處理項目並傳回結果。

您將建立狀態機器，在整數陣列上執行乘法。假設您作為輸入提供的整數陣列是 ，`[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]`而乘數因數是 `7`。然後，在將這些整數乘以係數 7 之後形成的陣列將是 `[7, 14, 21, 28, 35, 42, 49, 56, 63, 70]`。

## 步驟 1：建立狀態機器
<a name="itembatcher-single-item-process-create-state-machine"></a>

在此步驟中，您將建立狀態機器的工作流程原型，將單一項目從一批項目傳遞到您將在[步驟 2](#itembatcher-single-item-process-config-resource) 中建立的每個 Lambda 函數叫用。
+ 使用以下定義，使用 [Step Functions 主控台](https://console.aws.amazon.com/states/home?region=us-east-1#/)建立狀態機器。如需有關建立狀態機器的資訊，請參閱《使用分散式地圖狀態教學》[步驟 1：建立工作流程原型](tutorial-map-distributed.md#use-dist-map-create-workflow)中的入門。 [在 Step Functions 中使用分散式映射複製大規模 CSV 資料](tutorial-map-distributed.md)

  在此狀態機器中，您可以定義*一個分散式映射狀態*，接受 10 個整數的陣列做為輸入，並將這些陣列項目分批傳遞給子工作流程執行。每個子工作流程執行都會收到一批三個項目做為輸入，並執行*內嵌映射狀態*。*內嵌映射狀態*的每個反覆運算都會叫用 Lambda 函數，並將項目從批次傳遞至函數。然後，此函數會將項目乘以係數 `7`並傳回結果。

  每個子工作流程執行的輸出都是 JSON 陣列，其中包含傳遞的每個項目的乘法結果。
**重要**  
請務必將下列程式碼中 Lambda 函數的 Amazon Resource Name (ARN) 取代為您將在[步驟 2 ](#itembatcher-single-item-process-config-resource)中建立之函數的 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:{{region}}:{{account-id}}: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"
          }
        }
      }
    }
  }
  ```

## 步驟 2：建立 Lambda 函數
<a name="itembatcher-single-item-process-config-resource"></a>

在此步驟中，您會建立 Lambda 函數，以處理從批次傳遞的每個項目。

**重要**  
確保您的 Lambda 函數與您的 AWS 區域 狀態機器位於相同的 下。

**建立 Lambda 函數**

1. 使用 [Lambda 主控台](https://console.aws.amazon.com/lambda/home)建立名為 的 **Python** Lambda 函數**ProcessSingleItem**。如需建立 Lambda 函數的資訊，請參閱《 [開始使用分散式地圖狀態](tutorial-map-distributed.md)教學》中的[步驟 4：設定 Lambda 函數](tutorial-map-distributed.md#use-dist-map-config-resource)。

1. 複製 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
       }
   ```

1. 建立 Lambda 函數後，複製頁面右上角顯示的函數 ARN。以下是範例 ARN，其中 {{`function-name`}}是 Lambda 函數的名稱 （在此情況下為 `ProcessSingleItem`)：

   ```
   arn:aws:lambda:{{region}}:123456789012:function:{{function-name}}
   ```

   您需要在步驟 [1 ](#itembatcher-single-item-process-create-state-machine)中建立的狀態機器中提供函數 ARN。

1. 選擇**部署**以部署變更。

## 步驟 3：執行狀態機器
<a name="itembatcher-param-task-run-state-machine"></a>

當您執行[狀態機器](#itembatcher-single-item-process-create-state-machine)時，*分散式映射狀態*會啟動四個子工作流程執行，每個執行會處理三個項目，而一個執行會處理單一項目。

下列範例顯示傳遞至子工作流程執行內其中一個[`ProcessSingleItem`](#itembatcher-single-item-process-config-resource)函數叫用的資料。

```
{
  "MyMultiplicationFactor": 7,
  "MyItem": 1
}
```

鑑於此輸入，以下範例顯示 Lambda 函數傳回的輸出。

```
{
  "statusCode": 200,
  "multiplied": 7
}
```

下列範例顯示其中一個子工作流程執行的輸出 JSON 陣列。

```
[
  {
    "statusCode": 200,
    "multiplied": 7
  },
  {
    "statusCode": 200,
    "multiplied": 14
  },
  {
    "statusCode": 200,
    "multiplied": 21
  }
]
```

狀態機器會傳回下列輸出，其中包含四個子工作流程執行的四個陣列。這些陣列包含個別輸入項目的乘法結果。

最後，狀態機器輸出是名為 的陣列`multiplied`，結合了針對四個子工作流程執行傳回的所有乘法結果。

```
[
  [
    {
      "statusCode": 200,
      "multiplied": 7
    },
    {
      "statusCode": 200,
      "multiplied": 14
    },
    {
      "statusCode": 200,
      "multiplied": 21
    }
  ],
  [
    {
      "statusCode": 200,
      "multiplied": 28
    },
    {
      "statusCode": 200,
      "multiplied": 35
    },
    {
      "statusCode": 200,
      "multiplied": 42
    }
  ],
  [
    {
      "statusCode": 200,
      "multiplied": 49
    },
    {
      "statusCode": 200,
      "multiplied": 56
    },
    {
      "statusCode": 200,
      "multiplied": 63
    }
  ],
  [
    {
      "statusCode": 200,
      "multiplied": 70
    }
  ]
]
```

若要將子工作流程執行傳回的所有乘法結果合併為單一輸出陣列，您可以使用 [ResultSelector](input-output-inputpath-params.md#input-output-resultselector) 欄位。在*分散式映射狀態*中定義此欄位，以尋找所有結果、擷取個別結果，然後將它們合併為名為 的單一輸出陣列`multiplied`。

若要使用 `ResultSelector` 欄位，請更新您的狀態機器定義，如下列範例所示。

```
{
  "StartAt": "Pass",
  "States": {
    ...
    ...
    "Map": {
      "Type": "Map",
      ...
      ...
      "ItemBatcher": {
        "MaxItemsPerBatch": 3,
        "BatchInput": {
          "MyMultiplicationFactor.$": "$.MyMultiplicationFactor"
        }
      },
      "ItemsPath": "$.MyItems",
      "ResultSelector": {
        "multiplied.$": "$..multiplied"
      }
    }
  }
}
```

更新的 狀態機器會傳回合併的輸出陣列，如下列範例所示。

```
{
  "multiplied": [7, 14, 21, 28, 35, 42, 49, 56, 63, 70]
}
```