Serverless本番運用入門 Vol2 — Event-Driven Architecture (EventBridge × SQS × SNS × Kinesis)

1. なぜEvent-Driven本番運用か — Vol1 同期処理からの架橋 + 第12軸結節点

IAM・EKS・復旧・AI・セキュリティ・コスト・マルチアカウント・Observability・Network・DevOps・Database の全11軸 23記事、そして第12軸 Serverless Vol1 (Lambda × API Gateway × Step Functions) を完遂したエンジニアが次に直面する壁 — それが 同期処理アーキテクチャの限界 です。Lambda の 15 分タイムアウト、API Gateway の 29 秒制限、Step Functions のステート爆発、サービス間の直結結合が引き起こす障害伝播。これらは「コードを最適化する」では解決できず、アーキテクチャパラダイムの転換が必要です。本 Vol2 では Event-Driven Architecture (EDA) の4本柱 — EventBridge / SQS / SNS / Kinesis Data Streams — を全12軸統合視点で体系化し、本番品質のイベント駆動運用パターンを確立します。

同期処理の限界5選

限界1: Lambda タイムアウト連鎖 (最大 15 分)
外部 API への同期呼び出しが遅延すると呼び出し元 Lambda がタイムアウトし、さらにその呼び出し元もタイムアウトする「タイムアウト連鎖」が発生します。SQS を挟んで処理を分離することで伝播を断ち切れます。

限界2: API Gateway 29 秒制限
API Gateway 統合タイムアウトは最大 29 秒です。重い処理を同期実行すると 29 秒超過で即タイムアウト。Event-Driven パターンでは即 202 Accepted を返し、処理は SQS 経由で非同期実行することで回避します。

限界3: スケール不均衡
高負荷スパイク時、受付側と処理側の同時実行上限の差が本番停止を引き起こします。SQS をバッファとして挟み込むことで、受け付け側と処理側のスケールを独立制御できます。

限界4: サービス間直結結合による障害伝播
Lambda から Lambda を直接 invoke するアーキテクチャでは、呼び出し先の障害が即座に伝播します。EventBridge / SQS を介した疎結合設計により、障害の伝播を遮断します。

限界5: 処理待ち中のコスト浪費
外部 API 応答待ち中も Lambda の課金が継続します。非同期 Event-Driven 設計により、実際の処理時間のみ課金される効率的なアーキテクチャを実現します。

Vol1 からの進化軸 — Synchronous → Event-Driven

Serverless Vol1 で確立した同期3本柱 (Compute / API / Workflow) の上に、本 Vol2 でイベント駆動層を重ね、Synchronous + Event-Driven のハイブリッド本番アーキテクチャを完成させます。

Event-Driven が解決する4本問題

Decoupling — 疎結合化
同期アーキテクチャではサービスA→B→Cの直結チェーンにより、1サービスの変更・障害が全体に影響します。EventBridge や SQS を挟んだ Event-Driven 設計では、Publisher は Consumer の存在を知らず独立してイベントを発行します。Consumer の追加・削除・変更はイベントのスキーマを維持する限り Publisher への影響ゼロで実施できます。

Scalability — 独立スケール
同期呼び出しでは呼び出し元と呼び出し先の処理速度が直接結びついているため、一方のスケールアップが他方を巻き込みます。SQS / Kinesis を挟むことで、発行側は書き込み性能のみ、消費側は処理性能のみを最適化すれば良くなります。週次バッチの大量イベントを SQS に蓄積しながら Lambda を段階的にスケールアウトするパターンが典型例です。

Resilience — 耐障害性
同期チェーンでは下流の一時障害が即時上流エラーに伝播します。SQS + DLQ の組み合わせでは、下流 Lambda が一時障害で失敗してもメッセージは SQS に残留し、復旧後に再処理できます。DLQ はその永続失敗メッセージを隔離し、障害範囲を最小化します。

Asynchrony — 非同期処理
リクエストの受付と処理を分離することで、受付 API は常に高速応答 (202 Accepted) を返せます。ユーザーへのレスポンスを待たせることなくバックグラウンドで非同期処理が完了するため、体感 UX とシステムスループットの両方が向上します。

同期 vs 非同期 選定基準

本 Vol2 で得られる4つの成果

1. 4本柱選定フレームワーク — EventBridge / SQS / SNS / Kinesis Data Streams の役割・スループット・順序保証・コストの4軸で最適サービスを選定する判断フロー。

2. EventBridge 本番運用 — Default Bus / Custom Bus / Rule / Pipes / Scheduler の全機能を実運用設定付きで解説。Cross-Account イベント中継・Schema Registry・Retry Policy 設計を Terraform テンプレートで提供。

3. SQS 本番運用 — Visibility Timeout × 6 原則・DLQ 設計・BatchItemFailures・FIFO vs Standard 選定・Lambda ESM 同時実行制御を実コード付きで解説。

4. Kinesis / SNS 本番運用 — Shard 設計式・Enhanced Fan-out・Hot Partition 回避 (Kinesis) および Subscription Filter Policy・Cross-Region Fanout・FIFO Topic (SNS) の本番品質設定を提供。

Vol2 学習ロードマップ

本 Vol2 は以下の章構成で Event-Driven Architecture の全領域をカバーします。§2 の選定マトリクスで4本柱の使い分けを把握してからサービス別詳解 (§3〜§6) を読むと理解が深まります。

Vol1: Lambda × API Gateway × Step Functions を読む

2. AWS Event-Driven 4本柱整理 + 選定マトリクス + 用途別決定木

AWS の Event-Driven エコシステムは4つのマネージドサービスで構成されます。EventBridge はイベントルーティングの中枢、SQS は非同期デカップリングキュー、SNS はパブリッシュ/サブスクライブ、Kinesis Data Streams はリアルタイムストリーミングを担います。4本柱の選定を誤ると、順序保証の欠如・Hot Partition 障害・過剰配信コストなどの典型的な本番障害に直結します。

4本柱 選定マトリクス (6軸評価)

各サービス概要

EventBridge — イベントルーティング中枢
EventBridge は AWS サービス・カスタムアプリ・SaaS パートナーからのイベントを受け取り、Event Pattern に基づいて複数の Target (Lambda / SQS / SNS / Step Functions 等) にルーティングします。Schema Registry でイベント仕様を型定義でき、Pipes によるノーコード ETL、Scheduler による柔軟なスケジュール実行も提供します。AWS サービス間イベント中継の第一選択です。

SQS — 非同期デカップリングキュー
SQS は Producer と Consumer の間にキューを挟み込み、非同期・デカップリング処理を実現します。Standard キュー (高スループット・at-least-once) と FIFO キュー (順序保証・重複排除) の2モードがあります。Lambda Event Source Mapping との組み合わせで、Visibility Timeout 管理・DLQ による失敗メッセージ隔離・BatchItemFailures による部分失敗ハンドリングが本番品質の鍵です。

SNS — パブリッシュ/サブスクライブ
SNS は1つの Topic に Publish されたメッセージを複数の Subscriber に一斉配信 (fan-out) します。Subscription Filter Policy で Subscriber ごとに配信メッセージを絞り込めます。SNS → SQS の組み合わせにより、ファンアウト後に各 Subscriber が独立してキュー処理できる耐久性の高いアーキテクチャを実現します。

Kinesis Data Streams — リアルタイムストリーミング
Kinesis Data Streams は Shard ベースのストリーミング基盤です。1 Shard あたり Write 1 MB/sec または 1,000 records/sec、Read 2 MB/sec のスループットを提供します。保持期間は 24 時間〜365 日で保持期間内であればストリームの先頭から再処理できます。Enhanced Fan-out (EFO) により複数 Consumer が独立した 2 MB/sec スループットで並列処理できます。

Terraform 例: EventBridge Event Bus + SQS Queue

# EventBridge Custom Event Bus + Rule
resource "aws_cloudwatch_event_bus" "main" {
  name = "${var.project}-event-bus"
  tags = { Project = var.project, Env = var.environment }
}

resource "aws_cloudwatch_event_rule" "order_created" {
  name  = "${var.project}-order-created"
  event_bus_name = aws_cloudwatch_event_bus.main.name
  event_pattern  = jsonencode({
 source= ["custom.order-service"]
 detail-type = ["order.created"]
 detail= { status = ["PENDING"] }
  })
}

resource "aws_cloudwatch_event_target" "order_sqs" {
  rule  = aws_cloudwatch_event_rule.order_created.name
  event_bus_name = aws_cloudwatch_event_bus.main.name
  target_id= "order-sqs-target"
  arn= aws_sqs_queue.order_queue.arn
}

# SQS Queue with DLQ (Long Polling + KMS + Redrive)
resource "aws_sqs_queue" "order_dlq" {
  name = "${var.project}-order-dlq"
  message_retention_seconds = 1209600
  kms_master_key_id= "alias/aws/sqs"
}

resource "aws_sqs_queue" "order_queue" {
  name  = "${var.project}-order-queue"
  visibility_timeout_seconds = 300
  message_retention_seconds  = 345600
  receive_wait_time_seconds  = 20
  kms_master_key_id = "alias/aws/sqs"
  redrive_policy = jsonencode({
 deadLetterTargetArn = aws_sqs_queue.order_dlq.arn
 maxReceiveCount  = 3
  })
}

統合アーキテクチャパターン 3選

パターン1: EventBridge → SQS → Lambda (Decoupled Processing)
EC2 / ECS / Lambda からカスタムイベントを EventBridge に発行し、Rule で SQS にルーティングして Lambda が非同期処理するパターンです。Publisher は Consumer の実装を知らず、Consumer の追加・変更が Publisher に影響しません。EventBridge の Archive+Replay でイベントを保持し、バグ修正後に再処理することもできます。

Publisher → aws.events.PutEvents → EventBridge Bus
  → Rule (Event Pattern match) → SQS Queue
  → Lambda Event Source Mapping → Lambda 処理 (DLQ 付き)

パターン2: SNS → SQS Fanout (1 Publisher → N Consumer)
1つの SNS Topic に対して複数の SQS キューをサブスクライブし、同じイベントを複数の独立したシステムで並列処理するパターンです。Subscription Filter Policy を設定することで、各 SQS には必要なイベントのみ配信されます。注文イベントを在庫サービス・通知サービス・分析サービスで同時処理する構成が典型例です。

Order Service → SNS Topic (order.created)
  ├─ Subscription (Filter: status=PENDING) → SQS (在庫処理)
  ├─ Subscription (Filter: なし) → SQS (通知処理)
  └─ Subscription (Filter: なし) → SQS (分析処理)

パターン3: Kinesis Data Streams → Lambda (Real-time Stream Processing)
大量のリアルタイムイベント (クリックストリーム / ログ / IoT データ 等) を Kinesis に書き込み、Lambda で並列ストリーム処理するパターンです。Enhanced Fan-out を使うと複数 Consumer が独立したスループット (2 MB/sec/Shard) で読み取れます。Parallelization Factor = 10 設定で 1 Shard に対して 10 Lambda を並列起動し処理速度を向上させます。

Data Sources → Kinesis.PutRecords (Partition Key: userId)
  → Kinesis Data Streams (Shard × N)
  → Lambda ESM (Parallelization Factor=10, Batch Window=5s)
  → Lambda 処理 (Enhanced Fan-out + DLQ)

4本柱 アンチパターン 4選

アンチパターン1: SNS のみで順序保証
SNS Standard は順序保証なしの Fan-out サービスです。複数 Lambda Subscriber に配信されるメッセージの順序は保証されません。「最初に確認メール、次に本確認メール」のような順序依存処理では SNS 単体では実現できません。順序保証が必要なら SQS FIFO または Kinesis を使います。

アンチパターン2: Kinesis でランダムアクセス
Kinesis Data Streams はシーケンシャル読み取り (先頭から順番) を前提とした設計です。「特定の順序 ID のレコードのみ取得」のようなランダムアクセスには対応していません。特定レコードの直接参照が必要な場合は DynamoDB や S3 を使います。

アンチパターン3: EventBridge → Lambda 直結 (DLQ なし)
EventBridge Rule の Target に Lambda を直接設定した場合、Lambda 処理が失敗するとデフォルトで 2 回リトライ後にイベントが消失します。DLQ (Dead Letter Queue) または On-Failure Destination を設定することで失敗イベントを保存し、後から再処理できます。

アンチパターン4: SQS FIFO の過剰利用
SQS FIFO は 3,000 TPS 上限があります。高スループット要件 (> 3,000 TPS) に FIFO を使うと即座にスロットリングが発生します。本当に順序保証が必要な処理を見極め、不要なら Standard キュー + べき等処理で代替します。

Terraform 例: EventBridge IAM Role + Lambda Event Source Mapping

本番運用では EventBridge が SQS に書き込むための IAM Role と、SQS を Lambda がポーリングするための Event Source Mapping が必要です。

# EventBridge → SQS への送信 IAM Role
resource "aws_iam_role" "eventbridge_to_sqs" {
  name = "${var.project}-eventbridge-sqs-role"
  assume_role_policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect = "Allow"
Principal = { Service = "events.amazonaws.com" }
Action = "sts:AssumeRole"
 }]
  })
}

resource "aws_iam_role_policy" "eventbridge_to_sqs" {
  name = "SendMessagePolicy"
  role = aws_iam_role.eventbridge_to_sqs.id
  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect= "Allow"
Action= ["sqs:SendMessage"]
Resource = [aws_sqs_queue.order_queue.arn]
 }]
  })
}

# Lambda Event Source Mapping (SQS → Lambda)
resource "aws_lambda_event_source_mapping" "sqs_to_lambda" {
  event_source_arn = aws_sqs_queue.order_queue.arn
  function_name = aws_lambda_function.order_processor.arn
  batch_size = 10
  maximum_batching_window_in_seconds = 5
  function_response_types= ["ReportBatchItemFailures"]

  scaling_config {
 maximum_concurrency = 50
  }
}

3. EventBridge本番運用 ★山場1

3.1 EventBridge 5機能の全体像と使い分け

Amazon EventBridge は「イベントバス + ルーティング」を中心に7つの機能で構成される。

3.2 Event Bus 設計パターン (デフォルト / カスタム / パートナーバス)

本番環境では Default Event Bus に独自アプリのイベントを乗せず、Custom Bus を環境単位・ドメイン単位で分離するパターンが推奨される。

Event Bus 選定基準表

本番推奨構成: prod-domain-order / prod-domain-payment のようにドメイン単位でカスタムバスを分割し、Rule の肥大化と意図しないイベントルーティングを防ぐ。

3.3 EventBridge Rule 実装例 (EventPattern + Terraform)

EventPattern 設計の3段階絞り込み指針

Event Pattern の精度が本番の安定性を決定する。source → detail-type → detail の順に具体化していく。

{
  "source": ["com.myapp.order"],
  "detail-type": ["OrderCreated"],
  "detail": {
 "status": ["CONFIRMED"],
 "amount": [{ "numeric": [">=", 10000] }]
  }
}

• source: 独自アプリは逆 DNS 形式 (com.myapp.order) で名前空間を確保する
• detail-type: イベント種別を文字列配列で指定 (大文字 / 小文字は厳密に区別される)
• detail: フィールド条件まで絞り込む (数値比較 / prefix マッチ / anything-but も利用可)

Terraform: aws_cloudwatch_event_bus + aws_cloudwatch_event_rule + aws_cloudwatch_event_target

resource "aws_cloudwatch_event_bus" "order" {
  name = "prod-domain-order"
}

resource "aws_cloudwatch_event_rule" "order_confirmed" {
  name  = "prod-order-confirmed"
  event_bus_name = aws_cloudwatch_event_bus.order.name

  event_pattern = jsonencode({
 source= ["com.myapp.order"]
 detail-type = ["OrderCreated"]
 detail = {
status = ["CONFIRMED"]
 }
  })

  description = "CONFIRMED 注文を処理 Lambda へ転送"
}

resource "aws_cloudwatch_event_target" "order_lambda" {
  rule  = aws_cloudwatch_event_rule.order_confirmed.name
  event_bus_name = aws_cloudwatch_event_bus.order.name
  target_id= "order-processor"
  arn= aws_lambda_function.order_processor.arn

  retry_policy {
 maximum_retry_attempts = 3
 maximum_event_age_in_seconds = 300
  }

  dead_letter_config {
 arn = aws_sqs_queue.order_dlq.arn
  }
}

本番 Rule 設定の必須項目:
– retry_policy: デフォルトは24時間・185回なので必ず絞り込む (最大リトライ3回・保持300秒が目安)
– dead_letter_config: 失敗イベントを SQS DLQ に保存し後から解析できるようにする
– Rule に最小権限 IAM Role を付与 (ターゲット ARN への送信権限のみ)

3.4 EventBridge Pipes 本番パターン (SQS → Filter → Lambda → Target)

EventBridge Pipes は Source → Filter → Enrichment (オプション) → Target を宣言的に接続するフルマネージドパイプラインで、追加の Lambda コードなしに ETL パターンを実現できる。

本番ユースケース: SQS → Filter → Lambda Enrichment → SNS Target

1. Source: SQS Standard Queue からバッチ取得 (Batch Size / Batch Window で集約制御)
2. Filter: body.eventType = "HIGH_PRIORITY" に一致するメッセージのみ通過させてコスト削減
3. Enrichment: Lambda で DynamoDB から顧客情報を取得してイベントにメタデータを付与
4. Target: SNS Topic に Enrichment 済みイベントを Fan-out して複数 Subscriber に配信

sequenceDiagram
 participant SQS as SQS Source<br>(Queue)
 participant Filter as Pipes Filter<br>(EventPattern)
 participant Lambda as Enrichment<br>(Lambda)
 participant SNS as Target<br>(SNS Topic)
 participant DLQ as DLQ<br>(SQS)

 SQS->>Filter: Batch メッセージ取得 (BatchSize=10)
 Filter-->>SQS: マッチしないメッセージはスキップ (コスト削減)
 Filter->>Lambda: フィルタ通過メッセージを Enrichment へ送信
 Lambda->>Lambda: DynamoDB 参照でメタデータ付与
 Lambda->>SNS: Enrichment 済みイベントを送信
 SNS-->>Lambda: 配信成功確認
 Lambda-->>DLQ: 処理失敗時は DLQ へ退避

Terraform: aws_pipes_pipe

resource "aws_pipes_pipe" "order_enrichment" {
  name  = "prod-order-enrichment-pipe"
  role_arn = aws_iam_role.pipes_role.arn

  source = aws_sqs_queue.order_input.arn
  source_parameters {
 sqs_queue_parameters {
batch_size = 10
maximum_batching_window_in_seconds = 5
 }
 filter_criteria {
filter {
  pattern = jsonencode({
 body = {
eventType = ["HIGH_PRIORITY"]
 }
  })
}
 }
  }

  enrichment = aws_lambda_function.order_enricher.arn
  enrichment_parameters {
 input_template = <<-EOT
{
  "orderId": <$.body.orderId>,
  "amount":  <$.body.amount>
}
 EOT
  }

  target = aws_sns_topic.order_notifications.arn
  target_parameters {
 sns_topic_parameters {
input_template = "<$.enriched>"
 }
  }
}

3.5 EventBridge Scheduler 実装例 (cron / rate / one-time)

EventBridge Scheduler は CloudWatch Events スケジュールルールの後継で、「柔軟スケジュール (Flexible Time Window)」と「one-time スケジュール」が追加された。

resource "aws_scheduler_schedule" "nightly_batch" {
  name = "prod-nightly-order-batch"
  group_name = "prod-batch"

  flexible_time_window {
 mode = "FLEXIBLE"
 maximum_window_in_minutes = 15
  }

  schedule_expression = "cron(0 2 * * ? *)"
  schedule_expression_timezone = "Asia/Tokyo"

  target {
 arn= aws_lambda_function.batch_processor.arn
 role_arn = aws_iam_role.scheduler_role.arn

 input = jsonencode({
action = "nightly_aggregation"
 })

 retry_policy {
maximum_retry_attempts = 2
maximum_event_age_in_seconds = 600
 }

 dead_letter_config {
arn = aws_sqs_queue.scheduler_dlq.arn
 }
  }
}

Flexible Time Window のメリット: 固定時刻スケジュールでは大量テナントが同時起動してスパイクが発生する。FLEXIBLE + 15分 を設定すると 2:00〜2:15 の範囲でランダムに分散起動し Lambda 同時実行数を平準化できる。

3.6 Schema Registry 活用 (スキーマ検出 + コード生成)

EventBridge Schema Registry はバスに流れるイベントのスキーマを自動検出し、TypeScript / Python / Java / Go 向けのバインディングコードを自動生成する。

活用フロー

1. Schema Discovery 有効化: バスに対してディスカバリーを有効化すると自動でスキーマを解析・保存する
2. Event Schema Explorer: コンソールまたは CLI でスキーマ確認 / バージョン管理
3. コードバインディング生成: AWS Toolkit (VS Code / IntelliJ) または CLI でコード生成
4. CI 組み込み: PR マージ前にスキーマバージョンチェックを走らせ破壊的変更をブロック

# TypeScript バインディング生成 (AWS CLI)
aws schemas get-code-binding-source \
  --registry-name aws.events \
  --schema-name aws.ec2@EC2InstanceStateChangeNotification \
  --language TypeScript \
  --schema-version "1" \
  --query CodeBindings.Content \
  --output text > ec2-state-change.ts

CI での活用例: スキーマバージョンを schemas.lock ファイルで管理し、detail フィールドの削除・型変更が検知されたらビルドを失敗させる。イベント仕様の後方互換性をコードレビュー前に機械的に保証できる。

3.7 Cross-Account Event Bus 設計 (IAM Resource Policy + Organizations 連携)

マルチアカウント構成では、送信元アカウント (Account A) から受信先アカウント (Account B) のカスタムバスにイベントを転送する。IAM Resource-based Policy の設定が鍵となる。

設定手順

1. 受信側 Account B: カスタムバスに Resource Policy を設定し送信元を特定アカウントに限定
2. 送信側 Account A: Rule の IAM Role に events:PutEvents 権限を付与 (送信先バス ARN のみ)
3. 送信側 Account A: Rule の Target に Account B のバス ARN を指定

受信側 Resource Policy (Account B に設定)

{
  "Version": "2012-10-17",
  "Statement": [
 {
"Sid": "AllowAccountAputEvents",
"Effect": "Allow",
"Principal": {
  "AWS": "arn:aws:iam::ACCOUNT_A_ID:root"
},
"Action": "events:PutEvents",
"Resource": "arn:aws:events:ap-northeast-1:ACCOUNT_B_ID:event-bus/prod-central-bus"
 }
  ]
}

送信側 IAM Role インラインポリシー (Account A で作成)

{
  "Version": "2012-10-17",
  "Statement": [
 {
"Effect": "Allow",
"Action": "events:PutEvents",
"Resource": "arn:aws:events:ap-northeast-1:ACCOUNT_B_ID:event-bus/prod-central-bus"
 }
  ]
}

Terraform: Cross-Account Event Bus Target

resource "aws_cloudwatch_event_target" "central_bus" {
  rule  = aws_cloudwatch_event_rule.order_completed.name
  event_bus_name = aws_cloudwatch_event_bus.order.name
  target_id= "central-event-bus"
  arn= "arn:aws:events:ap-northeast-1:${var.central_account_id}:event-bus/prod-central-bus"
  role_arn = aws_iam_role.cross_account_events.arn
}

Organizations 連携: AWS Organizations を利用する場合、Resource Policy の Principal に {Service: events.amazonaws.com} と Condition: {StringEquals: {aws:PrincipalOrgID: o-xxxxx}} を組み合わせることで、特定 OU 配下の全アカウントからのイベント受信を1エントリで許可できる。

4. SQS本番運用 ★山場2

4-1. Standard vs FIFO 選定基準

SQS の2モードは「順序保証が必要か」「スループットが最優先か」の2軸で決まる。

選定の原則: 順序保証が不要ならば Standard を選ぶ。FIFO はスループット上限 (300 TPS) があるため、高スループット要件では Standard + べき等処理が正解。

4-2. DLQ 設計パターン (MaxReceiveCount + リドライブポリシー)

DLQ が未設定の場合、maxReceiveCount 回の受信失敗後にメッセージは永久消失する。

MaxReceiveCount 推奨値: 3〜5 回。

リドライブポリシー (Redrive Policy) 戦略

# DLQ 再処理 Lambda — try/except パターン
import boto3
import json
import os

sqs = boto3.client('sqs')

def handler(event, context):
 main_queue_url = os.environ['MAIN_QUEUE_URL']
 batch_item_failures = []

 for record in event['Records']:
  try:
body = json.loads(record['body'])
error_type = body.get('error_type', 'unknown')
receive_count = int(record['attributes']['ApproximateReceiveCount'])

if error_type == 'transient' and receive_count <= 3:
 sqs.send_message(
  QueueUrl=main_queue_url,
  MessageBody=json.dumps(body)
 )
else:
 save_to_dead_archive(body, record)
  except Exception as e:
print(f"DLQ reprocessing error: {e}")
batch_item_failures.append({"itemIdentifier": record['messageId']})

 return {"batchItemFailures": batch_item_failures}

4-3. Long Polling 設定 (WaitTimeSeconds=20)

Short Polling はメッセージ不在でも即時応答するため、空レスポンスが大量発生しコストが増大する。ReceiveMessageWaitTimeSeconds=20 を設定することで最大 20 秒間メッセージ到着を待機する。

• コスト削減効果: メッセージが疎な環境で API 呼び出しを最大 90% 削減
• Lambda ESM ではデフォルトで Long Polling 適用 (明示設定不要)
• SQS Queue レベルで設定すると全コンシューマーに一括適用できる

# Long Polling 設定確認コマンド
aws sqs get-queue-attributes \
  --queue-url https://sqs.ap-northeast-1.amazonaws.com/123456789012/order-queue \
  --attribute-names ReceiveMessageWaitTimeSeconds

4-4. Visibility Timeout 設計 (Lambda タイムアウト × 6 の根拠)

Visibility Timeout はメッセージ受信後に他のコンシューマーから見えなくなる時間。この時間内に処理が完了しない場合、メッセージが再度可視になり 重複処理 が発生する。

AWS 推奨設計式: Visibility Timeout = Lambda タイムアウト × 6

Lambda タイムアウト  30秒  → Visibility Timeout3分 (180秒)
Lambda タイムアウト5分  → Visibility Timeout  30分
Lambda タイムアウト  15分  → Visibility Timeout  90分 (最大値 12時間)

根拠: Lambda の最大再試行回数 (3回) × タイムアウト値 + 処理バッファを合算すると係数 6 が最小安全マージン。処理が長時間化する場合は ChangeMessageVisibility API で処理中に Timeout を延長することも可能。

4-5. Lambda Batch Failure Pattern (BatchItemFailures)

バッチ処理で一部メッセージのみ失敗した場合、全バッチ再処理では成功済みメッセージの重複処理が発生する。BatchItemFailures を使うと 失敗メッセージ ID のみ を SQS に返し部分的な再試行が可能になる。

設定前提: Lambda Event Source Mapping の FunctionResponseTypes: ["ReportBatchItemFailures"] を有効化する (後述 Terraform 例参照)。

# BatchItemFailures による部分失敗ハンドリング
import json


def handler(event, context):
 batch_item_failures = []

 for record in event['Records']:
  try:
body = json.loads(record['body'])
process_message(body)
  except Exception as e:
print(f"Failed: {record['messageId']} reason={e}")
batch_item_failures.append(
 {"itemIdentifier": record['messageId']}
)

 return {"batchItemFailures": batch_item_failures}


def process_message(body):
 order_id = body.get('order_id')
 if not order_id:
  raise ValueError("order_id is required")
 # ビジネスロジック実装

BatchItemFailures を返さない場合はバッチ内のいずれかが例外を上げると全メッセージが SQS に戻され全件再処理される。BatchItemFailures を返すことで成功済みメッセージへの重複処理・副作用を防止できる。

4-6. Terraform 設計例 (aws_sqs_queue / DLQ / Lambda ESM)

# SQS メインキュー — Long Polling + Visibility Timeout + DLQ 設定
resource "aws_sqs_queue" "main" {
  name  = "order-processing-queue"
  visibility_timeout_seconds = 180# Lambda タイムアウト (30s) × 6
  message_retention_seconds  = 86400 # 24時間
  receive_wait_time_seconds  = 20 # Long Polling 有効化

  redrive_policy = jsonencode({
 deadLetterTargetArn = aws_sqs_queue.dlq.arn
 maxReceiveCount  = 3
  })

  tags = {
 Environment = "production"
 Service  = "order-processing"
  }
}

# DLQ + CloudWatch 監視アラーム
resource "aws_sqs_queue" "dlq" {
  name = "order-processing-dlq"
  message_retention_seconds = 1209600 # 14日間
}

resource "aws_cloudwatch_metric_alarm" "dlq_depth" {
  alarm_name = "order-dlq-messages-visible"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 1
  metric_name= "ApproximateNumberOfMessagesVisible"
  namespace  = "AWS/SQS"
  period  = 60
  statistic  = "Maximum"
  threshold  = 0

  dimensions = {
 QueueName = aws_sqs_queue.dlq.name
  }

  alarm_actions = [aws_sns_topic.alerts.arn]
}

# SQS キューポリシー — SNS からの SendMessage を許可
resource "aws_sqs_queue_policy" "main" {
  queue_url = aws_sqs_queue.main.id

  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [
{
  Effect = "Allow"
  Principal = { Service = "sns.amazonaws.com" }
  Action = "sqs:SendMessage"
  Resource  = aws_sqs_queue.main.arn
  Condition = {
 ArnEquals = {
"aws:SourceArn" = aws_sns_topic.orders.arn
 }
  }
}
 ]
  })
}

# Lambda Event Source Mapping — BatchItemFailures + Maximum Concurrency
resource "aws_lambda_event_source_mapping" "sqs_trigger" {
  event_source_arn = aws_sqs_queue.main.arn
  function_name = aws_lambda_function.order_processor.arn
  batch_size = 10
  maximum_batching_window_in_seconds = 5
  function_response_types= ["ReportBatchItemFailures"]

  scaling_config {
 maximum_concurrency = 50
  }
}

5. SNS本番運用 ★山場3

5.1 SNS Standard vs FIFO Topic 選定

SNS は Standard Topic と FIFO Topic の2モードを提供する。FIFO Topic は SQS FIFO キューのみをサブスクライバーとして受け入れる制限があり、Lambda や HTTP エンドポイントへの直接配信は不可となる点を設計前に確認する。

FIFO Topic 採用の判断基準

FIFO Topic が必要なのは「同一エンティティへの状態更新を順序保証して複数システムに配信する」ケースに限定される。注文ステータスが PENDING → CONFIRMED → SHIPPED の順で変化する場合、SQS FIFO をサブスクライバーとした FIFO Topic で配信順序を保証できる。ただし全サブスクライバーが SQS FIFO でなければならない制約から Fan-out の柔軟性が大きく制限される。スループット上限 (300 msg/sec) も Standard Topic 比で圧倒的に低いため、高流量の通知配信には Standard Topic + べき等処理の組み合わせを推奨する。

5.2 Subscription Filter Policy — 4種のフィルタ式

Subscription Filter Policy は SNS Topic の各 Subscription に設定するフィルタルールで、Subscriber が受け取るメッセージを絞り込む。未設定の Subscription は全メッセージを受信する ため、本番では全 Subscription に必ず設定する。

Filter Policy Scope の選択

• MessageAttribute (推奨): メッセージ送信時に付与するメタデータ属性で絞り込む。Payload 解析不要で低レイテンシ・低コスト。
• MessageBody (Payload-based): JSON メッセージ本文のフィールドで絞り込む。属性を付与できない外部システムからのメッセージに対応できる。

Filter Policy 5種の式 + 複合例

{
  "status": ["CONFIRMED", "SHIPPED"],
  "amount": [{"numeric": [">=", 10000]}],
  "source": [{"prefix": "com.myapp.order"}],
  "priority": [{"exists": true}]
}

複合条件の評価ルール: 条件間は AND 評価、各配列内の要素は OR 評価となる。上記は (status=CONFIRMED OR SHIPPED) かつ (amount>=10000) かつ (sourceがcom.myapp.orderで始まる) かつ (priority属性が存在する) の全条件を満たすメッセージのみを配信する。

Filter Policy の適用タイミング: Filter Policy の変更は非同期で反映される。CloudWatch Metrics の NumberOfNotificationsFilteredOut を監視し、変更後の配信数が想定通りか確認する。

5.3 SNS → SQS Fanout アーキテクチャ + mermaid02

SNS → SQS Fanout は「1つのイベントを複数のキューで独立処理する」疎結合パターンの標準実装だ。SNS Topic に複数の SQS Queue をサブスクライブし、各 Queue に固有の Filter Policy を設定することで、Subscriber ごとに必要なイベントのみを配信する。

sequenceDiagram
  participant P as Order Service
  participant SNS as SNS Topic<br/>order-events
  participant SQS1 as SQS Queue<br/>inventory-queue
  participant SQS2 as SQS Queue<br/>notification-queue
  participant SQS3 as SQS Queue<br/>analytics-queue (us-east-1)
  participant L1 as Lambda<br/>inventory-processor
  participant L2 as Lambda<br/>notification-sender
  participant L3 as Lambda<br/>analytics-writer

  P->>SNS: Publish(order.created, status=CONFIRMED, amount=15000)
  SNS->>SQS1: Deliver (Filter: status=CONFIRMED, amount>=10000)
  SNS->>SQS2: Deliver (Filter: なし / 全配信)
  SNS->>SQS3: Deliver (Cross-Region: us-east-1 / Filter: なし)
  SQS1->>L1: ESM Trigger → 在庫引当
  SQS2->>L2: ESM Trigger → メール/Push通知
  SQS3->>L3: ESM Trigger → 分析 DWH 書込
  Note over SNS,SQS3: Publisher は Consumer を知らない<br/>Subscriber 追加は Publisher 変更不要

Fanout 設計の3原則

1. べき等処理の徹底: SNS は at-least-once 配信のため、同一メッセージが複数回配信される可能性がある。Lambda 側で MessageId をキーにした DynamoDB 重複チェックを実装する。
2. SQS DLQ の必須設定: 各 SQS Queue に DLQ を設定し、処理失敗メッセージを隔離する。MaxReceiveCount=3〜5 でリトライ後に DLQ 転送する設定が本番標準。
3. Subscription ごとの Filter Policy: 不要なメッセージの配信を防ぐことで、下流 Lambda の無駄な実行を削減してコストを抑制する。

5.4 Cross-Region Fanout — 構成と注意事項

SNS Topic からクロスリージョンの SQS Queue へのサブスクリプションが可能だ。プライマリリージョン (ap-northeast-1) の SNS Topic から DR リージョン (us-east-1) の SQS Queue へ直接配信し、リージョン障害時のフェイルオーバー処理を実現する。

設定要件

1. SQS Queue Resource Policy: 別リージョンの SNS からメッセージを受け入れるために、宛先 SQS Queue の Resource Policy で sns.amazonaws.com からの sqs:SendMessage を許可する。
2. KMS 暗号化: SNS → SQS 間の通信は HTTPS で自動暗号化されるが、SQS 側 KMS キーは宛先リージョンのキーを使用するため別途設定が必要。
3. サブスクリプション確認: SQS プロトコルの SNS Subscription は自動確認されるため、HTTP エンドポイントと異なり手動確認は不要。

Cross-Region 推奨ユースケース

5.5 Lambda 連携本番設定 — DLQ + Retry Policy

SNS から Lambda を直接呼び出す構成では Retry Policy と Throttle Policy が本番品質を決定する。

SNS Lambda Subscription Retry Policy

{
  "deliveryPolicy": {
 "healthyRetryPolicy": {
"numRetries": 5,
"minDelayTarget": 20,
"maxDelayTarget": 300,
"numNoDelayRetries": 0,
"numMinDelayRetries": 1,
"numMaxDelayRetries": 4,
"backoffFunction": "exponential"
 },
 "throttlePolicy": {
"maxReceivesPerSecond": 100
 }
  }
}

推奨構成: SNS → SQS → Lambda ESM

SNS から Lambda を直接呼ぶより、SQS を経由する構成が本番品質で優れる理由は3点ある。

1. SQS の Visibility Timeout による分散処理制御が可能
2. Lambda ESM の ReportBatchItemFailures で部分失敗のみリトライ
3. SQS DLQ で失敗メッセージを永続保存して後から再処理できる

直接 SNS → Lambda 構成では最大リトライ後に失敗したイベントはデフォルトで消失するため、Lambda 側に On-Failure Destination (DLQ) を必ず設定する。

5.6 Terraform 例

# SNS Standard Topic (KMS 暗号化有効)
resource "aws_sns_topic" "order_events" {
  name  = "${var.project}-order-events"
  kms_master_key_id = "alias/aws/sns"
  tags = {
 Project  = var.project
 Environment = var.environment
  }
}

# SNS Subscription → SQS (Filter Policy 付き在庫処理向け)
resource "aws_sns_topic_subscription" "order_to_inventory" {
  topic_arn  = aws_sns_topic.order_events.arn
  protocol= "sqs"
  endpoint= aws_sqs_queue.inventory_queue.arn
  filter_policy_scope = "MessageAttributes"
  filter_policy = jsonencode({
 status = ["CONFIRMED", "SHIPPED"]
 amount = [{ "numeric" = [">=", 1000] }]
  })
}

resource "aws_sns_topic_subscription" "order_to_analytics" {
  topic_arn = aws_sns_topic.order_events.arn
  protocol  = "sqs"
  endpoint  = aws_sqs_queue.analytics_queue.arn
}

# SNS Topic Policy (同一アカウント Publish + SQS Subscribe 許可)
resource "aws_sns_topic_policy" "order_events" {
  arn = aws_sns_topic.order_events.arn
  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [
{
  Sid = "AllowAccountPublish"
  Effect = "Allow"
  Principal = { AWS = "arn:aws:iam::${var.account_id}:root" }
  Action = ["sns:Publish"]
  Resource  = aws_sns_topic.order_events.arn
},
{
  Sid = "AllowSQSSubscribe"
  Effect = "Allow"
  Principal = { Service = "sqs.amazonaws.com" }
  Action = ["sns:Subscribe"]
  Resource  = aws_sns_topic.order_events.arn
  Condition = {
 ArnLike = {
"aws:SourceArn" = "arn:aws:sqs:*:${var.account_id}:*"
 }
  }
}
 ]
  })
}

6. Kinesis Data Streams本番運用 ★山場4

6.1 Shard 設計 — スループット計算式と設計手順

Kinesis Data Streams のスループット単位は Shard だ。1 Shard あたりの制限を理解し、設計時に十分な Shard 数を確保しないと ProvisionedThroughputExceededException が頻発する。

Shard スループット制限

必要 Shard 数の計算式

# Write Shard 数 (大きい方を採用)
write_shard_by_record = ceil(records_per_sec / 1000)
write_shard_by_size= ceil(mb_per_sec / 1)
write_shard_count  = max(write_shard_by_record, write_shard_by_size)

# Read Shard 数 (Standard Consumer の場合)
read_shard_count = ceil(consumer_count * read_mb_per_sec / 2)

# 最終 Shard 数
required_shards = max(write_shard_count, read_shard_count)

設計例: 書き込み 3,000 records/sec、平均レコードサイズ 2 KB、Consumer 5 台の場合

write_shard_by_record = ceil(3000 / 1000) = 3
write_shard_by_size= ceil(6 MB/sec / 1) = 6
read_shard_count= ceil(5 * 2 MB / 2) = 5
required_shards = max(6, 5) = 6

スパイクを考慮して 安全係数 1.5〜2 倍 を掛けるのが本番運用の標準だ。上記の場合は 9〜12 Shard での構成を推奨する。

6.2 On-Demand vs Provisioned モード選定

Kinesis Data Streams には On-Demand モード と Provisioned モード の2つの容量設定モードがある。

選定基準: 月次コスト試算で On-Demand が Provisioned より 20% 以上高い場合は Provisioned に移行するタイミングとみなす。移行は AWS Console または Terraform の stream_mode_details ブロックで実施し、ダウンタイムなく切り替えられる。

6.3 Enhanced Fan-out (EFO) — 移行タイミング判断

Standard Consumer では全 Consumer が 1 Shard の 2 MB/sec を共有するため、Consumer 数が増えると個別の読み取りスループットが低下する。Enhanced Fan-out (EFO) では Consumer ごとに独立した 2 MB/sec のプッシュ型配信を受け取れる。

EFO 移行を検討する判断基準

EFO vs Standard Consumer 比較

6.4 Lambda Trigger 設計 — bisect-on-error / Parallelization Factor / Tumbling Window

Lambda Event Source Mapping (ESM) で Kinesis を処理する際の本番品質設定を解説する。

本番推奨 ESM 設定値

bisect_batch_on_function_error の効果

Lambda がバッチ処理中にエラーを出した場合、バッチを2分割して再試行する。1 レコードが原因の場合、最大 log2(batch_size) 回の分割で障害レコードのみを特定して DLQ に転送できる。デフォルトの false では1レコードのエラーがバッチ全体を無限リトライさせ、IteratorAge が際限なく増加する。

Parallelization Factor の効果

1 Shard に対して最大 10 の Lambda 並列実行を設定できる。Shard 数 × Parallelization Factor が同時実行 Lambda 数の最大値となる。Lambda の同時実行上限 (デフォルト 1,000) と合わせて設計する。

Tumbling Window の活用

tumbling_window_in_seconds = 60 を設定すると、60 秒間のレコードを集約してから Lambda を1回呼び出す。リアルタイム集計・カウント集計のユースケースで有効だ。Lambda 関数内で state を shardId-00000000001 などのキーで管理し、ウィンドウ内の集計値を積算する。

6.5 Hot Partition 検知 — CloudWatch メトリクス監視

Hot Partition は特定の Shard に書き込みが集中し、ProvisionedThroughputExceededException が頻発する状態だ。Partition Key に「日付」「サービス名」「固定文字列」などカーディナリティの低い値を使うと発生する。

検知に使用する CloudWatch メトリクス

CloudWatch アラーム設定 (Terraform)

resource "aws_cloudwatch_metric_alarm" "kinesis_write_throttle" {
  alarm_name = "${var.project}-kinesis-write-throttle"
  namespace  = "AWS/Kinesis"
  metric_name= "WriteProvisionedThroughputExceeded"
  dimensions = { StreamName = aws_kinesis_stream.main.name }
  statistic  = "Sum"
  period  = 60
  evaluation_periods  = 3
  threshold  = 1
  comparison_operator = "GreaterThanOrEqualToThreshold"
  alarm_actions = [aws_sns_topic.alerts.arn]
}

resource "aws_cloudwatch_metric_alarm" "kinesis_iterator_age" {
  alarm_name = "${var.project}-kinesis-iterator-age"
  namespace  = "AWS/Kinesis"
  metric_name= "GetRecords.IteratorAgeMilliseconds"
  dimensions = { StreamName = aws_kinesis_stream.main.name }
  statistic  = "Maximum"
  period  = 60
  evaluation_periods  = 5
  threshold  = 60000
  comparison_operator = "GreaterThanThreshold"
  alarm_actions = [aws_sns_topic.alerts.arn]
}

6.6 Hot Partition 回避 — ランダムパーティションキー実装

Hot Partition を根本的に回避するには Partition Key のカーディナリティを高める必要がある。

アンチパターン: 低カーディナリティの Partition Key

import boto3
import json

def bad_example(order_data):
 client = boto3.client('kinesis', region_name='ap-northeast-1')
 response = client.put_record(
  StreamName='order-stream',
  Data=json.dumps(order_data).encode('utf-8'),
  PartitionKey='order-service'
 )
 return response

上記の PartitionKey='order-service' (固定値) は全レコードが同一 Shard に集中する典型的なアンチパターンだ。

推奨: ランダム Suffix によるキー分散 (try/except パターン)

import boto3
import json
import random

def put_to_kinesis(stream_name, order_id, payload, region='ap-northeast-1'):
 client = boto3.client('kinesis', region_name=region)
 suffix = random.randint(0, 99)
 partition_key = f"{order_id}-{suffix}"
 try:
  response = client.put_record(
StreamName=stream_name,
Data=json.dumps(payload).encode('utf-8'),
PartitionKey=partition_key
  )
  return response
 except client.exceptions.ProvisionedThroughputExceededException as e:
  raise RuntimeError(f"Kinesis throttled on put_record: {e}") from e

def batch_put_to_kinesis(stream_name, records, region='ap-northeast-1'):
 client = boto3.client('kinesis', region_name=region)
 kinesis_records = []
 for record in records:
  suffix = random.randint(0, 999)
  partition_key = f"{record['order_id']}-{suffix}"
  kinesis_records.append({
'Data': json.dumps(record).encode('utf-8'),
'PartitionKey': partition_key
  })
 try:
  response = client.put_records(
StreamName=stream_name,
Records=kinesis_records
  )
  failed_count = response.get('FailedRecordCount', 0)
  if failed_count > 0:
raise RuntimeError(f"{failed_count} records failed in put_records")
  return response
 except Exception as e:
  raise RuntimeError(f"Kinesis put_records failed: {e}") from e

Partition Key 設計の4原則

1. 高カーディナリティ: ユーザー ID・注文 ID など固有 ID を基本とする
2. ランダム Suffix の範囲: Shard 数の 10 倍以上の範囲でランダムを生成する (Shard=10 → suffix 0〜99)
3. 順序保証とのトレードオフ: 同一ユーザーの処理順序を守るなら userId のみを Partition Key にする (Shard 内順序保証) が Hot Partition リスクと直結する
4. put_records の部分失敗処理: put_records はバッチ内の部分失敗を FailedRecordCount で返す。失敗レコードのみ再試行するロジックを必ず実装する

6.7 Terraform 例

# Kinesis Data Stream (Provisioned / KMS暗号化 / 保持7日)
resource "aws_kinesis_stream" "main" {
  name = "${var.project}-order-stream"
  shard_count= 6
  retention_period = 168

  stream_mode_details {
 stream_mode = "PROVISIONED"
  }

  encryption_type = "KMS"
  kms_key_id= "alias/aws/kinesis"

  tags = {
 Project  = var.project
 Environment = var.environment
  }
}

# Lambda Event Source Mapping (Kinesis → Lambda / bisect有効)
resource "aws_lambda_event_source_mapping" "kinesis_to_lambda" {
  event_source_arn = aws_kinesis_stream.main.arn
  function_name = aws_lambda_function.stream_processor.arn
  starting_position= "TRIM_HORIZON"
  batch_size = 200
  maximum_batching_window_in_seconds = 10
  parallelization_factor = 5
  bisect_batch_on_function_error  = true
  tumbling_window_in_seconds= 60

  destination_config {
 on_failure {
destination_arn = aws_sqs_queue.kinesis_dlq.arn
 }
  }
}

# Kinesis Enhanced Fan-out Consumer + 書き込み Throttle アラーム
resource "aws_kinesis_stream_consumer" "order_processor" {
  name = "${var.project}-order-processor"
  stream_arn = aws_kinesis_stream.main.arn
}

resource "aws_cloudwatch_metric_alarm" "kinesis_throttle_alert" {
  alarm_name = "${var.project}-kinesis-shard-write-throttle"
  namespace  = "AWS/Kinesis"
  metric_name= "WriteProvisionedThroughputExceeded"
  dimensions = { StreamName = aws_kinesis_stream.main.name }
  statistic  = "Sum"
  period  = 60
  evaluation_periods  = 3
  threshold  = 1
  comparison_operator = "GreaterThanOrEqualToThreshold"
  alarm_actions = [aws_sns_topic.alerts.arn]
}

7. 詰まりポイント7選 + アンチパターン演習5問

EventBridge / SQS / SNS / Kinesis を本番運用で初めて使うエンジニアがはまりやすい7つの詰まりポイントと、Before → After 形式で学ぶアンチパターン演習5問を解説する。各詰まりは「なぜ詰まるか」→「どう解くか」の2段構成で整理した。

詰まり1: EventBridge DetailType / source 不一致でルールにマッチしない

なぜ詰まるか

EventBridge Rule の Event Pattern は source と detail-type フィールドが 完全一致 で評価される。独自アプリから PutEvents する際に source を myapp (小文字) と設定したのに、Rule Pattern 側に Myapp (大文字 M) と書いてしまうだけでイベントが一切マッチしない。detail-type のスペルや大文字小文字のブレも同様にマッチ失敗を引き起こす。CloudWatch Metrics の MatchedEvents が 0 のまま、InvocationAttempts も 0 という状態になり「Lambda が呼ばれていない」とだけ見えるため原因特定が遅れる。

どう解くか

1. EventBridge Console の「Test event pattern」機能でサンプルイベント JSON を貼り付けてマッチを確認する
2. source は逆 DNS 形式 (com.myapp.order) を採用し、大文字混入を防ぐ命名規則を統一する
3. CloudWatch Metrics の MatchedEvents を Rule 単位でダッシュボード監視し、0 の場合はアラートを出す

{
  "source": ["com.myapp.order"],
  "detail-type": ["OrderCreated"],
  "detail": {
 "status": ["CONFIRMED"]
  }
}

詰まり2: SQS DLQ 無限ループ (MaxReceiveCount=1 設定ミス)

なぜ詰まるか

MaxReceiveCount=1 に設定すると、Lambda が1回でもエラーを返した瞬間にメッセージが DLQ に転送される。DLQ に Lambda トリガーを設定している場合、DLQ のメッセージを Lambda が処理してまたエラーになり → 別の DLQ がなければメッセージが消失する。あるいは DLQ の Lambda が無限リトライを繰り返してコストが爆発する。一時的なネットワーク瞬断・下流 API の 503 でも即 DLQ 転送されるため、想定外の DLQ 滞留が発生しても「設定ミス」とは気づかない。

どう解くか

• MaxReceiveCount は 3〜5 に設定し、一時的なエラーで即 DLQ 転送しない
• DLQ には Lambda 自動トリガーを設定しない。CloudWatch Alarm で DLQ 滞留を検知して人手でリドライブ実行する
• DLQ 自体にさらに DLQ を設定する場合は保持期間 14 日・自動トリガーなしを徹底する

詰まり3: SNS Filter Policy の JSON syntax エラー

なぜ詰まるか

SNS Subscription Filter Policy は正しい JSON 形式と SNS 独自のフィルタ条件式を組み合わせる必要があるが、Console でのエラーメッセージが曖昧なため何が間違っているかわかりにくい。よくある誤りは (a) 値を配列でなく文字列として直書きする / (b) anything-but を anything_but と書く / (c) prefix マッチを {"prefix": "HIGH"} ではなく "prefix:HIGH" と書くケース。

どう解くか

• AWS CLI aws sns set-subscription-attributes で設定し、エラーメッセージを CLI で確認する
• Filter Policy は必ず JSON Linter でバリデーション後に適用する
• attribute-based filtering か payload-based filtering かを最初に決め、混在しない
• 変更後は CloudWatch Metrics NumberOfNotificationsFilteredOut で期待通りか検証する

{
  "event_type": ["ORDER_CREATED", "ORDER_UPDATED"],
  "priority": [{ "prefix": "HIGH" }],
  "amount": [{ "numeric": [">=", 10000] }]
}

詰まり4: Kinesis Hot Partition (単調増加キー使用)

なぜ詰まるか

Kinesis Data Streams の Partition Key は hash(key) mod Shard数 で書き込み先 Shard を決定する。date (例: "2026-05-14") や "fixed-key" などの偏ったキーを使うと特定 Shard に書き込みが集中し ProvisionedThroughputExceededException が頻発する。Shard を追加しても Partition Key の偏りが解消されないため、Shard 数増加がコスト増だけで効果がない状態になる。

どう解くか

• Partition Key には十分な散らばりを持つ識別子 (User ID / Order ID / UUID) を使う
• さらに分散を強制したい場合はランダム Suffix を付与する
• CloudWatch Metrics IncomingRecords を Shard 単位で表示し、特定 Shard への偏りがないか定期モニタリングする

import random


def get_partition_key(user_id: str) -> str:
 suffix = random.randint(0, 9)
 return f"{user_id}-{suffix}"

詰まり5: Lambda + SQS バッチ全件リトライ (BatchItemFailures 未実装)

なぜ詰まるか

Lambda が SQS バッチを処理中に1件でも例外が発生し、BatchItemFailures を返さずに例外を伝播させると SQS ESM はバッチ全件を再試行する。成功済みメッセージも再処理されるため、メール二重送信・決済二重実行などの副作用が本番に流出する。Event Source Mapping に FunctionResponseTypes: ["ReportBatchItemFailures"] を設定しても Lambda 側の実装が伴わないと効果がない点が見落とされやすい。

どう解くか

FunctionResponseTypes: ["ReportBatchItemFailures"] を ESM に設定した上で、Lambda が batchItemFailures リストを返す実装を行う。

import json


def handler(event, context):
 batch_item_failures = []

 for record in event['Records']:
  try:
body = json.loads(record['body'])
process_order(body)
  except Exception as e:
print(f"Processing failed: {record['messageId']} error={e}")
batch_item_failures.append(
 {"itemIdentifier": record['messageId']}
)

 return {"batchItemFailures": batch_item_failures}


def process_order(body):
 order_id = body.get('order_id')
 if not order_id:
  raise ValueError("order_id is required")

詰まり6: Event Sourcing の重複処理 (idempotency 未実装)

なぜ詰まるか

SQS Standard は at-least-once 配信であり、ネットワーク障害やタイムアウト時にメッセージが重複して Lambda に配信される。idempotency (冪等性) を実装していない場合、同一メッセージが複数回処理されビジネス上の副作用が発生する。SQS FIFO の重複排除ウィンドウ (5分) は同一 MessageDeduplicationId に限るため、独自アプリの処理ロジック内での冪等性保証とは別物である点に注意する。

どう解くか

• DynamoDB 冪等キー管理: メッセージ ID (messageId) または業務キー (注文 ID 等) を DynamoDB に条件付き書き込み (ConditionExpression: "attribute_not_exists(pk)") し、2回目以降の処理を早期リターンで排除する
• Lambda Powertools Idempotency: @idempotent デコレータで DynamoDB ベースの冪等処理を宣言的に実装できる
• 冪等キーの保持期間は Visibility Timeout よりも長く設定し、有効期間内の重複を確実に排除する

詰まり7: Cross-Account EventBus routing 失敗 (Resource Policy 設定漏れ)

なぜ詰まるか

クロスアカウント EventBridge 転送では、受信側アカウントのカスタムバスに Resource Policy を設定する必要がある。Resource Policy 未設定または ARN のリージョン / Account ID が一致しない場合、送信側 Rule の Target 呼び出しが AccessDenied で失敗する。CloudWatch Metrics FailedInvocations が上昇し DLQ にイベントが蓄積するが、「送信は成功しているのに受信側 Lambda が動かない」という症状に見えるため原因特定に時間がかかる。

どう解くか

1. 受信側バスの Resource Policy に送信元 Account ID を明示的に許可する (events:PutEvents アクション + 完全 ARN)
2. 送信側 Rule の IAM Role インラインポリシーで対象バス ARN に events:PutEvents を付与する (ARN はリージョン + Account ID + バス名を含む完全形式)
3. aws cloudtrail lookup-events --lookup-attributes AttributeKey=EventName,AttributeValue=PutEvents で AccessDenied エラーの詳細を確認する

{
  "Version": "2012-10-17",
  "Statement": [
 {
"Sid": "AllowCrossAccountPutEvents",
"Effect": "Allow",
"Principal": {
  "AWS": "arn:aws:iam::ACCOUNT_A_ID:root"
},
"Action": "events:PutEvents",
"Resource": "arn:aws:events:ap-northeast-1:ACCOUNT_B_ID:event-bus/prod-central-bus"
 }
  ]
}

アンチパターン演習5問 (Before → After)

実際の設定ミスを Before → After 形式で修正する演習。各問を解いてから解答を確認すること。

演習1: EventBridge Rule 不正 Event Pattern

Before (不正 — すべての S3 イベントにマッチしてしまう):

{
  "source": ["aws.s3"]
}

After (正解 — 特定バケット・特定操作のみマッチ):

{
  "source": ["aws.s3"],
  "detail-type": ["Object Created"],
  "detail": {
 "bucket": {
"name": ["prod-upload-bucket"]
 },
 "object": {
"key": [{ "prefix": "orders/" }]
 }
  }
}

解説: source のみの Pattern はそのサービスの全イベントを受信する。detail-type と detail フィールドまで絞り込んで意図しない Target 呼び出しを防ぐ。

演習2: SQS DLQ Lambda トリガー BatchItemFailures 未実装

Before (不正 — 例外を直接 raise して全件リトライ):

import json


def handler(event, context):
 for record in event['Records']:
  body = json.loads(record['body'])
  process(body)


def process(body):
 if body.get('type') == 'bad':
  raise ValueError("invalid type")

After (正解 — BatchItemFailures で失敗件のみリトライ):

import json


def handler(event, context):
 batch_item_failures = []
 for record in event['Records']:
  try:
body = json.loads(record['body'])
process(body)
  except Exception as e:
print(f"Failed {record['messageId']}: {e}")
batch_item_failures.append({"itemIdentifier": record['messageId']})
 return {"batchItemFailures": batch_item_failures}


def process(body):
 if body.get('type') == 'bad':
  raise ValueError("invalid type")

演習3: SNS Filter Policy 不正 syntax

Before (不正 JSON — 値が配列でなく文字列直書き):

{
  "event_type": "ORDER_CREATED",
  "priority": "HIGH"
}

After (正解 JSON — 値は必ず配列):

{
  "event_type": ["ORDER_CREATED"],
  "priority": ["HIGH"],
  "amount": [{ "numeric": [">=", 10000] }]
}

解説: SNS Filter Policy はフィルタ条件値を 配列 で指定しなければならない。文字列を直書きすると設定は受け付けられるが、配信フィルタリングが正常に機能しない。

演習4: Kinesis Shard 計算ミス

Before (不正設計 — Shard 数の計算根拠なし):

トラフィック要件: 5,000 msg/sec / 平均レコードサイズ 800 bytes
設計: 感覚で 1 Shard から運用開始
→ ProvisionedThroughputExceededException が即頻発

After (正解計算式):

Write 制限: 1 Shard = 1,000 records/sec または 1 MB/sec

レコード数ベース: 5,000 ÷ 1,000 = 5 Shard
サイズベース: 5,000 × 800 bytes = 4,000,000 bytes/sec ÷ 1,000,000 = 4 Shard
必要 Shard 数  : max(5, 4) = 5 Shard
安全マージン: 5 × 1.3 = 6.5 → 切り上げ 7 Shard で設計

解説: Write 制限はレコード数ベースとサイズベースの 両方を計算し大きい方 を採用する。安全マージン 1.3 倍を乗せてから切り上げる。

演習5: ASL Distributed Map + EventBridge + SQS 非同期パターン

Before (不正設計 — Step Functions から Lambda を同期直接呼び出し):

{
  "Type": "Map",
  "ItemsPath": "$.orders",
  "Iterator": {
 "StartAt": "ProcessOrder",
 "States": {
"ProcessOrder": {
  "Type": "Task",
  "Resource": "arn:aws:lambda:ap-northeast-1:123456789012:function:order-processor",
  "End": true
}
 }
  }
}

問題: Lambda の同時実行数上限に達すると Map イテレータがスロットリングされ全体が待ち状態になる。

After (正解 — Distributed Map から SQS 経由で非同期処理):

{
  "Type": "Map",
  "ItemProcessor": {
 "ProcessorConfig": {
"Mode": "DISTRIBUTED",
"ExecutionType": "STANDARD"
 },
 "StartAt": "EnqueueOrder",
 "States": {
"EnqueueOrder": {
  "Type": "Task",
  "Resource": "arn:aws:states:::sqs:sendMessage",
  "Parameters": {
 "QueueUrl": "https://sqs.ap-northeast-1.amazonaws.com/123456789012/order-queue",
 "MessageBody.$": "States.JsonToString($)"
  },
  "End": true
}
 }
  },
  "MaxConcurrency": 100,
  "ToleratedFailurePercentage": 10
}

解説: Distributed Map は SQS に sendMessage して Lambda は SQS ESM で非同期処理する。MaxConcurrency で Map の同時実行数を制御し Lambda 側のスロットリングを防ぐ。ToleratedFailurePercentage で一定割合の失敗を許容しながら処理を継続できる。

8. まとめ + Vol3予告 + 落とし穴10選 + 全12軸クロスリンク

Vol2 完遂サマリ — Event-Driven 4本柱統合の設計原則

本 Vol2 では EventBridge (イベントルーティング) / SQS (非同期キュー) / SNS (Pub-Sub) / Kinesis Data Streams (ストリーム) の4本柱を、Vol1 同期3本柱との統合視点で体系化した。Event-Driven Architecture の本番品質は「サービス選定の正確さ」だけでなく「Visibility Timeout / DLQ / Filter Policy / Shard 設計といったパラメータの精密さ」で決まる。疎結合・耐障害・スケーラブルな本番アーキテクチャは、Vol1 + Vol2 の7本柱を組み合わせることで初めて完成する。

EventBridge で得た設計力: Event Pattern の3段階絞り込み (source → detail-type → detail) / Pipes による Source-Filter-Enrichment-Target パイプライン / Scheduler の Flexible Time Window / Cross-Account Resource Policy / Schema Registry による CI 組み込み。いずれも「まず疎結合化してからルーティング精度を上げる」という順序で習得することが重要。

SQS で得た設計力: Visibility Timeout = Lambda タイムアウト × 6 という設計式 / DLQ の必須設定と maxReceiveCount 3〜5 の根拠 / BatchItemFailures による部分失敗ハンドリング / Long Polling によるコスト削減。SQS は「バッファ」として機能し、受付側と処理側のスケールを完全に独立させる。

SNS で得た設計力: Standard vs FIFO の選定 / Subscription Filter Policy による配信絞り込み (attribute-based が推奨) / SNS → SQS Fanout パターンによる1:N 疎結合配信。Subscriber 側のコスト管理は Filter Policy の設計精度に直結する。

Kinesis で得た設計力: Shard 設計式 (Write レコード数・サイズベースの max) / Hot Partition 回避のための Partition Key 分散設計 / Enhanced Fan-out による複数 Consumer 独立処理。ストリーム処理は「Shard 設計の初期精度」が後の拡張コストを左右する。

落とし穴10選

Serverless本番運用シリーズ Vol3 予告

Vol1: Lambda × API Gateway × Step Functions を読む

全12軸 + Serverless Vol1/Vol2 クロスリンク (25記事)

Vol1: Lambda × API Gateway × Step Functions に戻る