サーバーレス設計

EventBridge Pipes・Schedulerを3ヶ月本番運用して見えた、地雷と活用パターン

Lambda + SQS + Step Functionsの複雑な絡み合いをEventBridge Pipes・Schedulerで整理。実装のコツと、本番で痛い目を見たトラブルシューティングをまとめました。

AWSサーバーレスEventBridge設計パターン運用知見

EventBridge Pipes・Schedulerで本番サーバーレスを3ヶ月回した話

先日プロジェクトで、Lambda + SQS + Step Functions の複雑な絡み合いを整理する必要があって、EventBridge Pipes と Scheduler を本格導入してみました。正直、最初は「こんなのわざわざ使うほど？」って懐疑的だったんですけど、3ヶ月回してみたら、複雑さが一気に減ったんですよね。

うちのチームでは、複数のマイクロサービスから発火するイベントを集約して、フィルタリングして、異なるフォーマットで別システムに送る。みたいな処理が増えていて、その度にLambdaを挟んで「中継役」を作ってた。それが累積すると、どれがどのイベントを変換してるのかわからなくなるんですよ。

PipesとSchedulerはそこを一気に解決してくれました。今回は実装した話と、実務で痛い目を見たポイントを共有します。

EventBridge Pipes：イベント変換の中心が生まれた

まず、Pipesについて。これ、要するに「EventBridgeのバージョンアップ版」じゃなくて、専用の「イベント変換・ルーティング」サービスなんですよね。流れとしては Source → Optional Filter → Optional Enrichment → Optional Dead Letter Queue → Target という感じ。

うちで導入した構成を図にするとこんな感じです。

graph TB
    subgraph "Source Services"
        SVC1["Order Service<br/>DynamoDB Streams"]
        SVC2["Payment Service<br/>SNS"]
        SVC3["Inventory Service<br/>SQS"]
    end

    subgraph "EventBridge Pipes"
        PIPE1["Pipe: OrderToPayout<br/>Filter + Enrichment"]
        PIPE2["Pipe: PaymentToNotif<br/>Transform"]
        PIPE3["Pipe: InventoryToAnalytics<br/>Batch Enrichment"]
    end

    subgraph "Targets"
        TGT1["Payout Lambda"]
        TGT2["Notification SQS"]
        TGT3["Analytics S3"]
    end

    subgraph "Monitoring"
        DLQ["Dead Letter Queue<br/>SQS"]
        CW["CloudWatch Logs"]
    end

    SVC1 --> PIPE1
    SVC2 --> PIPE2
    SVC3 --> PIPE3
    PIPE1 --> TGT1
    PIPE2 --> TGT2
    PIPE3 --> TGT3
    PIPE1 -.-> DLQ
    PIPE2 -.-> DLQ
    PIPE3 -.-> DLQ
    PIPE1 -.-> CW
    PIPE2 -.-> CW
    PIPE3 -.-> CW

これが地味に便利なんですよね。具体例でいうと、注文サービスのDynamoDB Streamsで「OrderCreated」イベントが発火する。でも、Payout（支払い）サービスが期待するフォーマットって別で、かつ特定の注文ステータス（例：paid_not_settled）のみ処理したい。そういう場合、従来はLambdaを挟んで変換・フィルタをしてました。

Pipesなら、JSON Path で条件を書いて、EventBridge Input Transformer または Lambda Enrichment で形式を変えるだけ。結果として、Lambda関数を4個減らせたんですよ。

実装した filtering + enrichment の例

ここが実務的なポイントです。フィルター条件をJSON Pathで書きます：

{
  "source": ["order.service"],
  "detail-type": ["Order Created"],
  "detail": {
    "status": ["paid_not_settled"],
    "amount": [{"numeric": [">", 1000]}]
  }
}

これで「支払い済みで未決済、かつ金額1000以上」の注文だけを通す。シンプルですね。

Enrichmentは、Lambda関数を指定するか、API呼び出しもできます。うちの場合、顧客マスター情報を Redis から取得する必要があったので、Lambda enrichment を書きました：

import json
import redis

def lambda_handler(event, context):
    rc = redis.StrictRedis(host=os.environ['REDIS_ENDPOINT'], decode_responses=True)
    
    customer_id = event['detail']['customer_id']
    customer_info = rc.get(f"customer:{customer_id}")
    
    return {
        **event['detail'],
        'customer_tier': json.loads(customer_info)['tier']
    }

PipesがこのLambdaを自動で呼び出して、戻り値をイベントにマージしてくれるんです。冷却時間？ほぼないです。SnapStart対応なので、99%のケースで100ms以内に実行される。

ただ、ここで痛い目を見ました。Enrichment Lambdaのタイムアウトが想像以上に効くんですよね。初期段階で「取得失敗時のリトライ」をEnrichment側で実装してたんですけど、それが延々とリトライされると、Pipeがハング状態になっちゃう。だから、Enrichment Lambdaは「絶対に成功するか、すぐに失敗する」設計にすべき。失敗時の詳細な補正は、Target側のLambdaで処理する方が安全です。

EventBridge Scheduler：定期実行が単純になった

Scheduler は、従来の EventBridge Rules の「定期実行」機能を独立させて強化したもの。cron式と柔軟なタイムゾーン設定、そしてめっちゃ便利な「最大試行回数」「バックオフ戦略」があります。

うちでは、毎日深夜2時にデータウェアハウスの増分ロードを走らせてたんですけど、従来は Step Functions + EventBridge Rules でやってました。今はSchedulerで十分ですよ。

# CDK での定義例
from aws_cdk import aws_scheduler as scheduler
from aws_cdk import aws_iam as iam
from aws_cdk import aws_sqs as sqs

queue = sqs.Queue(self, "DataLoadQueue")

schedule = scheduler.Schedule(
    self, "DailyDataLoadSchedule",
    expression=scheduler.ScheduleExpression.at("cron(0 2 ? * * *)"),  # 毎日2時（UTC）
    timezone="Asia/Tokyo",
    target=scheduler.ScheduleTarget(
        arn=queue.queue_arn,
        role=iam.Role(
            self, "ScheduleRole",
            assumed_by=iam.ServicePrincipal("scheduler.amazonaws.com")
        ),
        retry_policy=scheduler.RetryPolicy(
            max_event_age=Duration.hours(2),
            max_retry_attempts=3
        ),
        dead_letter_queue=sqs.Queue(self, "DataLoadDLQ")
    )
)

これが地味に便利な理由が、タイムゾーン対応なんですよね。東京時間で「毎日午後3時」と指定できるんです。従来の EventBridge Rules は UTC 固定で、計算間違えて朝4時に走ってた。みたいなことが何度もあった。

痛い目を見た話

運用を始めて2週間で、Scheduler が SQS に 1時間ごとに無限再送信を続けてた事件がありました。原因は、ターゲットの SQS 権限が不足してたから。Scheduler はリトライを自動で続けるので、権限エラーに気づくのが遅れたんですよね。

学んだポイント：

  • Scheduler は「失敗した = リトライする」という設計。権限周りは本番投入前に3回は検証すべき
  • Max Event Age を設定しないと、古いメッセージが永遠に溜まる
  • Dead Letter Queue はほぼ必須。これがないと失敗の詳細が見えない

複合構成：Pipes + Scheduler で非同期処理パイプライン

うちでやった本格的な構成は、Scheduler で定期発火したイベントを、Pipes で変換・フィルタして、複数のターゲットに分散させるやつです。

xychart-beta
    title EventBridge Pipeline処理量 (1日)
    x-axis [00:00, 06:00, 12:00, 18:00, 24:00]
    y-axis "イベント数" 0 --> 5000
    line [200, 1800, 3200, 2100, 450]
    line [150, 900, 1400, 980, 200]
    line [50, 200, 400, 300, 100]

グラフの見方としては、上の線がSchedulerからのスケジュール実行、中がPipesでフィルタされたもの、下がDLQに送られたエラーイベント。ほぼ一貫性がありますね。

実装上の大事なポイントをまとめると：

Input Transformer のシンプル化 Pipesのデフォルト変換じゃ足りないケースが多いので、Lambda enrichmentにしがちですが、まずは Input Transformer（EventBridge ネイティブな JSON 変換）を試すんですよ。冷却時間ゼロですから。

DLQ の監視 DLQ にメッセージが溜まってることに気づかずに、本番で「あれ、イベント処理されてない」となりました。CloudWatch Alarms で DLQ の ApproximateNumberOfMessagesVisible を監視することが必須です。

Pipes と Scheduler の権限設計 IAM ロールの最小権限原則で、Pipes は「SQS SendMessage」だけ、Scheduler は「SQS SendMessage」だけ、という感じで分けるんです。共有ロールにしたらアクセス権のデバッグが地獄になります。

実運用で見えた良い点・しんどい点

ここからは、3ヶ月運用してわかったリアルな話です。

良かった点：

  • Lambda関数を4個削減できた → CloudWatch Logs のコストが月3万円減った
  • イベント変換の仕様が可視化された。Pipes の定義ファイル見たら「あ、ここでフィルタされてるのか」がわかる
  • Enrichment で外部API呼び出しする場合、Pipes側でタイムアウト・リトライを一括管理できる
  • テスト環境と本番環境で、cron式だけ変えて同じ定義を使い回せる

しんどかった点：

  • CloudWatch ログで Pipes のエラーを追跡しづらい。「なぜフィルタされたのか」「Enrichment で何が起きたのか」が黒箱になりやすい
  • Input Transformer の JSON Path 記法が複雑。$.detail.nested[0].value みたいなのはいいけど、条件付き変換とかになると、結局Lambda enrichmentに逃げることになる
  • Scheduler のタイムゾーン（Asia/Tokyo）と Lambda のデフォルトタイムゾーン（UTC）のズレで、初期段階で想定と違う時刻に走る事案が2回あった
  • DLQ を見落とすと、失敗したイベントがどこへ行ったのか不明。CloudFormation で DLQ も自動作成することが大事です

実装パターンの選択

実装パターンを比較するなら、こんな感じですね：

要件EventBridge PipesLambda + SQSStep Functions
イベント変換⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐
スケーラビリティ⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐
運用のしやすさ⭐⭐⭐⭐⭐⭐⭐⭐⭐
コスト最安中程度高い
デバッグの容易さ⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐
複雑なワークフロー⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐

正直に言うと、Pipes は「シンプルなイベント変換・ルーティング」に特化してる。複雑なビジネスロジックが必要なら、Step Functions + Lambda の方が長期的には楽ですよ。でも、「イベント A をフィルタしてイベント B に変換して SQS に送る」みたいなのなら、Pipes 一択です。

Scheduler は、従来の EventBridge Rules + Lambda（定期実行用）を完全に置き換えました。タイムゾーン対応と DLQ がデフォルトで使えるのがでかい。ただし、非常に複雑なスケジュール（例：「毎月第2営業日の午前9時」）は、Step Functions や Lambda + EventBridge Rules の組み合わせの方が安全かもしれません。

チームで共有した実装ガイドライン

うちのチームで、Pipes・Scheduler を導入する際の最低限のチェックリストを作ったんですけど、紹介しますね：

DLQ は必須

  • CloudFormation で Pipes 定義と同時に DLQ (SQS) を作成
  • CloudWatch Alarms で DLQ メッセージ数を監視

Input Transformer で足りるか、Lambda enrichment が必要か

  • JSON Path で変換できるなら Input Transformer
  • 外部 API 呼び出しや複雑なロジックなら Lambda enrichment
  • Lambda enrichment の場合は、必ず SnapStart 有効化

Scheduler のタイムゾーン

  • Asia/Tokyo に統一
  • cron 式は UTC ではなく、指定したタイムゾーンで解釈される点を確認

ログ戦略

  • CloudWatch Logs に Pipes のすべてのイベントを送出
  • CloudWatch Insights で「失敗したイベント」を集計するクエリを用意

テスト環境での検証

  • 本番投入前に、DLQ へのリトライが正しく動くか確認
  • 権限不足時の挙動（無限リトライ）が起きないか確認

イベント駆動とサーバーレスの相乗効果

EventBridge Pipes・Scheduler をサーバーレス基盤に組み込むと、マイクロサービス間の疎結合が飛躍的に高まります。

従来構成では、サービス A → Lambda（中継）→ サービス B みたいに、Lambda が「宛先を知る」必要がありました。Pipes を導入すると、サービス A はイベントを発火するだけ。「誰が使うのか」は Pipes の設定で一元管理される。結果、サービス間の依存性が減って、デプロイ順序も気にならなくなるんですよ。

Specifically、うちで見えた効果：

  • デプロイリスク低下：サービス A を更新しても、Pipes が変わらなければ他のサービスに影響なし
  • スケーリングが独立：Pipes のリージョン間レプリケーション設定 1 つで、複数リージョン運用が可能
  • 監視の一元化：CloudWatch で Pipes のスループットを見れば、全体的なイベント処理量が把握できる

まとめ

3ヶ月運用して、EventBridge Pipes と Scheduler は「小規模〜中規模のサーバーレスアーキテクチャに最適」という確信が持てました。特に以下の場合、導入を検討する価値があります：

  1. 複数のマイクロサービスからイベントが発火する → Pipes でフィルタ・変換を一元管理すると、Lambda関数を大幅削減できる

  2. 定期実行タスク（バッチ処理など）がある → Scheduler で cron 管理を一元化。タイムゾーン問題も消える

  3. イベントの形式が複数混在している → Pipes の Input Transformer で標準化。ただし複雑すぎるなら Lambda enrichment

  4. 本番環境での信頼性が重要 → DLQ + CloudWatch Alarms で、失敗したイベントを確実に追跡できる

もし Lambda 関数が「単なる中継」になってたら、Pipes への置き換えを検討する価値あり。権限周りのテストは 3 回やって、DLQ は必ず用意する。正直まだ完全には使いこなせてないけど、今のところ地味に便利な選択肢だなと感じてます。

U

Untanbaby

ソフトウェアエンジニア｜AWS / クラウドアーキテクチャ / DevOps

10年以上のIT実務経験をもとに、現場で使える技術情報を発信しています。 記事の誤りや改善点があればお問い合わせからお気軽にご連絡ください。

関連記事

サーバーレス設計

Lambda SnapStartで冷却60%削減したら予想外のバグに遭遇した話

サーバーレス設計

Lambda SnapStartで冷却時間が60%削減された話。Java/Pythonで本番検証

サーバーレス設計

DynamoDB Single Table Designを本番2年使って正直に書く｜RDS出身者が感じた壁と突破口

サーバーレス設計

HTTP APIに乗り換えて月30万円削減した話｜REST APIとの選び分け方

サーバーレス設計

AppSyncを1年本番運用して後悔したスキーマ設計の話【2026年版】

サーバーレス設計

AppSync GraphQL、チーム本番運用で1年しんどかった話と今の現実解

← 記事一覧に戻る