Step Functionsを使って初めてループや分岐をやってみた！

#AWS Step Functions

#AWS

芳賀健

2018.10.15

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは。芳賀です。
入社してから３カ月間、オンプレ畑で育ったエンジニアがひたすらAWSと格闘して、初めてサーバレスなアプリケーションを作り、Step Functionsを覚えたので、ちょっとブログにしてみました。

AWS Step Functionsとは？

AWSのサーバレスサービスのLambdaを使った事はあるでしょうか？
サーバレスなLambdaですが、Lambda関数から別なLambda関数を呼ぶ場合にはプログラムコードで次に呼び出すLambda関数名を設定したり、何らかのアクションを間に挟む必要がありました。このため、ちょっとしたカスタマイズでもプログラムの改造が必要になりました。今回、ブログへ書いたAWS Step Functionsを利用すれば、その必要がなくなります。雑な言い方をすれば、個々のLambda関数を簡単につなぐためのAWSサービスになります。
Lambda関数については、下記の記事を参考に作成してみてください。
初めてのサーバーレスアプリケーション開発～LambdaでDynamoDBの値を取得する～

このブログでやってみること

今回は、３つの事を試してみます。
１． Step FunctionsからLambdaを呼び出す
２． Lambda間のデータの渡し方
３．分岐を使ったループ処理

Step FunctionsからLambdaを呼び出す

まずはLambdaを作成し、StepFunctionsから呼び出してみましょう。
Step Functionsから呼び出すLambda関数を作成します。
・関数名 : SampleCall
・ランタイム : Python 3.6
・IAMロール : （Lambdaが実行できるロールを設定してください）

import json
def lambda_handler(event, context):
    inta = 1
    intb = 2
    return {
        "statusCode": 200,
        "returnCount": inta + intb,
        "body": json.dumps('Hello from Lambda!')
    }

次にStep Functionsを作る
・名前 : SampleStepFunctions
・IAMロール : （Lambdaが実行できるロールを設定してください）
・ステートマシンの定義

{
    "Comment": "Sample AWS Step functions flow",
    "StartAt": "TaskSampleCall",
    "States": {
        "TaskSampleCall": {
            "Comment": "SampleCallの呼び出し",
            "Type": "Task",
            "Resource": "arn:aws:lambda:ap-northeast-1:(AWSユーザーID):function:SampleCall",
            "InputPath": "$",
            "ResultPath": "$.SampleCallResult",
            "OutputPath": "$",
            "End": true
        }
    }
}

・下記のようなStepFunctionsのフロー図が出来上がります

作成したStep Functionsを実行するには「実行の開始」を押してください。Step Functionsは実行する毎に実行名を付けて走らせます。
・実行名 : State-20181015112154（←適宜に名前を付けてください）
・入力パラメータ : 今回はデフォルト値とします

{
    "Comment": "Insert your JSON here"
}

実行ステータスが「成功」であれば、Step Functniosがエラーなく実行されています。実行結果として、下記のように出力されていれば成功です。

{
  "Comment": "Insert your JSON here",
  "SampleCallResult": {
    "statusCode": 200,
    "body": "\"Hello from Lambda!\""
  }
}

Lambda間のデータの渡し方

さて、Lambda関数をStep Functionsから呼び出せたら、Lambda同士を連携させてみましょう。

連携させる１つ目の関数を作成します。
・関数名 : SampleSetting

import json
def lambda_handler(event, context):
    inta = 1
    return {
        "foo": inta
    }

連携させる２つ目の関数を作成します。
・関数名 : SampleAddition

import json
def lambda_handler(event, context):
    intb = 2
    return {
        "var": intb + event['SettingResult']['foo']
    }

連携させる３つ目の関数を作成します。
・関数名 : SampleMultiply

import json
def lambda_handler(event, context):
    intc = 3
    return {
        "baz": intc * event['AdditionResult']['var']
    }

続いて作成した３つのLambdaを組み合わせるStep Functionsを作ります
・名前 : SampleCalculationFlow

{
    "Comment": "Sample Calculation flow",
    "StartAt": "TaskInitialSetting",
    "States": {
        "TaskInitialSetting": {
            "Comment": "初期の値を設定します",
            "Type": "Task",
            "Resource": "arn:aws:lambda:ap-northeast-1:(AWSユーザーID):function:SampleSetting",
            "InputPath": "$",
            "ResultPath": "$.SettingResult",
            "OutputPath": "$",
            "Next": "TaskAddition"
        },
        "TaskAddition": {
            "Comment": "足し算をおこないます",
            "Type": "Task",
            "Resource": "arn:aws:lambda:ap-northeast-1:(AWSユーザーID):function:SampleAddition",
            "InputPath": "$",
            "ResultPath": "$.AdditionResult",
            "OutputPath": "$",
            "Next": "TaskMultiply"
        },
        "TaskMultiply": {
            "Comment": "掛け算をおこないます",
            "Type": "Task",
            "Resource": "arn:aws:lambda:ap-northeast-1:(AWSユーザーID):function:SampleMultiply",
            "InputPath": "$",
            "ResultPath": "$.MultiplyResult",
            "OutputPath": "$",
            "End": true
        }
    }
}

・下記のようなStepFunctionsのフロー図が出来上がります

このStep Functionsを下記のパラメータを設定して、実行します。実行する毎に実行名を付けて走らせます。
・実行名 : State-20181015113646（←適宜に名前を付けてください）
・入力パラメータ : 今回はデフォルト値とします

{
    "Comment": "Insert your JSON here"
}

下記のように出力されていれば成功です。

{
    "Comment": "Insert your JSON here",
    "SettingResult": {
        "foo": 1
    },
    "AdditionResult": {
        "var": 3
    },
    "MultiplyResult": {
        "baz": 9
    }
}

Step Functionsでは、Lambdaなどとデータをやり取りする場合、Pathと呼ばれる命令を駆使します。

`InputPath`：

Lambda関数へパラメータとして渡すデータを定義します。これによって、Lambda内のeventにて値を取得する事ができます。

`ResultPath`：

Lambda関数からのreturn値を定義します。
今回は結果を分かりやすくするため、全て別な名前を付けました。同じ名前を付けた場合には、その値へ上書きする事もできます。

`OutputPath`：

Step Functionsで次のステートへ渡したいデータを定義します。今回は、$としました。これはInputPathで受け取ったパラメータを全て次のステートへ引き渡す事を意味しています。必要なパラメータだけ渡す場合には、中身のJSON項目名を指定する事も可能です。

さらに詳しく確認したい場合には、AWS 開発者ガイドに InputPath および OutputPath を使用したフィルタリングがありますので、こちらも参考にしてください。

分岐を使ったループ処理

ここまでは一気通貫のステートマシンを作成していましたが、分岐を使ったループについてお話します。
まず、ループ処理でカウントアップとループ継続チェックするLambda関数を作成します。
・名前 : SampleCountUp

import json

def lambda_handler(event, context):

    idx = event['iterator']['idx']
    step = event['iterator']['step']
    maxcount = event['iterator']['maxcount']

    idx = idx + step
    if idx <= maxcount:
        is_continue = True
    else:
        is_continue = False

    # response data
    response = {
        "idx": idx,
        "step": step,
        "maxcount": maxcount,
        "continue": is_continue
    }
    return response

次にLambdaを組み合わせるStep Functionsを作る
・名前 : SampleLoopFlow

{
    "Comment": "Sample loop flow",
    "StartAt": "ConfigureCount",
    "States": {
        "ConfigureCount": {
            "Type": "Pass",
            "Result": {
                "idx": 1,
                "step": 1,
                "maxcount": 3,
                "continue": true
            },
            "ResultPath": "$.iterator",
            "Next": "wait_seconds"
        },
        "wait_seconds": {
            "Type": "Wait",
            "Seconds": 5,
            "Next": "ChoiceLoopStopState"
        },
        "ChoiceLoopStopState": {
            "Type": "Choice",
            "Choices": [{
                "Comment": "ループ継続フラグcontinueをチェックする",
                "Variable": "$.iterator.continue",
                "BooleanEquals": true,
                "Next": "CheckStatus"
            }],
            "Default": "SuccessProcess"
        },
        "CheckStatus": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:ap-northeast-1:(AWSユーザーID):function:SampleCountUp",
            "InputPath": "$",
            "ResultPath": "$.iterator",
            "OutputPath": "$",
            "Next": "wait_seconds"
        },
        "SuccessProcess": {
            "Type": "Succeed"
        }
    }
}

・下記のようなStep Functionsのフロー図が出来上がります

このStep Functionsを下記のパラメータを設定して、実行します。実行する毎に実行名を付けて走らせます。
・実行名 : State-20181015131826（←適宜に名前を付けてください）
・入力パラメータ : 今回はデフォルト値とします

{
    "Comment": "Insert your JSON here"
}

下記のように出力されていれば成功です。

{
  "Comment": "Insert your JSON here",
  "iterator": {
    "idx": 4,
    "step": 1,
    "maxcount": 3,
    "continue": false
  }
}

Step Functionsでは、Lambda等を呼び出すTaskを含め６つのStatesがあり、今回は、以下の４つを使ってループを実現しています。
・Pass：入力された値を単純に出力へ渡したり、固定データを出力します
・Wait：指定した時間（日時）まで処理を待たせます
・Choice：入力された値で条件選択します
・Succeed：成功として処理を停止させる

それを踏まえて、今回のループ処理を確認しますと。
ConfigureCountステートで"Type": "Pass"を用いて、ループを処理するためのカウンタidx、増分step、最大ループ回数maxcount、ループ継続フラグcontinueを定義しています。

        "ConfigureCount": {
            "Type": "Pass",
            "Result": {
                "idx": 1,
                "step": 1,
                "maxcount": 3,
                "continue": true
            },

wait_secondsステートでは、人が処理の流れを追い易くするために待ち時間を設定しています。待ち時間は５秒です。

        "wait_seconds": {
            "Type": "Wait",
            "Seconds": 5,

ChoiceLoopStopStateステートでは、ループ継続フラグcontinueをチェックし、ループを継続するか判断分岐してます。注意点として無限ループにならないよう条件設定には気を付けてください。

        "ChoiceLoopStopState": {
            "Type": "Choice",
            "Choices": [{
                "Comment": "ループ継続フラグcontinueをチェックする",
                "Variable": "$.iterator.continue",
                "BooleanEquals": true,
                "Next": "CheckStatus"
            }],
            "Default": "SuccessProcess"
        },

CheckStatusステートは、Lambda関数を呼んでいます。Lambda内部では実装コードの通り、カウンタidxのカウントアップや継続判断などの処理が行われます。

Statesの"Type"について詳しく確認したい場合には、AWS 開発者ガイドにStep Functions の詳細 » Statesがありますので、こちらも参考にしてください。

Step Functionsの勘所

今回、わたしがStep Functionsを使ってハマった部分や実装する上での注意点です。

ステートマシンの中で流れるデータをしっかり設計する
input/output/resultの動作を把握する
分岐で戻るステップを変えることでループが組める
プログラムの大きな条件分岐などは、Step Functionsに任せる
ただしループカウンタのカウントアップはLambda関数内で処理が必要
Serverless frameworkの利用を考えるならJSONからYAMLへの書き換えが必要
CI/CDについて今回、説明しませんでしたが、Step FunctionsのフローはJSONですがServerless frameworkでデプロイする場合にはYAMLでステートマシンを記述する必要があります