接手过线上服务的人应该都有过这种经历凌晨告警响了登录服务器在几千行日志里 CtrlF 找那一个 requestId人还没找到眼睛先花了。尤其是日志全部以 JSON 形式落盘、请求上下文又层层嵌套的时候普通文本搜索工具基本处于报废状态。今天聊的这个 Python 日志提取工具就是专门为这种场景写的——它能把海量嵌套的 JSON 日志拉平、过滤、转成结构化表格让你从“满屏括号里找关键字”直接跳到“打开 Excel 就能按条件筛”。这个工具解决的痛点很明确运维排查、后端定位、数据分析师们拿到一堆多行 JSON 日志但缺少一个能“按字段查询、按路径提取、按条件导出”的轻量方案。网上现成的日志平台一般又重又繁琐本地临时分析时最需要的就是这种几十行代码、随取随用的小脚本。下面我把整个设计思路、核心实现、踩坑过程完整拆开适合有一定 Python 基础、想自己写类似工具的人参考零基础的读者按步骤操作也能跑通。1. 需求拆解嵌套 JSON 日志为什么让人头大1.1 日志样本长什么样先说最常见的形态。现在的微服务框架大多会把日志直接输出成 JSON比如下面这种{ timestamp: 2025-06-01T10:24:31.248Z, level: INFO, service: order-service, trace: { traceId: abc123, spanId: span-001, parentSpanId: null }, message: create order success, context: { userId: 87291, order: { orderId: ORD-20250601-001, items: [ {sku: A1001, price: 199.9, count: 2}, {sku: B2200, price: 59.0, count: 1} ], totalAmount: 458.8 }, requestInfo: { ip: 10.0.12.33, userAgent: Mozilla/5.0 ... } } }平时排查问题你想知道某个订单号的请求是不是失败了会怎么做大多数人都是先grep ORD-20250601-001然后把命中的几十行 JSON 全拖出来一个一个用眼睛扫。要是每行日志结构都一样倒还好麻烦的是不同服务的日志字段并不完全相同有的多一层context.order有的少一个context.requestInfo字段层级全看开发同学当时怎么打的日志。1.2 文本查找救不了你的三个原因第一JSON 日志里的字段路径本身就长。你在搜索框里输入context.order.orderId或者只看orderId碰巧嵌套层次深的字段名相同比如items[0].id和request.id同时存在普通搜索会把不相关的结果全带出来只能继续靠人工分辨。第二多行 JSON 和单行 JSON 处理方式不一样。有的日志系统会把一个对象拆成多行输出有的会整行打印。多行日志用grep匹配时你只看到命中片段前后文切断后非常难读。单行日志虽然好grep但一行可能长达几千字符终端显示又自动换行肉眼根本对不齐嵌套层级。第三日志里的字段值类型多样提取后的二次分析很费劲。你想统计接口平均耗时用文本工具只能搜出latency: 123然后还得自己复制数字到表格里算。与其每次反复做这种手工活不如直接让工具把每个字段提出来、铺平成一行后续用 Excel 或者 pandas 随便怎么算都行。一句话归纳嵌套 JSON 日志之所以“看晕”不是因为信息不够而是因为信息缺少一个稳定、可重复的“展开和定位”机制。我们这个工具要做的就是这个机制。2. 整体设计先想清楚再写代码2.1 技术栈与依赖选择核心语言就是 Python因为生态里处理 JSON、CSV、Excel 都很方便适合做这种“本地小工具”。依赖方面我会尽量只用标准库这样你拿过去在公司内网、离线环境直接跑不用折腾pip install。需要用到这几个标准库json解析每行日志。这个库官方文档说得很清楚但真正用的时候你会发现它才是全场核心。csv把抽出来的平面化字段输出成表格方便导入 Excel 或数据库。argparse让脚本能在命令行里通过参数指定输入文件、输出文件、过滤条件避免每次改代码。os和sys处理路径、编码和异常退出。如果条件允许可以再装第三方库pandas做后续数据分析。但我特意不把它写进工具核心因为很多服务器上没法随便装包标准库方案是兼容性最好的底裤。当然如果你本机有pandas等 CSV 输出后再pd.read_csv()做透视表效果会好很多。2.2 提取流程的四段式设计整个工具我拆成四步读取 → 解析 → 展平 → 过滤输出。读取原始日志按行流式读取避免一次加载几个 GB 文件 ↓ 解析每行为 JSON 对象跳过坏行 ↓ 递归展平嵌套字段例如 context.order.orderId ↓ 按过滤条件筛选 输出 CSV / JSONL选择“按行读取”而不是json.load(整个文件)原因很简单生产环境的日志文件动辄几百 MB 甚至几个 GB一次性读进内存会直接把你测试机搞到 swap。按行读取每次只保留一行在内存里虽然 Python 处理慢一点但至少不会挂。实际测试里一个 700MB 的日志文件普通笔记本跑完大概也就 40 秒到一分多钟对一个排查工具来说完全能接受。过滤条件设计成“字段路径 期望值”的字典形式例如{ service: user-service, level: ERROR, context.method: payment }这样做的好处是简单直观而且能天然支持多条件“且”的关系。如果需要“或”把规则拆成一个包含多个字典的列表任何一个字典命中都算匹配即可。2.3 为什么选择“展平”而不是“保留原始结构”日志本身是树形嵌套的但输出表格时绝大多数分析场景都更希望“一行一个对象、每列一个字段”。展平之后字段名变成点路径比如context.order.totalAmount列一展开清清楚楚。我也试过直接保留嵌套结构转成 JSONL 输出结果发现用途受限别人拿到文件还得自己二次解析。反而展平成 CSV 之后不光能用 Excel 做筛选还能导入字段权限受限的数据库做归档甚至作为数据中台的原始明细表。所以除非你下游系统明确要求原始 JSON否则展平是更省事的选择。3. 核心实现递归展平与条件过滤3.1 读取大文件的正确姿势代码第一段先解决“读文件”这个基础问题。有一个细节很多人会忽略日志文件编码经常不是 UTF-8可能是 GBK 或者带 BOM。我封装了一个带编码兜底的读取函数def open_log_file(path): for enc in (utf-8, gbk, utf-8-sig): try: f open(path, r, encodingenc) f.peek() return f except UnicodeDecodeError: f.close() continue return open(path, r, encodingutf-8, errorsignore)f.peek()会触发读取一小段内容如果编码不对在打开文件当下就抛出异常而不是等你处理几百行之后才报错。加errorsignore兜底是为了防止少数极端脏数据让整个脚本崩溃。接着按行读取每行尝试解析为 JSONdef iter_json_lines(file_handle): for line in file_handle: line line.strip() if not line: continue if line.startswith({) or line.startswith([): try: yield json.loads(line) except json.JSONDecodeError: continue这里只处理以{或[开头的行可以提前过滤掉那些框架自己打印的纯文本日志比如启动 banner 或堆栈信息。堆栈信息通常会多行且不是合法 JSON直接跳过即可。3.2 将嵌套 JSON 递归展平这是整个工具的核心函数。它的目标很简单把一个嵌套字典变成一层字典键名使用点号表示层级。数组则用下标[0]、[1]来表示这样items[0].price和items[1].price会成为不同的字段。def flatten_json(data, parent_key, sep.): flattened {} if isinstance(data, dict): for key, value in data.items(): new_key f{parent_key}{sep if parent_key else }{key} flattened.update(flatten_json(value, new_key, sepsep)) elif isinstance(data, list): for index, item in enumerate(data): new_key f{parent_key}[{index}] flattened.update(flatten_json(item, new_key, sepsep)) else: flattened[parent_key] data return flattened逻辑不复杂但有两个点值得说。一是当父级字典里同时存在普通键和子字典时flattened.update()可以让嵌套结果自然合并不会互相覆盖。二是列表里如果有对象items[0].price和items[1].price是并发存在的不会互相打架如果列表里装的是纯数字那么[0]、[1]这些路径可以直接取出标量。实际操作中你还会遇到一个情况JSON 字段值为null。flatten_json会把None当作普通值存下来输出 CSV 时就是空单元格这样不影响下游判断。如果想把null直接丢弃可以在递归出口处加一个判断if data is not None但我不建议这么干因为“字段不存在”和“字段为 null”在问题排查时有完全不同的含义。3.3 过滤规则怎么设计拿到展平后的字典下一步就是过滤。我的方案是“点路径精确匹配 可选模糊匹配”双模式。def evaluate_record(flat_record, filters): if not filters: return True for field, expected in filters.items(): if field not in flat_record: return False actual flat_record[field] if callable(expected): if not expected(actual): return False elif isinstance(expected, list): if actual not in expected: return False elif str(expected).startswith(~): keyword str(expected)[1:] if keyword not in str(actual): return False else: if actual ! expected: return False return True这里的callable分支支持你直接传一个 lambda想怎么折腾都行。比如过滤“耗时超过 1000ms 的错误请求”lambda v: isinstance(v, (int, float)) and v 1000字符串以 ~ 开头这个约定是给命令行用的写成context.method~payment就能把包含payment的请求都捞出来比精确匹配灵活得多。注意这个“~”前缀是我自己的约定如果你改成别的符号记得一并改文档。3.4 输出格式与落盘策略过滤完的记录我习惯同时输出两个格式CSV 便于直接打开JSONL 便于保留原始结构供程序二次处理。CSV 输出最大的坑是字段名不一致。不同日志行的字段集合可能不同比如 A 行有context.userIdB 行没有。如果直接用csv.DictWriter按第一行的字段名写后面行的额外字段会被丢弃。解决办法是跑两遍第一遍扫出所有可能的字段名第二遍再写文件。def collect_all_fields(records): all_fields [] seen set() for record in records: for field in record.keys(): if field not in seen: seen.add(field) all_fields.append(field) return all_fields但这意味着要把记录缓存进内存对超大日志不太友好。折中办法是先设定一个--fields参数让用户指定要输出的字段如果没指定工具默认输出前 100 条记录里出现过的全部字段并打一行警告提示“字段列表可能不完整”。这种取舍在实战中非常实用因为绝大多数时候我们关心的字段就那么几个直接指定反而更环保。JSONL 输出就比较简单直接把扁平化之后的字典json.dumps(..., ensure_asciiFalse)写到文件里就行字段名同样是点路径。4. 实战演示从原始日志到筛选结果4.1 准备测试数据为了让你直观看到效果我先构造一个小型样例。假设日志文件app.log里有三行分别是成功请求、失败请求和普通信息{timestamp: 2025-06-01T10:24:31.248Z, level: INFO, service: order-service, trace: {traceId: abc123}, message: create order success, context: {userId: 87291, order: {orderId: ORD-001, items: [{sku: A1001, price: 199.9, count: 2}], totalAmount: 399.8}}} {timestamp: 2025-06-01T10:25:12.109Z, level: ERROR, service: order-service, trace: {traceId: xyz999}, message: payment timeout, context: {userId: 87291, order: {orderId: ORD-002, items: [{sku: B2200, price: 59.0, count: 1}], totalAmount: 59.0}, requestInfo: {method: payment, latency: 1520}}} {timestamp: 2025-06-01T10:25:40.542Z, level: WARN, service: user-service, trace: {traceId: qwe000}, message: load high, context: {userId: 10086, requestInfo: {latency: 83}}}如果直接看原始日志人眼要找“所有serviceorder-service且levelERROR的记录”得来回扫好几遍。接下来我们用工具跑一遍。4.2 执行命令与运行结果我把完整脚本命名为jsonlog_extract.py命令行参数如下python jsonlog_extract.py \ --input app.log \ --output result.csv \ --filter serviceorder-service \ --filter levelERROR \ --fields timestamp,level,service,context.order.orderId,context.requestInfo.method,context.requestInfo.latencyargparse支持同一个--filter出现多次每次解析成键值的形式然后自动转成字典传给evaluate_record。跑完之后result.csv里是这样timestamp,level,service,context.order.orderId,context.requestInfo.method,context.requestInfo.latency 2025-06-01T10:25:12.109Z,ERROR,order-service,ORD-002,payment,1520我特意把fields列表压缩到几个关键字段结果一眼就能看出这条错误来自订单ORD-002触发点是payment操作耗时 1520ms。之前那种“眼睛盯屏幕找出错请求还要确认是哪个接口”的过程到这里就结束了。如果你不指定--fields工具会输出所有字段。但那样 CSV 的列会非常宽建议实际使用时按需指定。4.3 结果解读与后续加工拿到 CSV 之后很多人会直接扔进 Excel 筛选。这里我多说一个进阶操作如果你本地装了 pandas可以继续做聚合分析import pandas as pd df pd.read_csv(result.csv) error_by_service df.groupby(service).size() avg_latency_by_method df.groupby(context.requestInfo.method)[context.requestInfo.latency].mean()比如你发现某类接口平均耗时明显偏高就可以倒回去看这个接口相关的日志。这个流程打通后你的排查速度会有明显提升——从“打开日志文件到处翻”变成“先跑一次提取脚本再对着表格看趋势”。我还建议把提取结果按时间排序后输出。日志本身不保证有序但 CSV 默认是按读取顺序来的所以最好在输出前按timestamp排序。代价是需要把全部记录载入内存对超大文件来说不一定划算。我的做法是提供一个--sort-by timestamp参数只在用户明确需要排序时才加载全量数据。5. 常见问题与排雷实录5.1 高频坑位速查表现象原因解决办法脚本跑着跑着内存飙升读取时把整行 JSON 加载后又缓存了全量记录检查代码里是否在循环外追加记录输出前不要无条件collect_all_fieldsCSV 里数字变成199.9却无法求和因为json.loads把数字解析成 float/int写 CSV 时被转成字符串这是 CSV 格式限制导入 Excel 后手动转数字或用 pandas 读取同名字段被后一条记录覆盖展平函数遇到相同点路径必然覆盖确认是否需要保留数组下标必要时给每行加自增序号某些行没有输出原始行不是合法 JSON被json.JSONDecodeError吞掉了加--debug模式打印被跳过的行号人工判断是否重要Windows 下 CSV 打开乱码UTF-8 编码不被 Excel 直接识别输出时加 BOM或指定--encoding gbk过滤条件匹配不上字段实际路径是context.userId你写成了context.user_id先不加过滤跑一次把所有字段名列出来逐一确认特别说一下Windows 乱码这个坑。同事用你脚本导出 CSV 在 Excel 里打开全是乱码不是数据错了是编码标志问题。解决办法很简单if output_format csv and output_encoding.lower() utf-8: csv_file.write(\ufeff) # 写入 UTF-8 BOM或者干脆命令行加--encoding gbk参数Excel 对 GBK 的支持一向完美。一个小细节能救你一整个下午。5.2 我自己踩过的三个坑第一个坑是递归展平遇到“循环引用”。JSON 本身不允许循环引用但你解析的日志数据如果是从某个接口拿来的偶尔会有数据结构里带了引用标记之类的东西导致isinstance(data, dict)永远走不到出口。我后来给递归函数加了一个深度计数器超过 10 层就强制把当前值转成字符串返回。虽然会丢一点嵌套信息但至少脚本不会死循环。第二个坑是时间字段的时区问题。很多日志的timestamp是带Z后缀的 UTC 时间导进 Excel 后没法直接和本地时间对比。我养成的习惯是--fix-timezone参数默认转成本地时区再输出from datetime import datetime, timezone, timedelta def to_local_time(ts_str): dt datetime.fromisoformat(ts_str.replace(Z, 00:00)) return dt.astimezone(timezone(timedelta(hours8))).isoformat()这个逻辑很简单但排查问题时特别救命。比如你按“下午三点”搜索结果日志时间全是“07:00”就会浪费很多时间。第三个坑是“多行日志合并”。有的日志框架会把堆栈信息换行输出导致一条日志的 JSON 主体被拆成两行。按行解析时第二条堆栈行不是合法 JSON直接被跳过。如果你需要保留完整堆栈就得做“按大括号计数法”的合并。我的做法是当一行内容里{和}数量不相等时继续读取下一行拼接直到括号数平衡。def iter_complete_json_lines(file_handle): buffer for line in file_handle: buffer line if buffer.count({) - buffer.count(}) 0: yield buffer.strip() buffer 注意这个方法对字符串里包含大括号的日志会误判但对绝大多数日志文本是可用的。如果你实在不放心就多写一个“状态机”严格处理引号那就是另一个项目了。5.3 让工具更好用的几个小扩展如果这个脚本你想日常一直用我建议再加三个功能。一是“字段名自动映射”。不同服务的日志经常用user_id和userId表示同一个东西你可以在脚本里写一张别名表展平之后再统一改名这样下游分析代码就不用跟着改了。二是“JSON 路径通配符”。处理数组时context.items[*].price这种写法很实用。实现方式是在展平函数里对列表下标改成*标记匹配时用正则或拆段遍历。代价是脚本复杂度上升不少但对数据清洗场景收益很大。三是“输出摘要统计”。在命令行里加一个--summary跑完直接打印一个简单表格每个level有多少条、每个service有多少条。不用打开 Excel 就能对数据量心里有数。我一般会把它默认打开省得每次输出文件还要自己去数行数。根据我个人经验这个日志提取工具最值得投入时间的地方不在解析本身而在“过滤条件的灵活度”和“输出字段的可控性”。把这两点做好脚本就能从一次性工具升级成团队的通用排查入口。如果你后续遇到类似问题不妨先把自己最常用的过滤场景写清楚再动手实现比一开始就追求全功能高效得多。最后再分享一个小技巧把提取后的 CSV 文件统一按日期归档比如logs_extract/2025-06-01/result.csv这样半年后想复盘某天的故障直接按目录找不用翻命令历史。
Python日志提取工具：将嵌套JSON展平为可筛选表格
接手过线上服务的人应该都有过这种经历凌晨告警响了登录服务器在几千行日志里 CtrlF 找那一个 requestId人还没找到眼睛先花了。尤其是日志全部以 JSON 形式落盘、请求上下文又层层嵌套的时候普通文本搜索工具基本处于报废状态。今天聊的这个 Python 日志提取工具就是专门为这种场景写的——它能把海量嵌套的 JSON 日志拉平、过滤、转成结构化表格让你从“满屏括号里找关键字”直接跳到“打开 Excel 就能按条件筛”。这个工具解决的痛点很明确运维排查、后端定位、数据分析师们拿到一堆多行 JSON 日志但缺少一个能“按字段查询、按路径提取、按条件导出”的轻量方案。网上现成的日志平台一般又重又繁琐本地临时分析时最需要的就是这种几十行代码、随取随用的小脚本。下面我把整个设计思路、核心实现、踩坑过程完整拆开适合有一定 Python 基础、想自己写类似工具的人参考零基础的读者按步骤操作也能跑通。1. 需求拆解嵌套 JSON 日志为什么让人头大1.1 日志样本长什么样先说最常见的形态。现在的微服务框架大多会把日志直接输出成 JSON比如下面这种{ timestamp: 2025-06-01T10:24:31.248Z, level: INFO, service: order-service, trace: { traceId: abc123, spanId: span-001, parentSpanId: null }, message: create order success, context: { userId: 87291, order: { orderId: ORD-20250601-001, items: [ {sku: A1001, price: 199.9, count: 2}, {sku: B2200, price: 59.0, count: 1} ], totalAmount: 458.8 }, requestInfo: { ip: 10.0.12.33, userAgent: Mozilla/5.0 ... } } }平时排查问题你想知道某个订单号的请求是不是失败了会怎么做大多数人都是先grep ORD-20250601-001然后把命中的几十行 JSON 全拖出来一个一个用眼睛扫。要是每行日志结构都一样倒还好麻烦的是不同服务的日志字段并不完全相同有的多一层context.order有的少一个context.requestInfo字段层级全看开发同学当时怎么打的日志。1.2 文本查找救不了你的三个原因第一JSON 日志里的字段路径本身就长。你在搜索框里输入context.order.orderId或者只看orderId碰巧嵌套层次深的字段名相同比如items[0].id和request.id同时存在普通搜索会把不相关的结果全带出来只能继续靠人工分辨。第二多行 JSON 和单行 JSON 处理方式不一样。有的日志系统会把一个对象拆成多行输出有的会整行打印。多行日志用grep匹配时你只看到命中片段前后文切断后非常难读。单行日志虽然好grep但一行可能长达几千字符终端显示又自动换行肉眼根本对不齐嵌套层级。第三日志里的字段值类型多样提取后的二次分析很费劲。你想统计接口平均耗时用文本工具只能搜出latency: 123然后还得自己复制数字到表格里算。与其每次反复做这种手工活不如直接让工具把每个字段提出来、铺平成一行后续用 Excel 或者 pandas 随便怎么算都行。一句话归纳嵌套 JSON 日志之所以“看晕”不是因为信息不够而是因为信息缺少一个稳定、可重复的“展开和定位”机制。我们这个工具要做的就是这个机制。2. 整体设计先想清楚再写代码2.1 技术栈与依赖选择核心语言就是 Python因为生态里处理 JSON、CSV、Excel 都很方便适合做这种“本地小工具”。依赖方面我会尽量只用标准库这样你拿过去在公司内网、离线环境直接跑不用折腾pip install。需要用到这几个标准库json解析每行日志。这个库官方文档说得很清楚但真正用的时候你会发现它才是全场核心。csv把抽出来的平面化字段输出成表格方便导入 Excel 或数据库。argparse让脚本能在命令行里通过参数指定输入文件、输出文件、过滤条件避免每次改代码。os和sys处理路径、编码和异常退出。如果条件允许可以再装第三方库pandas做后续数据分析。但我特意不把它写进工具核心因为很多服务器上没法随便装包标准库方案是兼容性最好的底裤。当然如果你本机有pandas等 CSV 输出后再pd.read_csv()做透视表效果会好很多。2.2 提取流程的四段式设计整个工具我拆成四步读取 → 解析 → 展平 → 过滤输出。读取原始日志按行流式读取避免一次加载几个 GB 文件 ↓ 解析每行为 JSON 对象跳过坏行 ↓ 递归展平嵌套字段例如 context.order.orderId ↓ 按过滤条件筛选 输出 CSV / JSONL选择“按行读取”而不是json.load(整个文件)原因很简单生产环境的日志文件动辄几百 MB 甚至几个 GB一次性读进内存会直接把你测试机搞到 swap。按行读取每次只保留一行在内存里虽然 Python 处理慢一点但至少不会挂。实际测试里一个 700MB 的日志文件普通笔记本跑完大概也就 40 秒到一分多钟对一个排查工具来说完全能接受。过滤条件设计成“字段路径 期望值”的字典形式例如{ service: user-service, level: ERROR, context.method: payment }这样做的好处是简单直观而且能天然支持多条件“且”的关系。如果需要“或”把规则拆成一个包含多个字典的列表任何一个字典命中都算匹配即可。2.3 为什么选择“展平”而不是“保留原始结构”日志本身是树形嵌套的但输出表格时绝大多数分析场景都更希望“一行一个对象、每列一个字段”。展平之后字段名变成点路径比如context.order.totalAmount列一展开清清楚楚。我也试过直接保留嵌套结构转成 JSONL 输出结果发现用途受限别人拿到文件还得自己二次解析。反而展平成 CSV 之后不光能用 Excel 做筛选还能导入字段权限受限的数据库做归档甚至作为数据中台的原始明细表。所以除非你下游系统明确要求原始 JSON否则展平是更省事的选择。3. 核心实现递归展平与条件过滤3.1 读取大文件的正确姿势代码第一段先解决“读文件”这个基础问题。有一个细节很多人会忽略日志文件编码经常不是 UTF-8可能是 GBK 或者带 BOM。我封装了一个带编码兜底的读取函数def open_log_file(path): for enc in (utf-8, gbk, utf-8-sig): try: f open(path, r, encodingenc) f.peek() return f except UnicodeDecodeError: f.close() continue return open(path, r, encodingutf-8, errorsignore)f.peek()会触发读取一小段内容如果编码不对在打开文件当下就抛出异常而不是等你处理几百行之后才报错。加errorsignore兜底是为了防止少数极端脏数据让整个脚本崩溃。接着按行读取每行尝试解析为 JSONdef iter_json_lines(file_handle): for line in file_handle: line line.strip() if not line: continue if line.startswith({) or line.startswith([): try: yield json.loads(line) except json.JSONDecodeError: continue这里只处理以{或[开头的行可以提前过滤掉那些框架自己打印的纯文本日志比如启动 banner 或堆栈信息。堆栈信息通常会多行且不是合法 JSON直接跳过即可。3.2 将嵌套 JSON 递归展平这是整个工具的核心函数。它的目标很简单把一个嵌套字典变成一层字典键名使用点号表示层级。数组则用下标[0]、[1]来表示这样items[0].price和items[1].price会成为不同的字段。def flatten_json(data, parent_key, sep.): flattened {} if isinstance(data, dict): for key, value in data.items(): new_key f{parent_key}{sep if parent_key else }{key} flattened.update(flatten_json(value, new_key, sepsep)) elif isinstance(data, list): for index, item in enumerate(data): new_key f{parent_key}[{index}] flattened.update(flatten_json(item, new_key, sepsep)) else: flattened[parent_key] data return flattened逻辑不复杂但有两个点值得说。一是当父级字典里同时存在普通键和子字典时flattened.update()可以让嵌套结果自然合并不会互相覆盖。二是列表里如果有对象items[0].price和items[1].price是并发存在的不会互相打架如果列表里装的是纯数字那么[0]、[1]这些路径可以直接取出标量。实际操作中你还会遇到一个情况JSON 字段值为null。flatten_json会把None当作普通值存下来输出 CSV 时就是空单元格这样不影响下游判断。如果想把null直接丢弃可以在递归出口处加一个判断if data is not None但我不建议这么干因为“字段不存在”和“字段为 null”在问题排查时有完全不同的含义。3.3 过滤规则怎么设计拿到展平后的字典下一步就是过滤。我的方案是“点路径精确匹配 可选模糊匹配”双模式。def evaluate_record(flat_record, filters): if not filters: return True for field, expected in filters.items(): if field not in flat_record: return False actual flat_record[field] if callable(expected): if not expected(actual): return False elif isinstance(expected, list): if actual not in expected: return False elif str(expected).startswith(~): keyword str(expected)[1:] if keyword not in str(actual): return False else: if actual ! expected: return False return True这里的callable分支支持你直接传一个 lambda想怎么折腾都行。比如过滤“耗时超过 1000ms 的错误请求”lambda v: isinstance(v, (int, float)) and v 1000字符串以 ~ 开头这个约定是给命令行用的写成context.method~payment就能把包含payment的请求都捞出来比精确匹配灵活得多。注意这个“~”前缀是我自己的约定如果你改成别的符号记得一并改文档。3.4 输出格式与落盘策略过滤完的记录我习惯同时输出两个格式CSV 便于直接打开JSONL 便于保留原始结构供程序二次处理。CSV 输出最大的坑是字段名不一致。不同日志行的字段集合可能不同比如 A 行有context.userIdB 行没有。如果直接用csv.DictWriter按第一行的字段名写后面行的额外字段会被丢弃。解决办法是跑两遍第一遍扫出所有可能的字段名第二遍再写文件。def collect_all_fields(records): all_fields [] seen set() for record in records: for field in record.keys(): if field not in seen: seen.add(field) all_fields.append(field) return all_fields但这意味着要把记录缓存进内存对超大日志不太友好。折中办法是先设定一个--fields参数让用户指定要输出的字段如果没指定工具默认输出前 100 条记录里出现过的全部字段并打一行警告提示“字段列表可能不完整”。这种取舍在实战中非常实用因为绝大多数时候我们关心的字段就那么几个直接指定反而更环保。JSONL 输出就比较简单直接把扁平化之后的字典json.dumps(..., ensure_asciiFalse)写到文件里就行字段名同样是点路径。4. 实战演示从原始日志到筛选结果4.1 准备测试数据为了让你直观看到效果我先构造一个小型样例。假设日志文件app.log里有三行分别是成功请求、失败请求和普通信息{timestamp: 2025-06-01T10:24:31.248Z, level: INFO, service: order-service, trace: {traceId: abc123}, message: create order success, context: {userId: 87291, order: {orderId: ORD-001, items: [{sku: A1001, price: 199.9, count: 2}], totalAmount: 399.8}}} {timestamp: 2025-06-01T10:25:12.109Z, level: ERROR, service: order-service, trace: {traceId: xyz999}, message: payment timeout, context: {userId: 87291, order: {orderId: ORD-002, items: [{sku: B2200, price: 59.0, count: 1}], totalAmount: 59.0}, requestInfo: {method: payment, latency: 1520}}} {timestamp: 2025-06-01T10:25:40.542Z, level: WARN, service: user-service, trace: {traceId: qwe000}, message: load high, context: {userId: 10086, requestInfo: {latency: 83}}}如果直接看原始日志人眼要找“所有serviceorder-service且levelERROR的记录”得来回扫好几遍。接下来我们用工具跑一遍。4.2 执行命令与运行结果我把完整脚本命名为jsonlog_extract.py命令行参数如下python jsonlog_extract.py \ --input app.log \ --output result.csv \ --filter serviceorder-service \ --filter levelERROR \ --fields timestamp,level,service,context.order.orderId,context.requestInfo.method,context.requestInfo.latencyargparse支持同一个--filter出现多次每次解析成键值的形式然后自动转成字典传给evaluate_record。跑完之后result.csv里是这样timestamp,level,service,context.order.orderId,context.requestInfo.method,context.requestInfo.latency 2025-06-01T10:25:12.109Z,ERROR,order-service,ORD-002,payment,1520我特意把fields列表压缩到几个关键字段结果一眼就能看出这条错误来自订单ORD-002触发点是payment操作耗时 1520ms。之前那种“眼睛盯屏幕找出错请求还要确认是哪个接口”的过程到这里就结束了。如果你不指定--fields工具会输出所有字段。但那样 CSV 的列会非常宽建议实际使用时按需指定。4.3 结果解读与后续加工拿到 CSV 之后很多人会直接扔进 Excel 筛选。这里我多说一个进阶操作如果你本地装了 pandas可以继续做聚合分析import pandas as pd df pd.read_csv(result.csv) error_by_service df.groupby(service).size() avg_latency_by_method df.groupby(context.requestInfo.method)[context.requestInfo.latency].mean()比如你发现某类接口平均耗时明显偏高就可以倒回去看这个接口相关的日志。这个流程打通后你的排查速度会有明显提升——从“打开日志文件到处翻”变成“先跑一次提取脚本再对着表格看趋势”。我还建议把提取结果按时间排序后输出。日志本身不保证有序但 CSV 默认是按读取顺序来的所以最好在输出前按timestamp排序。代价是需要把全部记录载入内存对超大文件来说不一定划算。我的做法是提供一个--sort-by timestamp参数只在用户明确需要排序时才加载全量数据。5. 常见问题与排雷实录5.1 高频坑位速查表现象原因解决办法脚本跑着跑着内存飙升读取时把整行 JSON 加载后又缓存了全量记录检查代码里是否在循环外追加记录输出前不要无条件collect_all_fieldsCSV 里数字变成199.9却无法求和因为json.loads把数字解析成 float/int写 CSV 时被转成字符串这是 CSV 格式限制导入 Excel 后手动转数字或用 pandas 读取同名字段被后一条记录覆盖展平函数遇到相同点路径必然覆盖确认是否需要保留数组下标必要时给每行加自增序号某些行没有输出原始行不是合法 JSON被json.JSONDecodeError吞掉了加--debug模式打印被跳过的行号人工判断是否重要Windows 下 CSV 打开乱码UTF-8 编码不被 Excel 直接识别输出时加 BOM或指定--encoding gbk过滤条件匹配不上字段实际路径是context.userId你写成了context.user_id先不加过滤跑一次把所有字段名列出来逐一确认特别说一下Windows 乱码这个坑。同事用你脚本导出 CSV 在 Excel 里打开全是乱码不是数据错了是编码标志问题。解决办法很简单if output_format csv and output_encoding.lower() utf-8: csv_file.write(\ufeff) # 写入 UTF-8 BOM或者干脆命令行加--encoding gbk参数Excel 对 GBK 的支持一向完美。一个小细节能救你一整个下午。5.2 我自己踩过的三个坑第一个坑是递归展平遇到“循环引用”。JSON 本身不允许循环引用但你解析的日志数据如果是从某个接口拿来的偶尔会有数据结构里带了引用标记之类的东西导致isinstance(data, dict)永远走不到出口。我后来给递归函数加了一个深度计数器超过 10 层就强制把当前值转成字符串返回。虽然会丢一点嵌套信息但至少脚本不会死循环。第二个坑是时间字段的时区问题。很多日志的timestamp是带Z后缀的 UTC 时间导进 Excel 后没法直接和本地时间对比。我养成的习惯是--fix-timezone参数默认转成本地时区再输出from datetime import datetime, timezone, timedelta def to_local_time(ts_str): dt datetime.fromisoformat(ts_str.replace(Z, 00:00)) return dt.astimezone(timezone(timedelta(hours8))).isoformat()这个逻辑很简单但排查问题时特别救命。比如你按“下午三点”搜索结果日志时间全是“07:00”就会浪费很多时间。第三个坑是“多行日志合并”。有的日志框架会把堆栈信息换行输出导致一条日志的 JSON 主体被拆成两行。按行解析时第二条堆栈行不是合法 JSON直接被跳过。如果你需要保留完整堆栈就得做“按大括号计数法”的合并。我的做法是当一行内容里{和}数量不相等时继续读取下一行拼接直到括号数平衡。def iter_complete_json_lines(file_handle): buffer for line in file_handle: buffer line if buffer.count({) - buffer.count(}) 0: yield buffer.strip() buffer 注意这个方法对字符串里包含大括号的日志会误判但对绝大多数日志文本是可用的。如果你实在不放心就多写一个“状态机”严格处理引号那就是另一个项目了。5.3 让工具更好用的几个小扩展如果这个脚本你想日常一直用我建议再加三个功能。一是“字段名自动映射”。不同服务的日志经常用user_id和userId表示同一个东西你可以在脚本里写一张别名表展平之后再统一改名这样下游分析代码就不用跟着改了。二是“JSON 路径通配符”。处理数组时context.items[*].price这种写法很实用。实现方式是在展平函数里对列表下标改成*标记匹配时用正则或拆段遍历。代价是脚本复杂度上升不少但对数据清洗场景收益很大。三是“输出摘要统计”。在命令行里加一个--summary跑完直接打印一个简单表格每个level有多少条、每个service有多少条。不用打开 Excel 就能对数据量心里有数。我一般会把它默认打开省得每次输出文件还要自己去数行数。根据我个人经验这个日志提取工具最值得投入时间的地方不在解析本身而在“过滤条件的灵活度”和“输出字段的可控性”。把这两点做好脚本就能从一次性工具升级成团队的通用排查入口。如果你后续遇到类似问题不妨先把自己最常用的过滤场景写清楚再动手实现比一开始就追求全功能高效得多。最后再分享一个小技巧把提取后的 CSV 文件统一按日期归档比如logs_extract/2025-06-01/result.csv这样半年后想复盘某天的故障直接按目录找不用翻命令历史。
