Amazon EventBridge Pipes によるイベント統合 - ソースとターゲットの接続パターン
SQS・DynamoDB Streams・Kinesis などのイベントソースとターゲットを直接接続し、フィルタリングとエンリッチメントを挟む設計パターンを紹介します。
Pipes の接続パターン
EventBridge Pipes はイベントソースとターゲットをポイントツーポイントで接続するサービスです。従来は SQS キューのメッセージを Step Functions に渡すために Lambda 関数を作成する必要がありましたが、Pipes ではソースとターゲットを直接接続できます。サポートされるソースは SQS、Kinesis Data Streams、DynamoDB Streams、Amazon MSK、セルフマネージド Kafka で、ターゲットは EventBridge バス、Step Functions、Lambda、API Gateway、SNS、SQS など多数のサービスに対応しています。バッチサイズや並行性の設定でスループットを制御します。
フィルタリングとエンリッチメント
フィルタリングでは EventBridge のイベントパターン構文を使い、特定の条件に合致するイベントのみをパイプラインに通します。DynamoDB Streams から INSERT イベントのみを抽出する、SQS メッセージの特定フィールドが閾値を超えるものだけを処理するといった制御が可能です。エンリッチメントステップではフィルタリング後のイベントを Lambda 関数や API Gateway エンドポイントに送信し、外部データの付加や形式変換を行います。入力トランスフォーマーはイベントの JSON パスを指定してターゲットが期待する構造にマッピングし、コード不要のデータ変換を実現します。
ソースとターゲットの組み合わせ
Pipes のソースは SQS 、 Kinesis Data Streams 、 DynamoDB Streams 、 MSK 、 MQ 、セルフマネージド Kafka をサポートします。ターゲットは Step Functions 、 Lambda 、 ECS タスク、 EventBridge バス、 API Gateway 、 SNS 、 SQS など 15 以上のサービスに対応します。エンリッチメントステップで Lambda や API Gateway を呼び出し、イベントデータに外部情報を付加してからターゲットに配信します。入力トランスフォーマーでイベントの構造を変換し、ターゲットが期待する形式に整形します。 イベント統合のアーキテクチャを網羅的に学ぶなら、技術書 (Amazon)を参照してください。
Pipes の料金
EventBridge Pipes の料金はリクエスト数で課金され、64 KB チャンクあたり約 0.40 ドル/100 万リクエストです (2026 年 8 月時点)。フィルタリングでターゲットに配信するイベントを絞り込むと、ターゲット側の処理コスト (Lambda の実行回数、Step Functions の状態遷移) を削減できます。エンリッチメントの Lambda 呼び出しは Lambda の標準料金が別途発生します。Pipes を使わずに Lambda でグルーコードを書く場合と比較して、Lambda の実行コストとメンテナンスコストを削減できます。
エラーハンドリングとリトライ
Pipes はソースの種類に応じたエラー処理を備えています。SQS や Kinesis、DynamoDB Streams では、処理に失敗したメッセージを再試行し、規定回数を超えたものはデッドレターキュー (DLQ) に退避できます。これにより、一部のイベントが失敗してもパイプライン全体が滞らず、問題のあるイベントだけを後から調査・再処理できます。バッチで処理する場合は、失敗した項目だけを部分的に再試行する設定も可能で、正常なイベントの重複処理を避けられます。DLQ に溜まったイベントを監視し、根本原因の検知につなげます。
順序保証とスループット制御
Kinesis Data Streams や DynamoDB Streams をソースにする場合、シャード単位で順序が保たれます。並行処理数を上げてスループットを稼ぎつつ、同じシャード内の順序は維持されるため、順序が重要なワークロードでも安全にスケールできます。バッチサイズとバッチ処理の待機時間を調整すると、レイテンシとスループットのバランスを取れます。ターゲット側の処理能力に合わせてこれらを設定し、下流のサービスに過剰な負荷をかけないようにします。SQS のように順序保証が不要なソースでは、並行性を高めて効率を優先できます。
Step Functions との組み合わせ
Pipes のターゲットに Step Functions を指定すると、イベントをきっかけに複数ステップのワークフローを起動できます。従来はソースのポーリングとワークフロー起動の橋渡しに Lambda を書く必要がありましたが、その接着コードが不要になります。短時間で完結する処理には同期実行のステートマシン、時間のかかる処理には非同期実行を選び、用途に応じて使い分けます。フィルタリングでワークフローを起動すべきイベントだけに絞れば、状態遷移の回数を抑えてコストを最適化できます。イベント駆動の業務処理を簡潔に組み立てられます。
監視と運用のポイント
Pipes の稼働状況は CloudWatch のメトリクスで把握します。実行数、失敗数、各ステージの所要時間などを監視し、異常があればアラームで通知します。エンリッチメントに使う Lambda や API のレイテンシがボトルネックになると、パイプライン全体のスループットが頭打ちになるため、ステージごとの所要時間を見て改善点を特定します。設定はコードで管理し、フィルタパターンや変換ルールの変更を CI/CD でレビューしてから反映すると、本番での予期せぬ挙動変化を防げます。ログを残し、どのイベントがどう処理されたかを追跡できるようにします。
まとめ
EventBridge Pipes はイベントソースとターゲットの直接接続により、グルーコードの Lambda を削減します。SQS、Kinesis、DynamoDB Streams など 6 種類のソースから 15 以上のターゲットに接続し、フィルタリングとエンリッチメントでイベントの選別と加工を実行します。入力トランスフォーマーでターゲットが期待する形式に変換し、統合コードの保守コストを大幅に削減します。