Amazon Step FunctionsのWorkflow StudioでParallel, Map, Choise状態の入出力を見る

# はじめに
　本記事では、Amazon Step FunctionsのWorkflow StudioでParallel, Map, Choise状態を含むステートマシンを作成し、その入出力を見ていきます。Step Functionsでは、Parallelで静的並列処理、Mapで動的並列処理を実現することができます。

　本記事の特徴は以下です。

* Workflow Studioを使ってグラフィカルに作る
* Parallel, Map, Choise状態とLambda関数を使用
* 実践的な処理というよりは、入出力を明確にするテスト的な処理を作る
* LambdaでPythonを使用しますが、簡単すぎるコードしかないため未経験者でも問題ありません

　以下の内容は含まれません。

* ASL(次節で説明)を使って作る方法

# Step Functionsとは？
　サーバレスオーケストレーションサービスと [公式][1] では言われています。分かり易く言い換えると、サーバレスなAWSサービス(主に [AWS Lambda][2] )をつなげて1つのワークフローを作成できるサービスです。

:::note info
<details>
<summary><a href="https://fixel.co.jp/blog/why_serverless_now/">サーバレス</a>とは？</summary>
　必要な機能に応じてAWSがサーバを用意し、使い終わったらリリースしてくれるので、ユーザーがサーバの存在を意識することなく、クラウドシステムを構築・運用できる仕組み。サーバがないわけではなく、AWSのサーバ自体は使っています。
</details>
:::

　Step Functionsを使うことで、処理フローの中のコンポーネント同士を疎結合しつつ、入出力の見通しも良くできます。

　Step Functionsでは、状態(1つの作業単位)を組み合わせてステートマシン(＝ワークフロー)を作成します。今回作るステートマシンは以下のようになっています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/bd37207d-9a79-2664-5176-a72d1e5235e2.png)

　ステートマシンを作成する方法は以下の2種類あります。

* [Amazon State Language (ASL)][3] を使う
* [Workflow Studio][4] を使う

　前者はStep Functions用のJSONベースの言語を習得する必要があり、学習コストが高いです。
　後者は以下のようなGUIで作成していくため、初心者でも取っつきやすいと思います。今回はそのWorkflow Studioを使います。その際、各状態の入出力関係でつまづくことが多いと思ったので、この記事ではそれを明らかにすることを目指しました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/ffb72b9e-1463-4e4f-b36e-d2e3869f9418.png)

# 準備　Lambda関数を作る
　今回使用するLambda関数をまず作っちゃいます。

### Add関数
　AWSコンソールのAWS Lambdaページで、[関数の作成]をクリックし、以下のような設定でLambda関数を作成します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/96f330a7-d883-e54a-7860-92f6401d2771.png)

　Lambda関数作成後に出てくるlambda_function.pyを以下に書き換え、[Deploy]をクリックします。

```python:lambda_function.py
def lambda_handler(event, context):
    return {
        'num': add(event['num0'], event['num1'])
    }

def add(num0, num1):
    return num0 + num1
```

### Sub関数
　同様にSubという名前のLambda関数を作成し、lambda_function.pyを以下に書き換え、[Deploy]をクリックします。

```python:lambda_function.py
def lambda_handler(event, context):
    return {
        'num': sub(event['num0'], event['num1'])
    }

def sub(num0, num1):
    return num0 - num1
```

### Sum関数
　同様にSumという名前のLambda関数を作成し、lambda_function.pyを以下に書き換え、[Deploy]をクリックします。

```python:lambda_function.py
def lambda_handler(event, context):
    return {
        'num': mysum(event)
    }

def mysum(event):
    l = [e['num'] for e in event]
    return sum(l)
```

# Workflow Studioを使ってステートマシンを作る
　ステートマシンを作っていきます。

　まず、AWSコンソールのStep Functionsページで、[ステートマシンの作成]をクリックし、以下のような設定で[次へ]行きます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/e1c152fd-e78a-331f-7e64-59e3a0ead3d2.png)

　するとWorkflow Studioの編集画面が開きます。ここに状態を表すカードを配置することで、処理のフローを定義します。

　左の[フロー]タブからParallelとMapのカードを配置してみましょう。両方のMapの[設定]タブ内「項目配列へのパス」に`$.input`と書いておきます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/74777835-7f9d-f7d8-6e4e-d9cd01ca5beb.png)

* Parallelは [静的並列処理][5] を表しています。以下の性質があります。
    * Parallelを使うことで処理の分岐が定義できる
    * 分岐先の処理は並列に実行される
    * 分岐先では異なる処理を実行することができる
    * Parallelが受け取った入力がそのまま分岐先の入力になる
    * 分岐先はすべて同じ入力を受け取る
    * ステートマシン作成時に並列数が決まる(静的)
* Mapは [動的並列処理][5] を表しています。以下の性質があります。
    * Mapは入力としてリストを受け取る
    * Mapを使うことでリストの各要素に対して同じ処理を実行できる
    * 各要素に対する処理は並列に実行される
    * 並列数はMapに入力される要素数に応じて変わる(動的)

　Map内に、先ほど作ったAddというLambda関数とSubというLambda関数を配置します。

　左の[アクション]タブからAWS Lambdaカードを2枚配置しましょう。カードをクリックして右の[設定]タブで、状態名を分かり易く変えておきましょう。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/873516ae-77a8-6fdf-14f1-4a59b20d6bc4.png)

　また、Addカードの[設定]タブの「Function name」において、プルダウンから先ほど作成したAdd関数を選択します。同様にSubカードもSub関数を選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/e4fdbe56-2119-bb1c-f595-0a820a615ecb.png)

　次に、リストの要素の和を取るSum関数を配置します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/19ac95b5-358c-71d0-8ee6-6c1d33aa7a5c.png)

　最後に、Choiceを使って、Parallelの2つの結果を比較してSum 1の方が大きかったら成功、そうでない場合失敗とする処理を定義します。

　左の[フロー]タブからChoiceをParallelの下に置くと以下になります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/f6854842-7c6d-d190-9d0f-387da26b871c.png)

　Choiceの[設定]タブの「Choice Rules」内にある「Default rule」の右にあるマークをクリックし、「オプションを選択」のプルダウンから「新しい状態を追加」を選択すると以下のようになります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/edb5ac9d-9b59-8e61-6c4e-205ec076efc9.png)

　Rule #1と書いてあるところに[フロー]からSuccessカードを置き、DefaultにFailカードを置きます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/278dfe4d-4883-e870-5b7c-d500e2856e24.png)

　そして、Successに行く条件(Rule #1)を編集します。Choiceカードをクリックし、[設定]タブの「Choice Rules」内にある「Rule #1」の右にあるマークをクリックし、[Add conditions]をクリックします。
　出てきたウィンドウに以下のような内容を入力し、[Save conditions]。`$[0].num`はsum 1の結果を表し、これが`$[1].num`(sum 2の結果)より大きいときにSuccess状態に移ります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/efad8b03-bed3-f2b6-d5b5-9278f7127682.png)

　ワークフローの作成は以上で終わりなので、Workflow Studio右上の[次へ]をクリックします。

　自動生成されたASLと状態図が表示されます。特にいじらず[次へ]。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/f80f49df-4128-6aa4-dede-9565004c7e8b.png)

　名前はお好みで変えてください。「新しいロールの作成」にチェックを入れておくと、Step FunctionsがAdd、Sub、Sum関数にアクセスできるポリシーを含んだロールが自動で作成されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/a32c2319-cb55-a466-8d12-b7091d28fb8b.png)

　[ステートマシンの作成]をクリックすると、以下のような画面になります。[実行の開始]をクリック。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/168d20a9-ceb7-4874-43e7-7928388a9ead.png)

　出てきた画面に以下のJSONテキストを書き込んで[実行の開始]。

```json
{
  "input": [
    {
      "num0": 1,
      "num1": 1
    },
    {
      "num0": 2,
      "num1": 1
    },
    {
      "num0": 5,
      "num1": 2
    }
  ]
}
```

　実行が成功すると、状態図がすべて緑になります。状態図のクリックされた箇所の詳細が右に表示され、入力に対する出力が直ぐに見れて便利です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/325733/94b3a0ae-5d50-a6a7-6709-0e67a2ed47f2.png)

# 各ステップの入力と出力
```json:全体の入力、Parallelの入力
{
  "input": [
    {
      "num0": 1,
      "num1": 1
    },
    {
      "num0": 2,
      "num1": 1
    },
    {
      "num0": 5,
      "num1": 2
    }
  ]
}
```

　全体の入力は`"input"`キーのみで、値はリストです。Parallelの入力もこれと全く同じです。  

```json:MapAddの入力
[
    {
      "num0": 1,
      "num1": 1
    },
    {
      "num0": 2,
      "num1": 1
    },
    {
      "num0": 5,
      "num1": 2
    }
]
```

　ParallelからMap AddとMap Subへの入力も全体の入力と同じになりますが、Mapの[設定]タブ内「項目配列へのパス」に$.inputと書いておいたので、 [JSONPath][6] の構文に従って`"input"`の値であるリストがマップに渡されます。

```json:Add[0]の入力
{
  "num0": 1,
  "num1": 1
}
```

　Map Addに渡されたリストのインデックス $i$ の要素がAdd[i]の入力になります。

```json:Add[0]の出力
{
  "num": 2
}
```

　Addは入力された2つの値の足し算結果を返します。

```json:MapAddの出力、Sum1の入力
[
  {
    "num": 2
  },
  {
    "num": 3
  },
  {
    "num": 7
  }
]
```

　Mapの出力は、各要素に対する処理の出力のリストになります。Map Addの出力はAdd[0]~Add[2]の出力のリストです。これがそのままSum 1に入力されます。

```json:Sum1の出力
{
  "num": 12
}
```

　Sumは入力されたリストの和を計算し、12を出力します。

```json:Sum2の出力
{
  "num": 4
}
```

　全く同様にSubの方も引き算が並列処理され、Sum 2の出力は4になります。

```json:Parallelの出力
[
  {
    "num": 12
  },
  {
    "num": 4
  }
]
```

　Parallelの出力はすべて(今回は2つ)の並列処理の出力のリストになります。

　最後に、Choiceによるif文のような分岐処理がなされます。Choiceの設定で、Successに行く条件は`$[0].num > $[1].num` としていましたので、`$[0].num = 12, $[1].num = 4` である今回の入力では条件を満たしてSuccessとなります。

# 最後に
　今回はLambdaとStep Functions標準の状態しか使いませんでしたが、本当にいろんなAWSサービスを処理フローに組み込めるので、興味がある方は挑戦してみてください。

　最後までご覧いただきありがとうございました。

[1]:https://docs.aws.amazon.com/ja_jp/step-functions/latest/dg/welcome.html
[2]:https://docs.aws.amazon.com/ja_jp/lambda/latest/dg/welcome.html
[3]:https://docs.aws.amazon.com/ja_jp/step-functions/latest/dg/concepts-amazon-states-language.html
[4]:https://docs.aws.amazon.com/ja_jp/step-functions/latest/dg/workflow-studio.html
[5]:https://devlog.arksystems.co.jp/2021/06/22/14428/
[6]:https://qiita.com/takkii1010/items/0ce1c834d3a73496ccef