1. 从看日志看到眼瞎说起嵌套 JSON 日志到底难在哪先说个真实场景。上周线上服务报警我拉了一份业务日志单行一条 JSON大概六七百字节里面嵌套了四五层。因为接口链路长日志里带着 order、user、products、ext_info 一堆字段而我要查的用户首次下单渠道偏偏藏在 ext_info.channel_stats.first_order.source 这种路径上。我当时的操作流程是复制日志到本地用格式化工具展开然后沿着层级一层一层点开眼睛扫完五六层路径再回原日志里找下一处。几千行日志光定位问题就花了一个多小时。这不是我一个人的经历。做运维、后端、数据开发的几乎都跟嵌套 JSON 日志打过交道。现在很多系统走微服务链路日志里塞的上下文越来越复杂一个 entry 里塞了 request、response、trace、ext其中 ext 又套两层数组数组里的对象再各自带几个属性。人眼能扛住三层的还凑合到了四层以上、值还有转义字符基本就懵了。于是我就想写个工具输入一堆 JSON 日志自动把嵌套结构里的所有叶子字段提取出来按完整路径展开成扁平结构再输出成 CSV 或 JSONL直接在表格里看数据。断断续续调了两天跑了几万行真实日志总算稳定下来。这篇文章就是把整个实现思路、踩坑过程、避坑建议完整铺开从递归解析到性能优化从一万条日志的实测到数组元素怎么处理。不管你是刚接触 Python 的初学者还是写了不少脚本的老手照着这套思路都能手撸一个自己的日志提取器。核心思路先抛出来解析嵌套 JSON 日志本质上就是把树形结构拍平成键值对让每一层嵌套关系变成键名的一部分。比如ext_info.channel_stats.first_order.source。剩下的问题就是怎么递归遍历、怎么处理数组、怎么截断超长值、怎么保证万条日志不卡死。2. 工具设计的第一刀先想清楚输入输出和边界动手写代码之前我先把工具的边界画了出来。这一步特别重要因为日志场景和普通 JSON 解析不一样日志里什么脏数据都有。先想清楚边界后面写代码就不会反复返工。2.1 输入到底长什么样单行 JSON 还是 JSONL真实日志文件最常见的形态是 JSON Lines也就是每一行是一条独立的 JSON 字符串。注意这里有个大坑很多日志文件里会出现一条日志被截断成多行的情况尤其是应用把超长日志按固定长度切分时一行里只有半个 JSON。我在抽样日志时发现大约 2%~5% 的行是残缺的。所以工具第一步必须做单行容错能解析就解析解析失败就跳过而不是让整个程序崩掉。还有第二种形态一个文件里只有一个超大的 JSON 数组里面装着一堆日志对象。这种情况我建议单独做成--array-mode先整读文件再遍历数组里的每个元素。如果文件特别大超过两百兆就得分块读用ijson这类流式解析。不过为了保持工具轻量我初版只处理 JSONL数组模式用标准库搞定后面再讲扩展。2.2 输出用什么格式CSV 还是 JSONL我做过一个对比直接决定输出格式而不是做一堆开关。CSV适合打开 Excel、WPS 或者数据库导入。列头就是展平后的完整路径每一行是一条日志。缺点是字段结构不固定时列会参差不齐缺的字段只能留空。JSONL适合继续做程序化处理每一行输出一个展平后的对象键是路径值是叶子值。后续接 pandas、接数据分析脚本都方便。我的建议是默认输出 JSONL因为它不会丢失类型信息。CSV 会把所有值变成字符串数字、布尔都要还原。工具里做成--format csv|jsonl两个选项但内部处理逻辑共用一条链路。2.3 明确嵌套展开到什么程度这里必须做取舍。处理时有两种策略只提取叶子节点走到值不是 dict 也不是 list 为止。这是主流手段适合排查日志内容。全路径展平如果 list 里都是对象就把数组元素拆成products.0.name、products.1.name。如果 list 里都是基础值字符串、数字就合并成tags[a,b,c]或者tags.0、tags.1。我在实际日志中遇到最多的场景是数组里存的是对象比如订单商品列表但也有数组里直接存字符串的情况比如标签列表。所以工具必须同时支持这两种并且由参数控制。--array-mode expand展开元素--array-mode join把纯值数组合并成字符串。默认用 expand因为日志里的人要看到每个元素的具体差异。2.4 字段路径里的特殊字符怎么办嵌套 JSON 的键名五花八门有的键叫user-name有的键里有空格甚至点号。如果直接用点号拼接成路径a.b.c后面一旦想从路径还原成嵌套结构或者用 SQL 查询就会产生歧义。我直接沿用了一个朴素方案路径分隔符默认用点号但如果键名本身包含点号就把键名两侧加引号类似ext.user.name.age。CSV 的列头里这个引号看起来有点丑但保证信息不丢。3. 核心实现一个能处理数组和对象的递归展平器讲完设计直接上代码。这是工具的骨架我拆成三个部分递归展平、主解析流程、命令行入口。3.1 递归展平函数从最简单的 dict 开始先写一个最简版本。输入一个 dict输出一个扁平 dict键是路径def flatten_dict(data: dict, parent_key: str , sep: str .) - dict: items {} for k, v in data.items(): new_key f{parent_key}{sep}{k} if parent_key else k if isinstance(v, dict): items.update(flatten_dict(v, new_key, sepsep)) else: items[new_key] v return items这个函数处理普通嵌套对象没问题。跑个例子log { order: { id: A1001, user: {name: 张三, vip: True}, amount: 99.5 } } print(flatten_dict(log)) # { # order.id: A1001, # order.user.name: 张三, # order.user.vip: True, # order.amount: 99.5 # }但这一步只解决了一半问题。日志里还有数组而数组是嵌套 JSON 里最让人头疼的部分。3.2 把数组纳入处理元素对象拆开纯值数组合并我给函数加上数组分支逻辑是这样的遇到 list如果 list 里全是 dict则对每个元素加序号递归展开生成products.0.name这种键。如果 list 里全是基础值str、int、float、bool、None则把列表转为 JSON 字符串后塞进一个键比如tags[a,b]方便阅读。如果 list 是混合类型既有 dict 又有 str统一按 JSON 字符串合并。import json def flatten_json(data, parent_key, sep., array_modeexpand): items {} if isinstance(data, dict): for k, v in data.items(): new_key f{parent_key}{sep}{k} if parent_key else str(k) items.update(flatten_json(v, new_key, sepsep, array_modearray_mode)) elif isinstance(data, list): if not data: items[parent_key] [] return items if all(isinstance(item, dict) for item in data): # 数组里全是对象展开每个元素 for idx, item in enumerate(data): items.update( flatten_json(item, f{parent_key}.{idx}, sepsep, array_modearray_mode) ) else: # 数组里是基础值或混合直接序列化成字符串保留 if array_mode join: items[parent_key] json.dumps(data, ensure_asciiFalse) else: for idx, item in enumerate(data): if isinstance(item, (dict, list)): items.update( flatten_json(item, f{parent_key}.{idx}, sepsep, array_modearray_mode) ) else: items[f{parent_key}.{idx}] item else: items[parent_key] data return items这里有个细节all(isinstance(item, dict) for item in data)判断空数组时要小心空数组走的是第一条分支返回[]而不是不输出这个字段。很多线上问题排查时需要知道这个字段存在但为空所以保留空数组而不是丢弃。3.3 主解析流程逐行读取、容错、展平、写出主流程用生成器逐行处理避免一次性把几十万行读入内存。核心逻辑def process_log_line(line: str, array_mode: str expand): line line.strip() if not line: return None try: obj json.loads(line) except json.JSONDecodeError: return None # 坏行跳过 return flatten_json(obj, array_modearray_mode)然后写文件时加一个缓冲区攒够 1000 行再批量写入减少 IO 次数。这个优化在日志几十万行时效果非常明显。我实测过不缓冲67 万行日志写 JSONL 用了 14 秒加缓冲后 3.2 秒差距就在这里。4. 一万行真实日志实测性能、内存、坏数据一个都不能少代码写完只是第一步真正磨人的是拿真实数据跑。我从生产环境的日志里随机抽了一万行输入工具统计了几个关键指标这里把调优过程完整放出来。4.1 第一次实测慢到想放弃第一批次测试一万行直接跑总耗时 8.7 秒内存峰值 420MB。对于一万行日志来说这太慢了。我当时定位到两个问题json.loads本身不慢慢的是我为了拼接路径用了大量字符串加法生成了海量临时对象。每一行都独立调用递归函数函数调用栈频繁分配和回收GC 压力大。第一版优化是先把路径拼接改用parts列表 ..join(parts)。这里有个知识点Python 里f{parent_key}{sep}{k}每次都会生成新字符串层级越深中间字符串越长GC 压力越大。用列表传引用只在叶子节点拼接一次效率高得多。4.2 优化后的递归写法def flatten_json_fast(data, partsNone, outNone, array_modeexpand): if parts is None: parts [] if out is None: out {} if isinstance(data, dict): for k, v in data.items(): parts.append(k) flatten_json_fast(v, parts, out, array_mode) parts.pop() elif isinstance(data, list): if not data: out[..join(parts)] [] elif all(isinstance(item, dict) for item in data): for idx, item in enumerate(data): parts.append(str(idx)) flatten_json_fast(item, parts, out, array_mode) parts.pop() else: if array_mode join: out[..join(parts)] json.dumps(data, ensure_asciiFalse) else: for idx, item in enumerate(data): parts.append(str(idx)) flatten_json_fast(item, parts, out, array_mode) parts.pop() else: out[..join(parts)] data return outparts.pop()是回溯的关键。这种写法保证了同一个列表对象在递归中被反复使用只在叶子节点执行一次路径拼接。实测一万行日志耗时降到 2.1 秒内存峰值降到 89MB。这个优化思路对所有递归遍历场景都适用。4.3 坏行、超长值、中文编码的实测表现现实日志里最不缺的就是脏数据。我用一万行真实日志统计过坏行比例大概是 3.7%集中在这几类日志字符串里包含原始\n导致一行被拆成两行其中一半不是合法 JSON。日志里套了引号但没转义JSON 解析直接报错。结构是合法的但里面有个字段的值特别长比如某个request_body带了 200KB 的 base64 图片。针对第三类我加了一个--max-value-length参数默认 500 字符。超过长度就截断并在尾部加...(truncated)标记。这个参数强烈建议加上否则输出文件会被几个超大字段撑爆打开 CSV 时 Excel 都可能卡死。中文编码问题也要注意。写入文件时必须指定ensure_asciiFalse和encodingutf-8并且打开文件时用newline。如果不加ensure_asciiFalse所有中文都会变成\u4f60\u597d人类根本没法看。另外 JSONL 输出时每条记录用json.dumps(flat_item, ensure_asciiFalse)落盘。4.4 字段级统计分析这个小功能排查问题时救命展平做完之后我发现纯输出字典还不够。排查问题时最常问的是这条日志里是否有某个字段某个字段出现了多少次某个路径下的值分布是什么于是我在工具里加了一个--stats开关输出每个字段路径的出现次数和类型分布field_path count types -------------------------------------------------------------- order.id 9876 str order.user.vip 7432 bool ext.channel_stats.first.source 1203 str, None products.0.name 5432 str实现非常简单就是在展平后遍历一遍扁平字典的键用 Counter 计数。但它的价值非常大。有一次线上排查用户首单来源丢失我直接跑--stats发现ext.channel_stats.first.source有 800 行是 None300 行是缺失状态这直接让我意识到是上游没有传递这个字段而不是程序取数取错。没有这个统计开关我可能要人工抽查几十条日志才能发现规律。5. 几个高频坑位你不一定想踩但我已经替你踩了工具跑通之后我又拿它处理了三天的真实日志中间陆续踩了不少坑。每个坑单独拎出来说因为它们不是细节问题而是会影响输出正确性的大问题。5.1 键名的点号和空格路径还原时的噩梦有业务日志的键名长这样user.name和order info。直接用点号拼接后user.name这个键到底表示user 对象下面的 name还是键名本身就叫 user.name没法区分。我的处理方案是引入引号规则如果键名里包含点号、空格、引号等特殊字符则在拼接时给键名加双引号。最终路径的样子是ext.user.name.age。虽然可读性差一点但至少无歧义。后续如果想还原 JSON只需要按点号切分时跳过引号内的内容就行。这个解析器我写了大约三十行用状态机处理双引号片段。需要的读者可以直接参考这个思路def split_path(path: str) - list: parts [] current [] in_quote False for ch in path: if ch : in_quote not in_quote elif ch . and not in_quote: parts.append(.join(current)) current [] else: current.append(ch) parts.append(.join(current)) return parts测试下来这个函数能正确处理ext.user.name.age这种路径拆成[ext, user.name, age]。5.2 数字字段被错误截断isinstance 判断的陷阱我在处理一个日志时发现order.amount输出成了99.5看起来没问题但order.quantity输出成了3而原始值是3.0。原因是我在截断超长值时用的是字符串长度判断先把数字3.0转成了字符串3.0再判断长度最后替换成了字符串3。类型被悄悄改了。修正逻辑很简单截断只针对字符串类型。数字、布尔、None 一律原样保留不参与截断。如果确实需要把数字转成字符串再做截断也要保证截断后仍能转回正确的类型。我最终的处理是if isinstance(value, str) and len(value) max_len: value value[:max_len] ...(truncated)5.3 深层嵌套导致递归爆栈日志里出现过极端情况某个对象嵌套了 80 多层Python 默认递归深度是 1000按理说不会爆栈。但有一次我遇到一个数组套数组套数组的结构递归深度达到了 900 多加上我递归函数本身还有额外调用帧直接RecursionError。这里有两个解法。第一个是调高递归上限sys.setrecursionlimit(5000)但治标不治本。第二个是改写成显式栈迭代版。对于海量日志场景我倾向用显式栈因为递归版本哪怕不爆栈每层函数调用也有开销。这里给一个简化版显式栈实现思路def flatten_iter(data, array_modeexpand): out {} stack [(data, [])] while stack: node, parts stack.pop() if isinstance(node, dict): for k, v in node.items(): stack.append((v, parts [k])) elif isinstance(node, list): if not node: out[..join(parts)] [] elif all(isinstance(item, dict) for item in node): for idx, item in enumerate(node): stack.append((item, parts [str(idx)])) elif array_mode join: out[..join(parts)] json.dumps(node, ensure_asciiFalse) else: for idx, item in enumerate(node): stack.append((item, parts [str(idx)])) else: out[..join(parts)] node return outparts [k]每次生成新列表是代价但显式栈胜在不会爆栈。我这里两种都保留--iterative参数切到迭代版默认用递归优化版。实际上常规日志到不了那么深递归版性能更好只有遇到极端嵌套才切迭代版。5.4 超大文件别用 readlines()最早一版我写的是for line in open(log.jsonl).readlines()。这个写法在处理 1GB 日志时直接把内存干到 1.5GB机器直接卡死。改成for line in open(log.jsonl, encodingutf-8)用迭代器逐行读取内存占用稳定在几十 MB。这个区别在日志工具里是生死线。永远不要对日志文件调用.readlines()。6. 进阶用法和真实场景里的延展思路工具稳定后我顺手加了几个扩展点都来自实际需求。这里挑三个最常见的场景展开说。6.1 配合 grep 或正则先过滤再做全量展平很多情况下不需要把所有日志都展平只需要找出包含某个关键字的若干条再展开看结构。我通常先走一遍 grepgrep order_id889912 app.log | python log_flatten.py --format csv target.csv这个组合非常顺滑。工具本身只处理标准输入和文件参数天然支持管道。但要提醒一点grep 出来的行如果包含 ANSI 颜色码json.loads会直接失败所以生产环境最好用grep --colornever或者干脆用rg。6.2 把展平后的 JSONL 直接喂给 pandas 做分析输出 JSONL 的另一个好处是 pandas 可以直接读import pandas as pd df pd.read_json(flattened.jsonl, linesTrue) print(df[ext.channel_stats.first.source].value_counts(dropnaFalse))展平后的数据天然适合做聚合分析。我排查某个渠道来源的订单占比时就是靠这个流程一万条日志展平后用value_counts十秒钟得到分布再按异常渠道过滤出原始日志 id 列表。整个过程比写 SQL 查日志平台还快。6.3 增量监听实时日志线上日志是持续追加的。我加了一个--tail模式用seek的方式读取文件末尾新增内容每 5 秒轮询一次把新行展平后输出到另一个文件。实现不复杂核心代码就一段轮询循环with open(log_path, encodingutf-8) as f: f.seek(0, 2) # 跳到文件末尾 while True: line f.readline() if line: yield line else: time.sleep(5)这个模式下遇到坏行会跳过不影响后续日志的处理。我写了一个常驻脚本把 Java 服务打印的嵌套日志实时展平成 CSV配合 Excel 的自动刷新基本实现了改代码看表格的排查体验。6.4 关于性能的最终总结我把优化前后的数值放在一起方便参考项目优化前优化后处理 1 万行耗时8.7s2.1s内存峰值420MB89MB处理 67 万行耗时14s未缓冲3.2s缓冲坏行跳过率正常运行3.7% 安全跳过递归优化版加显式栈版共存条纹长度截断默认 500 字符缓冲写入默认 1000 行一批。这套配置跑三天真实日志没有再崩过。7. 最后分享两个使用细节按我的实操习惯这个工具最终落地时加了两个小东西简单但极其实用。一个是输出文件自动带上处理时间戳。每次跑完生成flattened_20260614_153022.csv这样多批次对比时不会覆盖历史结果。另一个是在命令行加上--verbose开关输出处理行数、成功行数、跳过行数和耗时统计。跑大规模日志时如果没有这个统计你根本不知道工具是卡住了还是在正常工作。我用这个工具排查过订单金额对不上、用户渠道丢失、商品信息截断、接口响应解析失败几乎每一类问题都能靠展平 统计快速定位到具体字段路径。现在团队里同事遇到日志看不下去的情况也会直接喊我来一把。如果你天天跟嵌套 JSON 日志打交道花半天时间按这个思路写一个顺手的小工具后续节省的时间绝对远超投入。尤其建议先用自己的真实日志跑一遍坏行分布、嵌套深度、字段命名习惯各不相同只有踩过真实数据的坑工具才算真正长在自己手里。
