イベントソースマッピングとストリーム処理

AWS Lambda によるサーバーレスアプリケーション

Claudio Canales

Senior DevOps Engineer

イベントソースマッピングの位置づけ

キューとストリームは、バッチ単位でポーリングされます。
イベントソースマッピングは、ポーリングとバッチ処理を管理します。

イベントソースマッピングの位置

イベントソースマッピングとは

ソースと関数をつなぐリソースです。
Lambda はソースをポーリングしてレコードを取得します。
Lambda はバッチとしてハンドラーを呼び出します。

イベントソースマッピングのフロー

プッシュとポーリングの違い

プッシュモデル

プッシュ型のソースは、イベントを即座に送信します。
例：Amazon S3 へのアップロード。

ポーリングモデル

ポーリング型のソースは、Lambda が定期的に確認します。
Lambda はバッチを構築してハンドラーを呼び出します。
レイテンシとリトライの動作が異なります。

プッシュとポーリングのモデル比較

キューとストリームの違い

キュー（Amazon SQS）

受信トレイのようなマネージドメッセージキューです。

ストリーム（DynamoDB Streams）

DynamoDB テーブルの変更ログです。
どちらも Records として届きますが、意味は異なります。

キューとストリームの比較

バッチイベントの構造（簡略版）

{
  "Records": [{
    "messageId": "abc-123",
    "body": "{\"order_id\": \"A-42\"}"
  }]
}

ほとんどのマッピングイベントは Records で始まります。
各レコードには messageId と body 文字列があります。
body を解析して独自のペイロードとして使用します。

ウォークスルー：各レコードの処理

def lambda_handler(event, context):
    records = event.get("Records", [])
    for record in records:
        body = record.get("body", "")
        print("BODY:", body)
    return {"statusCode": 200}

Records をデフォルトのリストで取得します。
各レコードをループして body を安全に読み取ります。
必要な情報をログに記録してから返します。

ウォークスルー：JSON ボディの解析

import json

def lambda_handler(event, context):
    record = event.get("Records", [])[0]
    payload = json.loads(record.get("body", "{}"))
    order_id = payload.get("order_id")
    print("ORDER_ID:", order_id)
    return {"statusCode": 200}

json をインポートし、安全なデフォルト値で body を取得します。
json.loads で解析してディクショナリを取得します。
必要なフィールドを取り出す前に、必ず検証しましょう。

バッチ処理：サイズとウィンドウ

バッチサイズ：1 回の呼び出しで処理するレコード数です。
バッチウィンドウ：バッチを満たすまで Lambda が待機する時間です。
バッチを大きくするとスループットが向上しますが、1 回あたりの処理量が増えます。
SQS では、バックログの確認に ApproximateAgeOfOldestMessage を活用しましょう。

バッチサイズのトレードオフ図

バッチサイズのトレードオフ

バッチが大きいと効率的ですが、レイテンシが増加します。
バッチが小さいと処理開始が早くなりますが、呼び出し回数が増えます。
ワークロードに応じて選択しましょう。

バッチサイズとレイテンシのグラフ

同時実行によるスケーリング

複数のバッチが同時に処理されます。
同時実行数を増やすとスループットが向上します。
ただし、ダウンストリームへの負荷も増加します。

バッチの並列処理

ダウンストリームシステムの保護

同時実行数が多すぎると、データベースや API に過負荷がかかります。
制限を設けて負荷をコントロールしましょう。
小さく安全な処理単位を優先しましょう。

ダウンストリーム負荷の保護

バッチの部分的な失敗

バッチには正常なレコードと異常なレコードが混在することがあります。
部分的な失敗を処理しないと、1 件の不正レコードでバッチ全体がリトライされます。
部分的な失敗処理を使うと、失敗したレコードだけを再試行できます。

部分的なバッチ失敗のフロー

ウォークスルー：部分的な失敗の報告（SQS）

def lambda_handler(event, context):
    failures = []
    for record in event.get("Records", []):
        try:
            process(record)
        except Exception:
            failures.append({"itemIdentifier": record["messageId"]})
    return {"batchItemFailures": failures}

失敗したレコードの messageId を batchItemFailures に収集します。
Lambda はそのレコードのみをリトライします。
正常に処理されたレコードは再送されません。

まとめ

イベントソースマッピングはキューとストリームをポーリングします。
Records をバッチとしてハンドラーに渡します。
バッチサイズとウィンドウはレイテンシとスループットのトレードオフです。
部分的な失敗処理により、成功済みレコードの再処理を防ぎます。

マッピングのまとめ図

練習しましょう！

AWS Lambda によるサーバーレスアプリケーション