关键逻辑是“先筛再导”,需用jq等工具在导出前按level或severity字段过滤error记录,注意大小写、字段名差异及JSONL逐行解析,采集端过滤更高效。
导出时只保留 error 级别日志的过滤逻辑
关键不是“导出”,而是“先筛再导”。绝大多数日志工具(如 grep、awk、jq)不支持导出时原生过滤,得靠管道或条件表达式提前切掉非 error 记录。
常见错误是直接用 cat logs.json | jq '.[]' > errors.json —— 这导出全部,没过滤。真正要的是加条件判断:
jq 'select(.level == "error" or .level == "ERROR" or .severity == "ERROR")' logs.json > errors.json- 注意大小写:有些系统用
error,有些用ERROR,还有用severity字段的,得看实际数据结构 - 如果字段名不固定(比如有的是
type、有的是log_level),先用head -5 logs.json | jq看一眼结构再写条件
Python 读取 JSONL 日志并提取 error 行
JSONL(每行一个 JSON)比普通 JSON 更常见于日志导出场景,json 模块不能直接 load 整个文件,得逐行解析,否则会报 JSONDecodeError: Extra data。
实操建议:
- 用
with open("app.log", "r") as f:配合for line in f:逐行读取 - 每行用
json.loads(line.strip())解析,捕获json.JSONDecodeError跳过脏数据 - 检查字段存在性:用
record.get("level", "").lower() == "error",避免KeyError - 示例片段:
import json<br>errors = []<br>with open("app.log") as f:<br> for line in f:<br> try:<br> record = json.loads(line.strip())<br> if record.get("level", "").lower() == "error":<br> errors.append(record)<br> except json.JSONDecodeError:<br> pass<br>with open("errors.json", "w") as out:<br> json.dump(errors, out, indent=2)
Logstash 或 Fluentd 导出前就过滤 error
如果日志已接入采集链路,没必要导出全量再筛——在采集端过滤更省资源、也避免磁盘写入冗余数据。
Logstash 的 if 判断写法必须带括号,漏掉会静默失败:
-
if [level] == "error" { ... }✅ -
if [level] == "error" {...}❌(缺少空格,语法错误) - Fluentd 用
@type grep插件时,regexp匹配的是字符串全文,不是字段值;想按字段过滤得用@type record_transformer+keep_keys配合条件 - 注意时间字段:有些配置里
@timestamp是字符串,直接比较可能失败,需先用date过滤插件标准化
导出 CSV 时 error 字段为空或错位怎么办
CSV 不是结构化格式,强行把嵌套 JSON 日志转 CSV 容易丢字段或列错位,尤其当 error 是对象(如 {"message": "timeout", "code": 504})时。
解决方向很明确:别硬转 CSV,改用 jq 提取扁平字段:
jq -r 'select(.level=="error") | [.timestamp, .message, .error.code, .error.message] | @csv' logs.json > errors.csv- 如果
.error可能不存在,用.error?.code(问号操作符)避免整个记录被跳过 - Windows 下用
jq注意换行符:jq -r -M防止\r\n混入字段内容 - 真正要 CSV 就别留嵌套结构——要么提前展开,要么接受部分字段为
null
最常被忽略的是字段一致性:同一份日志里,有些 error 记录带堆栈,有些只有 message,导出成表格时列数会变。要么统一用 jq 做默认值填充,要么接受它本来就不适合 CSV。










