step functions Mapを使用して動的に並列処理してみた

DMMデータインフラ部に所属している[yuua](https://qiita.com/yuua0216)です
DMMデータインフラ部では、一部プロダクトにて[aws step functions](https://aws.amazon.com/jp/step-functions/) を活用してプロダクト開発を行っております。

# 背景
複数のサイトを運営しており、その際のデータを収集するため一部APIリクエストを使用しています
その際に正しい値がPOSTされているかを毎回手動でチェックしていたため
こちらをStep functionsを使用し並列に自動で行うための仕組みを作成しました。

# 構成
今回はStep functionsのMapがどういうものなのかをわかりやすくするため
数値のListを生成し偶数か奇数かを判定する簡易的なLambda関数を実装し、step functionsで並列処理します

<img width="600" alt="architecture.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/6055/6dc17792-baf7-5989-5849-299c05a6485e.png">

# step functions map概要

```
Map 状態 ("Type": "Map") を使用して、入力配列の要素ごとに一連のステップを実行できます。   
Parallel 状態は同じ入力を使用して複数のステップのブランチを実行しますが、
Map 状態は状態入力の配列の複数のエントリに対して同じステップを実行します。 
```

[こちら](https://docs.aws.amazon.com/ja_jp/step-functions/latest/dg/amazon-states-language-map-state.html)のaws公式ドキュメントの説明通り、入力配列の要素ごとにステップを実行することができ、
Parallelは同じデータに対する実行ですが、Mapは動的な個々(異なる)の値に対してステップを実行することができます。

# aws resources作成

terraformで構成を管理書くことが多いのでterraformで作成します

## step functionsの内容を定義するjsonファイル

+ Statesで実行するlambda関数を定義しそれを実行します
+ Nextでは次のステップで実行するtask名を定義します
+ Mapの定義自体は通常のTaskと同じように定義します
+ Mapのパラメータに関してはlambdaのeventで渡されます
+ concurrencyはMapの最大並列実行数です、Lambdaを指定する場合は関数の最大実行数を考慮して設定してください
+ Retry/Catchに関してはMap内の処理が一つでも失敗した場合ステートマシン自体が失敗になり処理が中止されてしまうため定義しています。(通常ではここでエラー内容の通知を行っていますが、今回はエラー処理自体をパスします)
+ 当然ですがRetryに関してはMap内のみRetryされます


```definition.json
{
    "Comment": "state machine",
    "StartAt": "MakePatternArray",
    "States": {
        "MakePatternArray": {
            "Type": "Task",
            "Resource": "${list_lambda_arn}",
            "Next": "ExecuteMap"
        },
        "ExecuteMap": {
            "Type": "Map",
            "MaxConcurrency": ${concurrency},
            "Iterator": {
                "StartAt": "ExecuteMaps",
                "States": {
                    "ExecuteMaps": {
                        "Type": "Task",
                        "Resource": "${map_lambda_arn}",
                        "Retry": [ {
                            "ErrorEquals": [ "States.ALL" ],
                            "IntervalSeconds": 2,
                            "BackoffRate": 2.0,
                            "MaxAttempts": 3
                        }],
                        "Catch": [{
                            "ErrorEquals": ["States.ALL"],
                            "Next": "MapException"
                        }],
                        "End": true
                    },
                    "MapException": {
                       "Type": "Pass",
                       "Result": "Error",
                       "End": true
                    }
                }
            },
            "End": true
        }
    }
}
  
```

+ lambdaやroleなどは別途作成してある想定

```stpfnc.tf
data "template_file" "definition_json" {
  template = file("definition.json")
  vars = {
    list_lambda_arn      = var.list_lambda_arn
    map_lambda_arn       = var.map_lambda_arn
    concurrency          = 10
  }
}

resource "aws_sfn_state_machine" "sfn_state_machine" {
  name       = "${var.step_function_name}-state-machine"
  role_arn   = aws_iam_role.state_machine_role.arn
  definition = data.template_file.definition_json.rendered
}

```

# Lambda関数の定義

## 数値の配列を返すlambda

```list.js
exports.handler = async () => {
    const response = [291,394,290,111,2,8,10]
    return response;
};
```

上では数値の配列を生成していますがObject自体を返すこともできます。
下記のようにObjectを生成してlambdaのeventをMap側で受け取ることも可能
ただし、渡せるデータ容量には制限があるので注意してください

```object.js
exports.handler = async () => {
    const response = [
     {
       'aaa': 123
       `bbb`: 333
     },
     {
       'aaa': 1231
       `bbb`: 3333
     }
    ]
    return response;
};
```

## 判定するLambda

整数値以外は許容していない、非常に適当な関数ですが...

```map.js
exports.handler = async (event) => {
    if (event % 2 == 0) {
      return '偶数'
    }
    return '奇数'
};
```

今回はstep functionsのDSL内にて `Catch` を使用してエラーが発生したものは `Pass`するようになっているため
一部が失敗してもstate machine自体は成功します


# state machineを起動する

起動方法はCloudWatch eventやCliなどいろいろありますが、ここではコンソールで起動/実行します


## 作成したstate machine
<img width="600" alt="statemachine.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/6055/625ab671-903d-ccc9-8cd5-e75e60cb12fa.png">

作成したstate machineから `実行の開始` を選択肢(パラメータが必要であれば入力/json)すると起動されます。
起動後の画面は上記のグラフインスペクターとLogなどが表示されます。

## 成功時

成功時はstate machine内のtaskが緑色になります
<img width="200" alt="success.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/6055/920eefdf-a687-e47c-3c06-bccf5908de32.png">

taskの選択もしくは実行イベント履歴からOutputされたデータの確認も可能です
<img width="600" alt="log.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/6055/69afea07-abe0-bb74-c401-554f98544ff1.png">


## エラー発生時

失敗時は下記の画像のように失敗したtaskが赤くなり、`Catch` を定義していない場合は
state machine全体が失敗になります。

<img width="200" alt="error.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/6055/1d8ab453-c4aa-764a-2cf6-42145cdb960e.png">

失敗したtaskを選択もしくは
実行イベント履歴にあるLogからエラー内容を確認することが可能です

# まとめ
step functionsのMapを使用することで動的に生成した異なるデータに対して簡単に並列にアプローチをすることができます。一連のworkflowの中で、動的データを並列に処理することができれば、処理の高速化などといったメリットになると思います。
step functionsには他にもいろいろなStatesや様々なawsサービスとの連携も簡単に行うことができ非常に便利なので
workflowエンジンをお考えの際は有力な選択肢としてぜひ一度使ってみてはいかがでしょうか。

データインフラ部では、AWSでのビッグデータ基盤関連の運用を行っており、
Step Functions なども活用したプロダクト開発を行っています。
中途採用などもおこなっておりますので、興味のある方は、一度弊社HPなどから
カジュアル面談など、是非ご応募ください。