EventBridge Pipes でイベントをフィルターする方法を教えてください。

所要時間2分

Amazon EventBridge Pipes を使用して AWS ソースからイベントをキャプチャし、そのイベントをターゲットに接続しようとしています。

簡単な説明

EventBridge Pipes は、次のソースからイベントデータを受信します。

Amazon DynamoDB ストリーム
Amazon Kinesis ストリーム
Amazon MQ
Amazon Managed Streaming for Apache Kafka (Amazon MSK) ストリーム
セルフマネージド Apache Kafka ストリーム
Amazon Simple Queue Service (Amazon SQS) キュー

データイベントをソースからターゲットに送信すると、リソースのオーバーヘッドおよび、追加コストが発生します。ターゲットはすべてのデータイベントを処理し、不要なイベントを破棄するビジネスロジックを実装する必要があります。ターゲットには、特定のデータイベントのみを送信することをおすすめします。

フィルター処理したイベントをターゲットに送信する前に改善するには、API 宛先または次の AWS サービスを使用します。

Amazon API Gateway
AWS Lambda
AWS Step Functions

解決策

注: EventBridge Pipes はイベントパターンを使用します。すべての EventBridge Pipes のソースには、コアメッセージまたはデータを含むフィールドがあります。

フィルター条件を使用してオーバーヘッドを軽減する

EventBridge Pipes は、次のフィルター条件を使用してソースが送信したイベントをフィルター処理します。

Metadata プロパティには、データイベントを生成するソースに関する情報が JSON 形式で含まれています。
Data プロパティには、ソースからのデータイベントのメッセージ本文が含まれます。メッセージ本文の形式は、ソースによって異なります。

Amazon SQS キューが送信したメッセージ本文の例

{    "key": "example-key-1",
    "tag": "eb-pipe-filter",
    "object name": "sampleimage-11.png",
    "sequencer": "617f0837b476e463",
    "Order Status": "Approved"
}

パイプが受信したイベントの例

{    "messageId": "c9652226-2d8f-49e9-9f06-beeb2a6f55ce",
    "receiptHandle": "AQEBoteuDChpmsb6765tA//hqtPta1W/utSYlmWuw3fjZnVvY+dxOUdj4cEjYZtLKGMGT4LUuvWbQPHb96Wwa+X0zUJo8ZiHC7dbul62hVbFTXjB7+cOKH91RXc3YMllpnd6nB3CNAOGIig6suCpEOGcrhTJ6dC45KoacROua1PIH454Ji8AP51TkQSnhkkeOJHInLHkvrd2sfoBo037kmMKJSo3kDgU7y92jbpJphRmijnNQX6gPQYSVifF3n3ApvSBMGQIUzQb65ZRLl6Mp2VPJQYF2RRjYShdrzhe3uON9H01m9f3LGcpL3yaX6yqFahfoEe3PiT6MMXHvIAaj+dKVSUqmnnk3n7X0n4WX1uFPgyjrI5KuJ64j2lf/dkcPMQEUPorjmagECmYpMrGlkBbZw==",
    "body": {
        "key": "example-key-1",
        "tag": "eb-pipe-filter",
        "object name": "sampleimage-11.png",
        "sequencer": "617f0837b476e463",
        "Order Status": "Approved"
    },
    "attributes": {
        "ApproximateReceiveCount": "1",
        "SentTimestamp": "1690589578649",
        "SenderId": "AIDAIMWXWRRSWVPYSJEH6",
        "ApproximateFirstReceiveTimestamp": "1690589578654"
    },
    "messageAttributes": {
        "Name": {
            "stringValue": "TestMessage",
            "stringListValues": [],
            "binaryListValues": [],
            "dataType": "String"
        }
    },
    "md5OfMessageAttributes": "62bd1c2ea8c5344caa6852a77543892d",
    "md5OfBody": "388e8d9be1776f61c0be0a089807afbc",
    "eventSource": "aws:sqs",
    "eventSourceARN": "arn:aws:sqs:ap-southeast-2:123456789012:test-sqs-source-pipe",
    "awsRegion": "ap-southeast-2"
}

注: EventBridge はポーリング中にデータを追加するため、上記のイベント例では、eventSource、eventSourceARN、awsRegion を使用してイベントパターンを作成することはできません。

イベントのフィルターパターン例を確認する

イベントのフィルターパターンとメタデータプロパティの照合

次のイベントフィルターパターンは、パイプが取得したイベントを、メタデータプロパティ attributes および stringValue と照合します。

{  "attributes": {
    "ApproximateReceiveCount": ["1"]
  },
  "messageAttributes": {
    "Name": {
      "stringValue": ["TestMessage"]
    }
  }
}

イベントのフィルターパターンとデータプロパティの照合

次のイベントフィルターパターンは、パイプが取得したイベントを、データプロパティ Order Status および Approved 値と照合します。

{  "body": {
    "Order Status": ["Approved"]
  }
}

イベントのフィルターパターンと、メタデータおよびデータプロパティの照合

次のイベントフィルターパターンは、パイプが取得したイベントをデータプロパティ Order Status およびメタデータプロパティ SenderId と照合します。

{  "attributes": {
    "SenderId": ["AIDAIMWXWRRSWVPYSJEH6"]
  },
  "body": {
    "Order Status": ["Approved"]
  }
}

照合が行われないイベントフィルターパターン

次のイベントフィルターパターンは、パイプにフィールド awsRegion が含まれているため、パイプが受信したイベントとの照合が行われません。

{  "body": {
    "Order Status": ["Approved"]
  },
  "awsRegion": ["ap-southeast-2"]
}

注: EventBridge は、ポーリング中にデータを追加する際に awsRegion フィールドを追加します。データプロパティフィールドの照合は行われましたが、イベントフィルターパターンではエラーが発生します。

マッチ成功の基準をレビューする

受信メッセージ本文とイベントフィルターパターンに設定ミスがある場合、EventBridge Pipes はメッセージをドロップします。

イベントのメッセージ本文に対するイベントフィルターパターンを作成する場合、そのパターンはソースが使用する形式と一致する必要があります。たとえば、SQS のメッセージ本文には、プレーン文字列または JSON スクリプトを使用できます。Kinesis ストリームと DynamoDB レコードは有効な JSON 形式である必要があります。セルフマネージド Apache Kafka ストリームと Amazon MQ メッセージは、UTF-8 エンコードされた文字列をプレーン文字列または JSON 形式で使用します。

ソースとそのフィルターパターンの詳細については、「Amazon EventBridge Pipes でのイベントフィルター」を参照してください。

EventBridge パイプを作成する

次の手順を実行します。

EventBridge コンソールを開きます。
ナビゲーションペインで [パイプ] を選択します。
[パイプを作成] を選択し、パイプの名前を入力します。
(オプション) [説明] にパイプの説明を追加します。
[パイプを構築] タブでソースタイプを選択し、ソースを設定します。
[ソース] にパラメータを入力し、[次へ] を選択します。
[サンプルイベントタイプ] で [AWS イベント] または [独自のイベントを入力] を選択します。
[イベントパターン] に、イベントのフィルターに必要なイベントパターンを入力します。
(オプション) [パターンをテスト] を選択します。
(オプション) [次へ] を選択し、[エンリッチメント] に移動します。
[次へ] を選択し、[ターゲット] に移動します。
[詳細] の [ターゲットサービス] でドロップダウンリストからターゲットを選択し、ターゲットのリソース情報を入力します。
パイプ設定を行います。
[パイプを作成] を選択します。