AWS Step Functionsで大規模な並列処理を実施できるDistributed Mapを試してみた #reinvent

こんにちは、CX事業本部 IoT事業部の若槻です。

先日開催されたAWS re:Invent 2022で、AWS Step Functionsで大規模な並列処理を実施できるStep Functions Distributed Mapが発表されました。

• AWS Step Functions launches large-scale parallel workflows for data processing and serverless applications
• Step Functions Distributed Map – A Serverless Solution for Large-Scale Parallel Data Processing | AWS News Blog

S3 Bucket内のオブジェクトをもとに大規模なMap処理が行えるとのことで、実際に現地会場でKeynoteを聞いていた時にとても興奮したのを覚えています。

今回は、Step Functions Distributed Mapを使用したWorkflowを実際に作成して試してみました。

やってみた

Workflow作成

Workflow Studioで作成していきます。

What's newとしてDistributed mode for Mapが示されています。[View]をクリック。

[Flow]でMapがUpdatedとなっていますね。[Updated]をクリック。

するとポップアップでDistributed Mapの軽い解説が出ます。さらにリンクをクリック。

すると従来のMap（Inline mode）と新しいMap（Distributed mode）の比較表が出てきました。分かりやすい。

MapをキャンバスにD&Dすると、Formで今までは無かった「Processing mode」というメニューが出るようになっていました。ここでMapのモードを指定するようです。

[Processing mode]でDistributedを選択するとメニューがガラリと変わりました。

ここから、メニューの各パラメータでどのような設定ができるのか確認してみます。

[Item source]では、Map stateが入力データをどこから受け取るかを定義します。State inputを選択すると直前のステップから渡されたJSON配列データが使用されます。またオプションとして[Provide a path to items array]を設定すると渡されるデータのパスを指定でき、[Modify items with ItemSelector]を設定すると渡されるデータを加工できます。

Amazon S3を選択するといずれか（後述）のAmazon S3データソースから取得されたアイテムのコレクションが使用されます。オプションとして[Limit number of items]を設定すると取得されるアイテム数の上限を指定でき、[Modify items with ItemSelector]を設定すると渡されるデータを加工できます。

Amazon S3データソースはS3 object list、JSON file in S3、CSV file in S3およびS3 Inventoryの4つから選択できます。

S3 object listを選択した場合は対象のBucketおよびPrefixを指定します。

JSON file in S3、CSV file in S3またはS3 Inventoryを選択した場合は、対象のオブジェクトが格納されたBucketおよびKeyを指定します。

ここでS3 Inventoryとは、S3 Bucketに格納されたオブジェクトのメタ情報（スキーマなど）を示すmanifest fileです。詳細は以下を参考ください。

• itemsource-example-s3-inventory - ItemReader - AWS Step Functions

[Item batching]を有効にすると、子実行毎に複数のアイテムをバッチ処理させることができ、有効化が推奨されています。

有効にした場合はバッチ毎の最大アイテム数、最大サイズおよびグローバルなJSON入力を指定できます。

Map Run内で実行されるLambda関数の同時実行数（既定は1,000）や、Distributed Map自体の子実行の同時実行数（10,000）を超える恐れがある場合はItem batchingを設定すれば良さそうですね。

[Set concurrency limit]を有効にしたい場合は子実行の同時最大実効数を制限できます。

[Child execution type]では、子実行のタイプを指定できます。

[Runtime settings]の追加設定として、[Set a tolerated failure threshold]ではMap Runでの子実行の許容失敗量を指定できます。既定では1つ以上失敗するとMap Runは失敗します。[Use state name as label in Map Run ARN]ではMap Run ARNにカスタムラベルを指定できます。既定ではMap stateの名前が使用されます。

[Export Map state results to Amazon S3]を有効にすると、Map Runの実行結果をS3 Bucketの指定の場所に格納できます。

今回はCSVファイルに記述された内容をもとにDistributed Mapを実施する次のようなWorkflowを作成しました。Map state内ではPassタスクを１つ実行するだけです。

{
  "Comment": "A description of my state machine",
  "StartAt": "Map",
  "States": {
    "Map": {
      "Type": "Map",
      "ItemProcessor": {
        "ProcessorConfig": {
          "Mode": "DISTRIBUTED",
          "ExecutionType": "STANDARD"
        },
        "StartAt": "Pass",
        "States": {
          "Pass": {
            "Type": "Pass",
            "End": true
          }
        }
      },
      "End": true,
      "ItemReader": {
        "Resource": "arn:aws:states:::s3:getObject",
        "Parameters": {
          "Bucket": "20221212-cmwakatsuki",
          "Key": "data.csv"
        },
        "ReaderConfig": {
          "InputType": "CSV",
          "CSVHeaderLocation": "FIRST_ROW"
        }
      },
      "MaxConcurrency": 1000,
      "Label": "Map"
    }
  }
}

グラフビューはこんな感じ。

S3 Bucketには3つのアイテムを記述した次のCSVファイルを格納しています。

id,name,value
d001,デバイス1,10
d002,デバイス2,5
d003,デバイス3,8

動作確認

Workflowを実行してみます。

実行が成功しました。CSVファイルから取得された3つのアイテムがWorkflowの出力結果として取得できています。

Event viewではMap Runが別途ページで見れるようです。開いてみます。

遷移先でMap Runの実行結果の詳細が確認できました。子実行の一覧もありますね。そのうち1つを開いてみます。

すると遷移先で子実行の個別の実行内容を確認できました。

おわりに

Step Functions Distributed Mapを使用したWorkflowを実際に作成して試してみました。

Step Functionsは今までの通常のMap Stateにおいてもある程度大きな並列実行には対応できましたが、今回のアップデートによりさらに大規模な実行に対応できるようになった形です。またS3 Bucketのオブジェクトを直接指定して実行できるのも素晴らしいですね。

またWorkflow作成時に確認した通りDistributed Mapの設定パラメータは今回試したほかにも様々あるので、今後ほかのパターンでも検証をやってみたいと思います。

参考

• [速報]Step Functionsで大量データの並列処理が可能に！Step Functions Distributed MapがGAされました！# reInvent | DevelopersIO

以上