Amazon EventBridge Pipes

解決する課題

• 「SQS から取り出して加工してから別サービスへ渡す」ような連携を、Lambda のグルーコードを書かずに組みたい
• ストリームやキューを自前でポーリングするコードの運用負担を減らしたい
• 不要なイベントを早い段階でフィルタし、下流の処理量とコストを抑えたい
• 取り出したイベントに**別データを付け足して（エンリッチ）**からターゲットへ届けたい

主要概念と用語

• パイプ: ソースとターゲットを1対1でつなぐ単位。1本のパイプに1つのソースと1つのターゲットを定義する
• ソース: イベントの取り込み元。SQS・Kinesis Data Streams・DynamoDB Streams・Amazon MQ・Managed Streaming for Kafka（MSK）・セルフマネージドな Kafka など
• フィルタ: 取り込んだイベントのうち、パターンに一致したものだけを通す段階。EventBridge のイベントパターンと同じ書式
• エンリッチ: ターゲットへ渡す前にイベントへ情報を付け足す段階。Lambda・Step Functions・API Gateway・API Destinations を呼べる
• ターゲット: 最終的な届け先。Lambda・Step Functions・SNS・SQS・EventBridge バス・API Destinations など多数
• ターゲット入力トランスフォーマー: ターゲットへ渡すペイロードの形を整える変換テンプレート
• バッチ: ストリーム/キュー系ソースを複数件まとめて取り込み・処理する単位

仕様・制限・クォータ

• 1本のパイプは 1ソース対1ターゲットの point-to-point 構成で、ファンアウト（1対多）はしない
• 処理は フィルタ→エンリッチ→ターゲットの固定パイプラインで、フィルタとエンリッチは任意、ターゲットは必須
• ソースには**ポーリング型（SQS・Kinesis・DynamoDB Streams・Kafka・MQ）**を選べ、Pipes 側がポーリングを肩代わりする
• ストリーム系ソースではバッチサイズやバッチウィンドウ、並列度などを調整できる
• フィルタを通過しなかったイベントは下流へ渡らず、エンリッチやターゲットの呼び出し対象にもならない
• 失敗時のリトライと、処理しきれないレコードを退避する **DLQ（デッドレターキュー）**を設定できる

内部の仕組み

パイプを作成すると、Pipes はソースの種類に応じて取り込みを行います。SQS や Kinesis、DynamoDB Streams、Kafka のようなポーリング型ソースでは、Pipes 側がポーリングを肩代わりするため、利用者がポーラーを書いて運用する必要がありません。取り込んだイベントはまずフィルタで評価され、パターンに一致したものだけが次へ進みます。続くエンリッチ段階では、必要に応じて Lambda や Step Functions、API を呼び出し、その応答でイベントを加工・補完します。最後にターゲット入力トランスフォーマーで形を整えてからターゲットを呼び出します。各段階は実行ロールを引き受けて他サービスの API を呼ぶため、権限は IAM ロールで与えます。失敗したレコードは設定に従ってリトライされ、それでも処理できないものは DLQ へ送られます。

設計パターン / ベストプラクティス

• グルーコードの置き換え: 「キューから取り出して整形して別サービスへ渡す」だけの Lambda を、フィルタとトランスフォーマーで置き換えて運用対象を減らす
• 早期フィルタでコスト削減: 不要なイベントはフィルタ段階で落とし、エンリッチやターゲットの呼び出し回数そのものを減らす
• エンリッチで参照データを付与: ターゲットが必要とする付加情報を、エンリッチ段階の Lambda や API でまとめて取得して渡す
• ストリームの橋渡し: DynamoDB Streams や Kinesis のイベントを Step Functions や EventBridge バスへつなぎ、変更データを後続処理へ流す
• 段階ごとの責務分離: フィルタ・変換は Pipes に任せ、ビジネスロジックはターゲット側に寄せて見通しを良くする

運用・監視

• パイプの稼働状況や失敗は CloudWatch メトリクスで監視する（実行数・失敗数・処理遅延など）
• イベントがターゲットへ届かない場合は、まずフィルタのパターンで落とされていないかを確認する
• ターゲットは呼ばれるのに失敗する場合は、実行ロールの権限とターゲット側の入力フォーマットを確認する
• 処理しきれなかったレコードは DLQ に残るので、ここを定期的に確認して取りこぼしを検知する
• ストリーム系ソースでスループットが不足する場合は、バッチサイズや並列度を調整する

コスト

• 主にパイプが処理したイベント数に応じた課金で、フィルタで落としたイベントは課金対象にならない設計
• 自前のポーラーや単純なグルー Lambda を持たずに済むため、その運用・実行コストを削減できる
• エンリッチで呼び出す Lambda や API、ターゲットの実行費用はそれぞれのサービス側で別途発生する
• 具体的な単価や課金単位は変動しうるため、最新の料金ページで確認する

セキュリティ

• パイプがソース取得・エンリッチ・ターゲット呼び出しに使う権限は実行ロール（IAM ロール）で与え、各段階に必要なアクションだけに最小権限で絞る
• ソースやターゲットが別アカウントにある構成も組めるため、クロスアカウントの権限設計を明確にする
• イベントに機微なデータが含まれる場合は、ログ出力やエンリッチ先での扱いに注意し、必要に応じて暗号化や Secrets Manager 経由の取得を検討する
• KMS による暗号化を組み合わせ、保管・伝送中のデータ保護を行える

関連サービス・比較

ハンズオン / CLI例

# SQSキューをソースに、フィルタを通したイベントをStep Functionsへ渡すパイプ
aws pipes create-pipe --name order-pipe \
  --role-arn "arn:aws:iam::123456789012:role/PipeExecutionRole" \
  --source "arn:aws:sqs:ap-northeast-1:123456789012:orders" \
  --source-parameters '{
    "FilterCriteria": {
      "Filters": [
        {"Pattern": "{\"body\":{\"status\":[\"PAID\"]}}"}
      ]
    }
  }' \
  --target "arn:aws:states:ap-northeast-1:123456789012:stateMachine:FulfillOrder" \
  --target-parameters '{
    "StepFunctionStateMachineParameters": {"InvocationType": "FIRE_AND_FORGET"}
  }'

# 作成したパイプの状態を確認
aws pipes describe-pipe --name order-pipe