Step Functionsで動的並列処理

概要

Step Functions でワークフローを組む際、前段の出力が可変長になることがある。こういうときに次段の並列数を動的に決定したい。これを叶えるための機能が存在するので実際に試してみた。

機能の解説は参考リンクに譲るとして、ここではコンソール画面から実際に設定、実行するまでの手順を書いていく。

設定方法

全体のワークフロー

Map を置いてその下に繰り返したい処理を入れる。

今回 Lambda 関数は Python3.9 で以下の様に実装している。

import json

def lambda_handler(event, context):
    return {
        'statusCode': 200,
        'body': json.dumps({"count": event.get("count")})
    }

Map の設定

「項目配列のパス」には反復処理したい配列へのパスを指定する。今回は入力として以下のような形を想定しているので $.counts を入れておく。

{
  "counts": [1, 2, 3, 4, 5]
}

このままだと Lambda への入力が数字だけになってしまうので、前述の関数のように event.get("count") で受け取れるように配列項目の変換を行う。

関数内では count というキーで取得したいので以下のように変換を行う。他にも $$.Map.Item.Index でインデックスを取得したりもできる。

{
  "count.$": "$$.Map.Item.Value"
}

今回、出力、エラー処理に関しては特に修正しない。

Lambda: Invoke の設定

ちゃんと想定の関数を指定する。だけ。

実行

以下で入力を与えて実行をしてみる。

{
  "counts": [
    1,
    21,
    39,
    40,
    52,
    68
  ]
}

Lambda 関数の入力は想定通り以下の様になっている。

{
  "count": 1
}

出力は以下。

{
  "statusCode": 200,
  "body": "{\"count\": 1}"
}

コンソール上から入力配列の各項目に関する結果を見ることができる。

入力の配列数を変えると起動される Lambda の個数も変わることが確認できる。便利。

参考 Links