NiFi消费Kafka中Filebeat日志时过滤out of range消息异常

两个处理器运行异常的核心原因非常明确：你单条Kafka消息里的内容根本不是合法JSON，只是多个独立JSON对象首尾硬拼的字符串，末尾还带了个没闭合的半截{，所有依赖标准JSON解析的处理器都没法正常识别结构：

• EvaluateJsonPath解析JSON失败时，默认会直接把原始流文件原封不动路由到输出关系，所以你看到输出和输入完全一致，根本没做提取
• SplitJson要求输入必须是合法JSON，要么是单个对象，要么是能通过JsonPath定位到的数组节点，传这种非法拼接字符串进去会触发解析逻辑异常，导致内存错误、无限拆分，最后生成几GB的无效垃圾数据是必然结果。

你为了避免日志乱序做合批的思路本身没问题，但拼接的时候没把多条日志封装成标准可解析结构（比如JSON数组、换行分隔的NDJSON），是所有问题的源头。

给你两个可落地的方案，优先选方案1，流程更短性能更高，天然保证日志原有顺序：

方案1：预处理+QueryRecord 一步过滤（无需拆分，适合仅需提取message字段的场景） #

1. 第一步：预处理拼接字符串为合法JSON数组
   串联3个ReplaceText处理器依次做替换，每个处理器替换策略都选「Regex Replace」：

• 第一个ReplaceText：清除末尾半截无效内容，正则匹配\{\s*$，替换值留空，删掉末尾没闭合的{和连带的空白字符
• 第二个ReplaceText：补全JSON对象分隔符，正则匹配\}\s*\{，替换值填},{，把首尾相连的}{替换成带逗号的标准分隔格式
• 第三个ReplaceText：包裹为合法JSON数组，正则匹配(?s)^(.*)$，替换值填[$1]，给整段内容首尾加上方括号，处理完的内容就是标准的、由单条日志JSON组成的合法数组。

2. 第二步：直接查询过滤目标内容
   新增QueryRecord处理器，配置Reader选JsonTreeReader，Writer选JsonRecordSetWriter，新增一条自定义SQL查询：

SELECT message FROM FLOWFILE WHERE LOWER(message) LIKE '%out of range%'

处理器执行后，所有匹配规则的message会按原顺序输出，不需要拆分单条日志，性能远高于拆分后过滤的方案。

方案2：预处理+拆分+属性过滤（适合需要保留单条日志完整结构的场景） #

1. 第一步和方案1的预处理完全一致，先把内容处理成合法JSON数组
2. 新增SplitJson处理器，JsonPath表达式配置为$.*，此时输入是合法数组，处理器会正常把每个日志JSON拆为独立流文件，不会产生冗余数据
3. 新增EvaluateJsonPath处理器，配置提取规则：属性名填log_msg，JsonPath表达式填$.message，提取目标选「flowfile-attribute」，把每条日志的message字段提取为流文件属性
4. 新增RouteOnAttribute处理器，新增匹配规则：${log_msg:toLower():contains("out of range")}，把匹配该规则的关系连接到后续输出链路，不匹配的关系直接设置为自动终止即可。

如果可以调整Filebeat侧配置，完全不需要在NiFi侧做正则预处理：

• 合批发送时直接将多条日志封装为标准JSON数组结构，或者按NDJSON规范（每条JSON单独占一行）拼接
• 也可以直接用NiFi的ConsumeKafkaRecord_2_6处理器替换普通的ConsumeKafka处理器，配合对应Reader直接解析批量消息，减少预处理步骤，稳定性更高。

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