

【AWS Step Functions】Lambdaの呼び出しと分岐処理
こんにちは。プラットフォーム技術部の渋谷です。
「AWSで複数処理をワークフロー的に実行させたい、しかもサーバーレスで」というニーズは多いのではないでしょうか。
今回はそういったニーズを実現するためのサービス AWS Step Functions について紹介します。
また、基本的なワークフローの書き方についても説明します。
実装にはJSON形式のコード ASL(Amazon States Language) を使用しますが、使ったことがない人でも簡単にできます。
AWS Step Functionsについて
AWS Step Functionsとは
AWS Step Functions はサーバーレスのオーケストレーションサービスです。
ビジュアルワークフローを用いて、分散アプリケーションおよびマイクロサービスのコンポーネントを容易に調整できます。
詳しくは、AWS公式の開発者ガイドをご覧ください。
Step Functionsが必要とされる理由
アプリケーション開発のベストプラクティスとしてよく取り上げられる「The Twelve-Factor App」の「VI. プロセス」では、次のように記載されています。
Twelve-Factorのプロセスはステートレスかつシェアードナッシング である。永続化する必要のあるすべてのデータは、ステートフルなバックエンドサービス(典型的にはデータベース)に格納しなければならない。
https://12factor.net/ja/processes
これは「実行されるプロセスが実行期間を超えてファイルシステムやメモリ変数に依存しないように設計する」という原則です。
しかし、多くのシステムでは複数のプロセスで動作しており、相互にやりとりが発生します。
Twelve-Facterの原則に基づけば、プロセスの状態はRDBなどの永続ストレージに保存しておく必要があり、独自に実装しようとするとかなり大変です。
このような複数のプロセスや関数をオーケストレーションしてくれる便利なサービスが AWS Step Functions です。
ユースケース
Step Functions を利用することで、次のような要望にも応えられます。
- 順序性がある複数のプロセスを同期実行させたい。
- スループット向上のため複数のプロセスを並列実行させたい。
- 状態を次のプロセスに連携させたい。
- 特定の条件下で別のプロセスに分岐させたい。
- 特定の条件下でリトライさせたい。
- 例外やエラーを補足したい。
今回は、特定条件下でプロセスを分岐させる処理について主に記載します。
Lambdaを呼び出す
早速 Step Functions を触っていきましょう。
Step Functions はステートマシンとタスクに基づいています。
ステートマシンはワークフローを意味し、タスクはワークフロー内のステートを意味します。
ステートマシンは、ASLを使用して定義されます。
{
"Comment": "A Hello World example of the Amazon States Language using Pass states",
"StartAt": "Hello",
"States": {
"Hello": {
"Type": "Pass",
"Result": "Hello",
"Next": "World"
},
"World": {
"Type": "Pass",
"Result": "World",
"End": true
}
}
}
ASLで記載されているフィールドの説明は次の通りです。
- Comment(オプション)
ステートマシンの説明。 - StartAt(必須)
Statesオブジェクトのいずれかの名前を指定。参照されているStatesから開始される。 - TimeoutSeconds(オプション)
ステートマシンを実行できる最大秒数。 - Version(オプション)
ステートマシンで使用されるASLのバージョン。デフォルトは1.0。 - States(必須)
ステート。"End": true フィールドがある場合、実行が停止して結果を返す。それ以外の場合、"Next": フィールドを探し、その状態を続行。
また、Statesには以下の8種類のTypeがあります。
上記の定義では固定の文字列を出力するためStates “Hello", “World"のTypeはどちらも"Pass"となっています。
- Task:ステートマシンで何らかの作業をする。
- Choice:実行の分岐を選択する。
- Succeed:成功で実行を停止する。
- Fail:失敗で実行を停止する。
- Pass:入力を出力に渡す。または、一部の固定データを出力する。
- Wait:一定時間待機する。または指定された日付・時刻まで待機する。
- Parallel:分岐を並列実行する。
- Map:ステップを動的に反復する。
Lambdaの作成
呼び出すLambdaを作成します。ランタイムにはPython 3.8を選択します。
import json
def lambda_handler(event, context):
return {
"statusCode": 200,
"body": json.dumps('Hello from Lambda!')
}
ステートマシンの作成
Lambdaを呼び出すステートマシンの定義は以下のように記載します。
今回は単純にLambdaを呼び出すだけですので、States “CallFunction"のTypeは"Task"としています。
{
"Comment": "Call Lambda Function.",
"StartAt": "CallFunction",
"States": {
"CallFunction": {
"Type": "Task",
"Resource": "<LambdaARN>",
"End": true
}
}
}
ステートマシンを作成したら「実行の開始」を押して実行してみましょう。
名前や入力値の項目が表示されますが、今回はデフォルトのままでOKです。
実行が正常に終了すると「実行ステータス」が成功に変化します。
「実行出力」タブを開くと、作成したLambdaの実行内容が出力されていることを確認できます。
{
"statusCode": 200,
"body": "\"Hello from Lambda!\""
}
条件分岐
Lambdaの呼び出しに成功しましたので、次はLambdaの結果によって分岐させます。
分岐ロジックを追加するには、Choice 状態 (“Type": “Choice")を利用します。
Choiceでは、次の追加フィールドがあります。
- Choices (必須)
ステートマシンが次に移行する状態を決定する Choice ルールの配列。 - Default (オプション、推奨)
Choices のいずれの移行も実行されない場合の移行先の状態の名前。
Lambdaの作成
先ほどと同様、まずはLambdaを作成します。
ランダムで1-100の数字を出力するようなLambdaを作成しました。ランタイムはPython3.8を選択します。
import random
def lambda_handler(event, context):
return {
"foo": random.randint(1,100)
}
ステートマシンの作成
ステートマシンの定義は「50以下」と「51以上」で分岐するように書いてみました。
{
"StartAt": "FirstState",
"States": {
"FirstState": {
"Type": "Task",
"Resource": "<LambdaARN>",
"Next": "ChoiceState"
},
"ChoiceState": {
"Type" : "Choice",
"Choices": [
{
"Variable": "$.foo",
"NumericLessThan": 50,
"Next": "Under50"
},
{
"Variable": "$.foo",
"NumericGreaterThanEquals": 51,
"Next": "Over51"
}
],
"Default": "DefaultState"
},
"Under50": {
"Type" : "Pass",
"Result" : "Under50",
"Next": "LastState"
},
"Over51": {
"Type" : "Pass",
"Result" : "Over51",
"Next": "LastState"
},
"DefaultState": {
"Type": "Fail",
"Cause": "Not Match"
},
"LastState": {
"Type": "Pass",
"End": true
}
}
}
実行
作成したステートマシンを実行してみます。
正常に処理が完了した場合、グラフインスペクターが表示されます。
「Under50」か「Over51」、どちらのステートを経由してきたが目視で確認できます。
また、グラフインスペクターの「FirstState」ステップを選択すると右側にステップの詳細を確認する項目が表示されます。
「ステップ出力」タブを選択すると実際にLambdaで出力された数字を確認できます。


また、And や Or を使用することにより、より複雑な条件による分岐を行うこともできます。
以下の定義では「31以上、81未満」と「31未満または81以上」で分岐をしています。
"ChoiceState": {
"Type": "Choice",
"Choices": [
{
"And": [
{
"Variable": "$.foo",
"NumericGreaterThanEquals": 31
},
{
"Variable": "$.foo",
"NumericLessThan": 81
}
],
"Next": "Between31-80"
},
{
"Or": [
{
"Variable": "$.foo",
"NumericLessThan": 31
},
{
"Variable": "$.foo",
"NumericGreaterThanEquals": 81
}
],
"Next": "Under31 or Over81"
}
],
"Default": "DefaultState"
},
最後に
今回はLambdaの呼び出しとChoiceを使った分岐について記載しました。
ASL について親しみがない人も多いと思いますが、自動でコードエラーチェックしたり、ワークフローの実行結果がグラフィカルに表示されたりと、間違いにすぐ気づけるようになっています。
プログラミングをしたことがない人でもとっつきやすいので、ぜひいろいろなワークフローを考えながら Step Functinos を作成してみてください。
Step Functions には今回記載した内容だけでなくたくさんの機能があります。
次回は、Parallelを使った並列処理やMapを使った動的並列処理について紹介したいと思います。