この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。
はじめに
StepFunctionsを使ってSQSに大量のメッセージを送信したかったのでAWS SDK integrationsのSendMessageBatchを試してみました。
ワークフローの全体像
全体像は以下の図の通りです。Lambda関数で作ったデータをSendMessageBatchでSQSに送信します。SendMessageBatchの仕様を踏まえたポイントは以下の通りです。
- SendMessageBatchでは各メッセージのIDを送信側で指定する必要があります
- SendMessageBatchで一度に送信できるメッセージは10件までです
- Mapで並列処理したいのでメッセージを10件ごとのネストしたリストにします
Lambda関数
上記を踏まえてダミーのデータを生成する以下のような関数を作りました。
※注: このコードだと実際に大量のデータを生成すると実行時間がかなり長くなります
import json
import uuid
def generate_messages(n:int):
return [{'Id': str(uuid.uuid4()), 'MessageBody': str(uuid.uuid4()) } for _ in range(n)]
def lambda_handler(event, context):
return [generate_messages(10) for _ in range(5)]この関数の実行結果は以下のようになります。リストの要素のキー名はSendMessageBatch と一致するようにしています。
[
[
{
"Id": "00ca4544-98ae-4f41-9437-c92e9f8fd8ee",
"MessageBody": "fee67a60-1f64-4dab-a995-8cd8b221392b"
},
{
"Id": "6b249be1-a14f-43d5-b1f0-7770f221d427",
"MessageBody": "5674575a-8c0f-45f3-b5c4-c19770022bf9"
},
{
"Id": "a72abc8d-130f-4053-a602-a1d2d43e3b41",
"MessageBody": "401d8afb-17c4-4489-a37a-d65f07ddae5f"
}
],
[
{
"Id": "c5cdfb14-fd78-42b2-a98f-3184bda7eefa",
"MessageBody": "b273ebbf-5ba1-48ae-a03c-75494cb32542"
},
{
"Id": "73860e6a-31e4-4bbb-8a1c-57af5bb8244f",
"MessageBody": "4415227d-717c-44eb-82c4-9a8e529acea1"
},
{
"Id": "3405839d-51af-403e-ac29-fcbb38bb34cd",
"MessageBody": "a835abaf-df76-41eb-9f5c-c232f68855b9"
}
]
]メッセージ送信 State
メッセージ送信を行うStateは以下のようになります。今回指定したパラメータは以下の通りです。
- QueueUrl キューのURL
- Entries メッセージのリスト（Lambdaの出力のリストの１要素を指定）
{
"Type": "Map",
"Iterator": {
"StartAt": "SendMessageBatch",
"States": {
"SendMessageBatch": {
"Type": "Task",
"End": true,
"y": {
"Entries.$": "$",
"QueueUrl": "https://sqs.ap-northeast-1.amazonaws.com/xxxxxxx/MyQueue"
},
"Resource": "arn:aws:states:::aws-sdk:sqs:sendMessageBatch"
}
}
},
"End": true
}SendMessageBatchステートの実行結果は以下のようになります。
{
"name": "SendMessageBatch",
"output": {
"Successful": [
{
"Id": "38633e7c-d26b-4c81-bdf3-b322f803b2f6",
"Md5OfMessageBody": "c3683ad70e1323ca7d94f43bb9b93e48",
"MessageId": "46c6e844-a260-4b6e-94f5-50a5c8f1a127"
},
{
"Id": "62c77d52-1224-4071-a08f-d610e348e809",
"Md5OfMessageBody": "28a1384230bea8b9f2cddfe281a900d9",
"MessageId": "5961faa8-50a9-41bc-9581-4e63eee0f092"
},
{
"Id": "56f40da7-977e-455f-8727-bf6a36849e2e",
"Md5OfMessageBody": "5811b32e4a99ebf010e2e3126702b7ff",
"MessageId": "b4fd172b-7399-4bd7-9f0b-f0c3020e02f6"
},
{
"Id": "59d07a20-ceee-4d0c-abb0-0bb70c2e78c0",
"Md5OfMessageBody": "a252facc027f419d952fe9380c7ce0f8",
"MessageId": "43a8fbd0-ea76-465d-8d02-3b022f7452ad"
},
{
"Id": "ab7cd096-9d24-4006-8d8b-e8436298fa01",
"Md5OfMessageBody": "9cc21716708dfe67ce1f9d9ac7f49f7c",
"MessageId": "b54f1739-e7b3-4956-a291-27302d2000a0"
},
{
"Id": "4bf9d1bf-4a5d-4090-9439-23dd79314017",
"Md5OfMessageBody": "ad87092f60055e3e7eb56428dc12f3b0",
"MessageId": "0f410434-d455-4ff3-b4d4-8d5040db6ee1"
},
{
"Id": "937d4b83-9cda-44bc-bbb9-83546780ccad",
"Md5OfMessageBody": "4ec8b0276c957e7eb77f168f8bf14c7e",
"MessageId": "df546f4d-7d7a-4df7-b587-d10e5321271b"
},
{
"Id": "cef988b1-b5a4-4f17-bd2d-00855f9f871c",
"Md5OfMessageBody": "6417372c284ee74d55bef32a5aef1521",
"MessageId": "78b6d202-8edc-41f7-845c-a8b6c2ffddac"
},
{
"Id": "e92ee03e-3f66-41af-96d3-c0d109a11a97",
"Md5OfMessageBody": "1a919be2c5062a74d06ff8a54cdebf81",
"MessageId": "5a6b008d-6f8b-4556-b4bf-ceffb1083d30"
},
{
"Id": "f3a32103-0056-4752-8197-51452431171a",
"Md5OfMessageBody": "87208e026194f7200291f946b0833096",
"MessageId": "c9b88071-299d-4468-a4d3-d9513ed516fb"
}
]
},
"outputDetails": {
"truncated": false
}
}まとめ
Lambda関数で生成したデータをSendMessageBatch のパラメータをどう指定したらいいのか少し悩みましたが、データをSendMessageBatchの仕様に合わせることで解決しました。Optimized integrationであるSendMessageとの性能の差やそれ以外の比較点が気になるので機会があれば比較してみたいと思います。
4
