如何在Filebeat配置文件中过滤Kafka输入的符合指定条件的JSON数据

你的配置存在核心问题导致过滤规则失效：

• YAML转义逻辑错误：你在单引号包裹的include_lines值中额外对双引号添加了转义符\。按照YAML语法规范，单引号包裹的字符串内部的反斜杠不会被识别为转义符，会直接作为正则的一部分参与匹配，此时你实际的匹配规则是查找包含\"event\":\"basket\"（带反斜杠）的内容，和测试数据的实际格式不匹配，导致过滤规则没有命中任何数据，Filebeat默认保留所有未被过滤规则命中的内容，所以两组数据都被输出。
• 直接匹配原始字符串的方案可靠性低：很容易因为JSON格式的缩进、键值顺序变化、空格差异导致匹配失败，更推荐结构化解析后按字段过滤。

额外提示：你配置的kafka hosts列表中每个地址末尾的逗号属于冗余内容，YAML列表不需要额外加逗号分隔，建议删除避免后续配置修改时出现语法错误。

方案1：直接修正include_lines配置 #

去掉不必要的转义符即可：

include_lines: ['"event":"basket"']

方案2（更推荐）：结构化解析后按字段过滤 #

先解析JSON为结构化字段，再通过处理器精准过滤，不受JSON格式变化影响：

filebeat.inputs:
- type: kafka
  hosts:
    - qa-parkbae-01.hanpda.com:9092
    - qa-parkbae-02.hanpda.com:9092
    - qa-parkbae-03.hanpda.com:9092
  topics: ["parkbae-test-topic1"]
  group_id: "test123"
  ssl.enabled: false

processors:
  # 解析message字段的JSON内容，直接展开到根层级
  - decode_json_fields:
      fields: ["message"]
      target: ""
      overwrite_keys: true
  # 仅保留event字段等于basket的事件
  - drop_event:
      when:
        not:
          equals:
            event: basket

内容的提问来源于stack exchange，提问作者박배선