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とは？

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

サーバレスとは？

　必要な機能に応じてAWSがサーバを用意し、使い終わったらリリースしてくれるので、ユーザーがサーバの存在を意識することなく、クラウドシステムを構築・運用できる仕組み。サーバがないわけではなく、AWSのサーバ自体は使っています。

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

　Step Functionsでは、状態(1つの作業単位)を組み合わせてステートマシン(＝ワークフロー)を作成します。今回作るステートマシンは以下のようになっています。

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

• Amazon State Language (ASL) を使う
• Workflow Studio を使う

　前者はStep Functions用のJSONベースの言語を習得する必要があり、学習コストが高いです。
　後者は以下のようなGUIで作成していくため、初心者でも取っつきやすいと思います。今回はそのWorkflow Studioを使います。その際、各状態の入出力関係でつまづくことが多いと思ったので、この記事ではそれを明らかにすることを目指しました。

準備　Lambda関数を作る

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

Add関数

　AWSコンソールのAWS Lambdaページで、[関数の作成]をクリックし、以下のような設定でLambda関数を作成します。

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

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]をクリックします。

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]をクリックします。

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ページで、[ステートマシンの作成]をクリックし、以下のような設定で[次へ]行きます。

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

　左の[フロー]タブからParallelとMapのカードを配置してみましょう。両方のMapの[設定]タブ内「項目配列へのパス」に$.inputと書いておきます。

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

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

　左の[アクション]タブからAWS Lambdaカードを2枚配置しましょう。カードをクリックして右の[設定]タブで、状態名を分かり易く変えておきましょう。

　また、Addカードの[設定]タブの「Function name」において、プルダウンから先ほど作成したAdd関数を選択します。同様にSubカードもSub関数を選択します。

　次に、リストの要素の和を取るSum関数を配置します。

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

　左の[フロー]タブからChoiceをParallelの下に置くと以下になります。

　Choiceの[設定]タブの「Choice Rules」内にある「Default rule」の右にあるマークをクリックし、「オプションを選択」のプルダウンから「新しい状態を追加」を選択すると以下のようになります。

　Rule #1と書いてあるところに[フロー]からSuccessカードを置き、DefaultにFailカードを置きます。

　そして、Successに行く条件(Rule #1)を編集します。Choiceカードをクリックし、[設定]タブの「Choice Rules」内にある「Rule #1」の右にあるマークをクリックし、[Add conditions]をクリックします。
　出てきたウィンドウに以下のような内容を入力し、[Save conditions]。$[0].numはsum 1の結果を表し、これが$[1].num(sum 2の結果)より大きいときにSuccess状態に移ります。

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

　自動生成されたASLと状態図が表示されます。特にいじらず[次へ]。

　名前はお好みで変えてください。「新しいロールの作成」にチェックを入れておくと、Step FunctionsがAdd、Sub、Sum関数にアクセスできるポリシーを含んだロールが自動で作成されます。

　[ステートマシンの作成]をクリックすると、以下のような画面になります。[実行の開始]をクリック。

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

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

　実行が成功すると、状態図がすべて緑になります。状態図のクリックされた箇所の詳細が右に表示され、入力に対する出力が直ぐに見れて便利です。

各ステップの入力と出力

全体の入力、Parallelの入力

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

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

MapAddの入力

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

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

Add[0]の入力

{
  "num0": 1,
  "num1": 1
}

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

Add[0]の出力

{
  "num": 2
}

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

MapAddの出力、Sum1の入力

[
  {
    "num": 2
  },
  {
    "num": 3
  },
  {
    "num": 7
  }
]

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

Sum1の出力

{
  "num": 12
}

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

Sum2の出力

{
  "num": 4
}

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

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サービスを処理フローに組み込めるので、興味がある方は挑戦してみてください。

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