如何在将Filebeat日志发送至Elasticsearch前进行解析筛选？

第一步：让Filebeat正确解析NDJSON日志 #

你的日志是NDJSON格式但被当作纯文本存入message字段，需在Filebeat配置里开启NDJSON解析逻辑：
在采集该日志的input区块中添加以下配置：

- type: filestream
  id: filebeat-log-input
  paths:
    - /var/log/filebeat/filebeat-*.ndjson
  parsers:
    - ndjson:
        overwrite_keys: true
        add_error_key: true
        expand_keys: true

• ndjson解析器专门处理换行分隔的JSON日志
• overwrite_keys: 允许解析后的字段覆盖Filebeat默认字段
• add_error_key: 解析失败时自动添加_jsonparse_error字段用于排查问题
• expand_keys: 自动展开嵌套的JSON键（例如将foo.bar拆分为嵌套对象）

配置生效后，Filebeat会将每条NDJSON解析为独立的键值对，不再存入message字段（仅解析失败时保留原始内容）。

第二步：筛选仅保留感兴趣的字段 #

通过Filebeat的processors实现字段过滤，常用三种方式：

方式1：用include_fields保留指定字段 #

在Filebeat配置的processors区块添加：

processors:
  - include_fields:
      fields: ["@timestamp", "log.level", "agent.id", "your_target_field1", "your_target_field2"]

直接列出需要保留的字段，其余字段会被自动丢弃。

方式2：用drop_fields移除不需要的字段 #

若需保留的字段较多，仅丢弃少数冗余字段时使用：

processors:
  - drop_fields:
      fields: ["ecs", "agent.version", "log.file.path"]
      ignore_missing: true

• ignore_missing: 字段不存在时不会触发报错

方式3：用script处理器实现复杂筛选 #

需要按条件灵活筛选字段时，使用Painless脚本处理器：

processors:
  - script:
      lang: painless
      source: |
        def requiredFields = ["@timestamp", "log.level", "custom_business_field"];
        ctx.keySet().removeIf(key -> !requiredFields.contains(key));

这段脚本会遍历所有字段，仅保留requiredFields列表内的内容。

验证配置 #

修改完成后，重启Filebeat使配置生效：

systemctl restart filebeat

可通过filebeat test config验证配置语法正确性，再到Kibana的Discover页面查看数据是否已按预期解析和过滤。

内容的提问来源于stack exchange，提问作者Rayne