如何使用 EventBridge 管道筛选事件？

2 分钟阅读

我想使用 Amazon EventBridge 管道从 AWS 源捕获事件并将该事件连接到目标。

简短描述

EventBridge 管道接收来自以下来源的事件数据：

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 管道使用事件模式。每个 EventBridge 管道源都有包含核心消息或数据的字段。

使用筛选条件来减少开销

为了筛选源发送的事件，EventBridge 管道使用以下筛选条件：

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 管道会丢弃该消息。

当您为事件的消息正文创建事件筛选模式时，该模式必须与源使用的格式相匹配。例如，SQS 的消息正文可以是纯字符串或 JSON 脚本。Kinesis 流和 DynamoDB 记录必须采用有效的 JSON 格式。自行管理的 Apache Kafka 流和 Amazon MQ 消息使用 UTF-8 编码的字符串，可以是纯字符串或 JSON 格式。

有关源及其筛选模式的详细信息，请参阅Amazon EventBridge 管道中的事件筛选。

创建 EventBridge 管道

完成以下步骤：

打开 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（创建管道）。