AWS Step Functions の Distributed Map ステートを試してみる

これまで、AWS Step Functions の Map ステートを試してみる記事を 2 つ（下記）書いてきました。

nobelabo.hatenablog.com

今回は、その続きとして 2022 年 12 月にアナウンスされた Distributed Map を試してみます。

aws.amazon.com

この記事の内容は、2022 年 12月時点で検証した内容に基づきます。リージョンは東京リージョンを使用しています。

これまでの Map ステートでは、並列度は 40 までしかサポートされていませんでした。

そのため、40 以上の並列度で処理をしたい場合は、Map ステートから別のステートマシンを呼出し、そのステートマシンでさらに Map ステートを使用するという Mapの入れ子構造を作成する必要がありました。

例えば、1024 の並列処理を行いたい場合は、親のステートマシンで 32 の並列度を指定した Map ステートを用意し、そこから呼び出した子となるステートマシンで 32 の並列度を指定した Map を構成することで、結果的に 32 × 32 = 1024 の並列度を実現していました。

しかし、今回リリースされた Distributed Map では、並列度が 10,000 までサポートされましたので、「Map の入れ子構造」を作成する必要はなくなりました。

では、実際に Distributed Map を使ったステートマシンを作成してみます。

今回も、Distributed Map に渡す配列は、下記の Lambda関数 ArrayGenerator で作成することにします。

この ArrayGenerator は Python 3.9 で作成されており、イベントオブジェクトから length というキーで指定された数値に基づき、動的に配列を生成してリターンします。

import json

def lambda_handler(event, context):
    print(event)
    length = event['length']
    items = []
    for id in range(length):
      items.append({'item': id})
    return items

次に、ステートマシンを作成します。下記の JSON を使用して、前回の記事と同様の手順で作成します。

なお、前回は Map から Lambda関数を invoke していましたが、今回はコスト面を考慮し、特に何もしないステートである Passステートを Distributed Map から実行する形にします。

ステートマシンの JSON は下記です。

(なお、Workflow Studio で GUI から作成することもできますが、その場合は Map ステートを選択して、「処理モード」として「分散」を指定します。)

{
  "Comment": "A description of my state machine",
  "StartAt": "Lambda Invoke",
  "States": {
    "Lambda Invoke": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "Payload.$": "$",
        "FunctionName": "arn:aws:lambda:ap-northeast-1:123412341234:function:ArrayGenerator"
      },
      "Retry": [
        {
          "ErrorEquals": [
            "Lambda.ServiceException",
            "Lambda.AWSLambdaException",
            "Lambda.SdkClientException"
          ],
          "IntervalSeconds": 2,
          "MaxAttempts": 6,
          "BackoffRate": 2
        }
      ],
      "ResultPath": "$.generatedArray",
      "Next": "Map"
    },
    "Map": {
      "Type": "Map",
      "ItemProcessor": {
        "ProcessorConfig": {
          "Mode": "DISTRIBUTED",
          "ExecutionType": "STANDARD"
        },
        "StartAt": "Pass",
        "States": {
          "Pass": {
            "Type": "Pass",
            "End": true
          }
        }
      },
      "End": true,
      "Label": "Map",
      "MaxConcurrency": 1000,
      "InputPath": "$.generatedArray.Payload"
    }
  }
}