AWS LambdaがSQSをイベントソースとしてサポートしました！

2018年06月29日
大栗宗（135）
0

大栗です。

先程AWS LambdaでSQSをイベントソースとしてサポートすることが発表されました。このエントリーではSQSからのLambdaの起動方法についてレポートします。

Announcement: AWS Lambda supports Amazon SQS as an event source

Lambdaのイベントソース

AWS Lambdaは以下のサービスのイベントを元に起動することができます。今回SQSのメッセージをトリガーとしてLambdaを起動することが可能になります。

API Gateway
AWS IoT
Alexa Skills Kit
Alexa Smart Home
CloudWatch Events
CloudWatch Logs
CodeCommit
Cognito Sync Trigger
DynamoDB
Kinesis
S3
SNS
SQS <- New!

SQSは気軽にメッセージを投げて非同期で処理するというケースが多くありましたが、EC2などでWorkerを立てたりLambdaを定期実行してポーリングをする必要があり面倒でした。しかし、今回の発表によりSQSへメッセージが投入されたことをトリガーとしてLambdaが自動起動して処理を開始できるようになりました。

注意点として、現時点で対応しているSQSのタイプは標準キューのみとなりFIFOキューには対応していません。

SQSをイベントトリガーにするときの動作

SQSがイベントソースの場合、Lambdaは自動でスケールして最大の自動実行数までスケールします。自動実行数のスケールはSQSのメッセージがからの場合にポーリングコストを抑えつつ、SQSに大量にメッセージがある場合に高いスループットを達成できるように設計されています

Amazon SQSのイベントソースマッピングが最初に有効になっている場合、またはメッセージが無い場合、Lambdaは5つのクライアントを使用してAmazon SQSキューを並行してロングポーリングします。
Lambdaはインフライトのメッセージ数を監視して数の増加を検出すると、毎分20回のReceiveMessageリクエストコールを増加させてLambda関数の同時実行数を毎分60増加させます。
ポーリングの頻度はReceiveMessageリクエストを同時に100回まで達し、Lambda関数の同時実行数は1,000までになります。
アカウントの同時実行数は最大になります。
存在するSQSキューに接続されているLambda関数の同時実行数の制限に達します。

AWS Lambdaはインフライトなメッセージの減少を検知すると、分20回のReceiveMessageリクエストコールを減らし、Lambda関数の同時実行数を毎分30減少させます。

イベントの内容

LambdaのテストイベントでSQSのイベントを確認すると以下の内容になっています。Lambda関数で処理を書くときの参考にして頂ければと思います

{
  "Records": [
    {
      "body": "Hello from SQS!",
      "receiptHandle": "MessageReceiptHandle",
      "md5OfBody": "7b270e59b47ff90a553787216d55d91d",
      "eventSourceARN": "arn:aws:sqs:ap-northeast-1:123456789012:MyQueue",
      "eventSource": "aws:sqs",
      "awsRegion": "ap-northeast-1",
      "messageId": "19dd0b57-b21e-4ac1-bd88-01bbb068cb78",
      "attributes": {
        "ApproximateFirstReceiveTimestamp": "1523232000001",
        "SenderId": "123456789012",
        "ApproximateReceiveCount": "1",
        "SentTimestamp": "1523232000000"
      },
      "messageAttributes": {}
    }
  ]
}

試してみる

早速サンプルの処理を書いてみます。以下の流れで試してみます。

SQSの作成
Lambda用Roleの作成
Lambda関数の作成
実行確認

ここでは東京リージョンでManagementConsoleからの操作を前提とします。通常のLambda関数の開発ではAWS SAMやServerless Frameworkなどを使用すると思いますが、簡素化のためにConsoleを利用します。また作成するLambda関数のらいいタイムはPython 3.6とします。

SQSの作成

SQSを作成します。

SQSのConsoleで新しいキューの作成をクリックします。

lambda-queueという名称でキューを作成します。ここではキューの詳細設定をデフォルトのままとします。

キューが作成されました。

Lambda用Roleの作成

次にLambda用のRoleを作成します。

IAMのConsoleでロールの作成をクリックします。

AWS サービスを選択してLambdaを選んで次のステップ: アクセス権限をクリックします。

次にAdministratorAccessをアタッチします。アタッチしたら次のステップ: 確認をクリックします。（本来は必要な権限のみ設定して下さい。検証中に必要な権限の詳細が判断できなかったのでAdministratorAccessにしています。）

lambda-queueという名前を付けてロールの作成をクリックします。

Lambda関数の作成

Lambda関数を作成します。

LambdaのConsoleで関数の作成をクリックします。

一から作成を選択します。以下の内容を入力して関数の作成をクリックします。

項目	内容
名前	test-sqs
ランタイム	Python 3.6
ロール	既存のロールを選択
既存のロール	lambda-queue

DesignerでSQSをクリックします。

以下の内容設定して、追加をクリックします。

項目	内容
SQS キュー	lambda-queue
バッチサイズ	1
トリガーの有効化	チェック

次に関数コードを入力します。サンプルの関数はチュートリアル通り以下の内容として保存します。ここでSQSメッセージの削除処理を入れていないことが分かります。

from __future__ import print_function
 
def lambda_handler(event, context):
    for record in event['Records']:
       print ("test")
       payload=record["body"]
       print(str(payload))

実行確認

実行して確認します。

SQSのConsoleでlambda-queueをして、[キューの操作]-[メッセージの送信]をクリックします。

サンプルメッセージとしてtest1と入力しメッセージの送信をクリックします。

CloudWatch Logsで実行結果を確認します。

CloudWatchのConsoleでログを選択します。そして/aws/lambda/test-sqsのロググループを参照します。

作成されているログストリームの内容を見ると、以下のようにLambda関数が実行されていることが分かります。

さいごに

SQSを利用して処理を非同期に実行するアーキテクチャはよく使用されると思いますが、今まではEC２などを使用する必要がありました。しかし、今回LambdaでSQSイベントソースが対応したことで小さなな処理などをLambdaで処理することが可能になりました。これからSQS -> Lambdaというパターンは多用されると思いますので、今携わっているシステムで活用できる箇所があるか検討してみると良いと思います。

前の記事へ