Lambda 関数でデータバッチ全体を処理する

このチュートリアルでは、分散マップ状態の ItemBatcher フィールドを使用して、Lambda 関数内の項目のバッチ全体を処理します。各バッチには、最大 3 つの項目が含まれます。分散マップ状態では、4 つの子ワークフロー実行が開始されます。それぞれの実行では 3 つの項目が処理され、1 つの実行では 1 つの項目が処理されます。子ワークフローを実行するたびに、バッチに存在する個々の項目を反復処理する Lambda 関数を呼び出します。

整数の配列を乗算するステートマシンを作成します。入力として指定した整数配列が [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] で、乗算係数が 7 であるとします。この場合、これらの整数に係数 7 を掛けた結果の配列は [7, 14, 21, 28, 35, 42, 49, 56, 63, 70] になります。

ステップ 1: ステートマシンを作成する

このステップでは、ステップ 2 で作成する Lambda 関数に、データのバッチ全体を渡すステートマシンのワークフロープロトタイプを作成します。

• 次の定義を使用して、Step Functions コンソールを使用してステートマシンを作成します。ステートマシンの作成については、チュートリアル「分散マップ状態の使用を開始する」の「ステップ 1: ワークフロープロトタイプを作成する」を参照してください。

  このステートマシンでは、入力として 10 個の整数の配列を入力として受け取り、この配列を 3 のバッチで Lambda 関数に渡す分散マップ状態を定義します。Lambda 関数は、バッチに存在する個々の項目を反復処理し、multiplied という名前の出力配列を返します。出力配列には、入力配列に渡された項目の乗算結果が含まれています。

  重要

  次のコードの 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": "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,
        "Label": "Map",
        "MaxConcurrency": 1000,
        "ItemBatcher": {
          "MaxItemsPerBatch": 3,
          "BatchInput": {
            "MyMultiplicationFactor.$": "$.MyMultiplicationFactor"
          }
        },
        "ItemsPath": "$.MyItems"
      }
    }
  }

ステップ 2 : Lambda 関数を作成する

このステップでは、バッチで渡されるすべての項目を処理する Lambda 関数を作成します。

重要

Lambda 関数がステートマシン AWS リージョン と同じ にあることを確認します。

Lambda 関数を作成するには

1. Lambda コンソールを使用して、ProcessEntireBatch という名前の Python 3.9 Lambda 関数を作成します。Lambda 関数の作成の詳細については、「分散マップ状態の使用開始」チュートリアルの「ステップ 4: Lambda 関数を設定する」を参照してください。 分散マップを使用した大規模 CSV データのコピー

2. Lambda 関数用の次のコードをコピーし、Lambda 関数の [コードソース] セクションに貼り付けます。

   import json
   
   def lambda_handler(event, context):
       multiplication_factor = event['BatchInput']['MyMultiplicationFactor']
       items = event['Items']
       
       results = [multiplication_factor * item for item in items]
       
       return {
           'statusCode': 200,
           'multiplied': results
       }

3. Lambda 関数を作成したら、ページの右上隅に表示されている関数の ARN をコピーします。ARN をコピーするには、 をクリックします。ARN の例を次に示します。ここで、function-name は Lambda 関数の名前 (この場合は ProcessEntireBatch) です。

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

   ステップ 1 で作成したステートマシンに関数 ARN を指定する必要があります。

4. [デプロイ] を選択して、変更をデプロイします。

ステップ 3: ステートマシンを実行する

ステートマシンを実行すると、分散マップ状態では、4 つの子ワークフロー実行が開始されます。それぞれの実行では 3 つの項目が処理され、1 つの実行では 1 つの項目が処理されます。

次の例は、子ワークフロー実行の 1 つによって ProcessEntireBatch 関数に渡されるデータを示しています。

{
  "BatchInput": {
    "MyMultiplicationFactor": 7
  },
  "Items": [1, 2, 3]
}

この入力の場合、次の例では Lambda 関数によって返される multiplied という名前の付いた出力配列を示しています。

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

ステートマシンは、4 つの子ワークフロー実行に multiplied という名前の付いた 4 つの配列を含む、次の出力を返します。これらの配列には、個々の入力項目の乗算結果が含まれます。

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

返されたすべての配列項目を 1 つの出力配列にまとめるには、ResultSelector フィールドを使用できます。分散マップ状態内でこのフィールドを定義すると、すべての multiplied 配列を検索し、その配列内のすべての項目が抽出され、それらの項目を 1 つの出力配列に結合できます。

ResultSelector フィールドを使用するには、次の例に示すようにステートマシン定義を更新します。

{
  "StartAt": "Pass",
  "States": {
    ...
    ...
    "Map": {
      "Type": "Map",
      ...
      ...
      "ItemsPath": "$.MyItems",
      "ResultSelector": {
        "multiplied.$": "$..multiplied[*]"
      }
    }
  }
}

更新されたステートマシンは、次の例に示すように統合された出力配列を返します。

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