如何使用 EventBridge Pipes 篩選事件？

3 分的閱讀內容

我想使用 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"
}

**注意：**在上述範例事件中，您無法使用 eventSource、eventSourceARN 和 awsRegion 建立事件模式，因為 EventBridge 會在輪詢期間新增資料。

查看事件篩選模式範例

事件篩選模式與中繼資料屬性相符

以下的事件篩選模式會比對那些透過管道接，並包含 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 console (EventBridge 主控台)。
在導覽窗格中，選擇 Pipes (管道)。
選擇 Create pipe (建立管道)，然後輸入管道的名稱。
(選用) 在 Description (描述) 中，請輸入管道的描述。
在 Build pipe (建置管道) 索引標籤上，選取來源類型，然後設定來源。
在 Source (來源) 中，輸入參數，然後選擇 Next (下一步)。
在 Sample event type (範例事件類型) 下，選取 AWS events (AWS 事件) 或 Enter my own (輸入我自己的事件)。
在 Event pattern (事件模式) 下，輸入必要的事件模式來篩選事件。
(選用) 選擇 Test pattern (測試樣式)。
(選用) 選擇 Next (下一步) 以瀏覽至 Enrichment (擴充)。
選擇 Next (下一步) 以瀏覽至 Target (目標)。
在 Details (詳細資訊) 下的 Target service (目標服務) 中，從下拉式清單中選取目標，然後輸入目標資源資訊。
設定管道設定。
選擇 Create pipe (建立規則)。