用 Python 标准库解析 JSON 日志的五个实用片段

举报
yd_227985793 发表于 2026/10/26 06:23:00 2026/10/26
【摘要】 服务日志是 JSON Lines 格式时,json、collections、datetime 三个标准库模块就能完成解析、容错、统计和输出。本文给出五个可直接复用的片段:流式逐行解析、坏行容错、defaultdict 分组统计、时间分桶以及写回文件时的 ensure_ascii 细节。

一、为什么从标准库开始

很多服务的日志已经是 JSON Lines 格式(每行一个 JSON 对象)。手头没有 ELK 这类重型工具时,Python 标准库就够用了:json、collections、datetime 三个模块,就能完成解析、统计、时间分桶和格式化输出,零依赖、放到任何机器上都能跑。

下面是我在日常运维里反复用到的五个片段,假设日志长这样:

{"ts": "2026-10-26T09:15:32", "level": "ERROR", "svc": "gateway", "code": 502, "msg": "upstream timeout"}

二、五个片段

1. 流式逐行解析,别一次性读入内存

import json

with open("app.log", encoding="utf-8") as f:
    for line in f:
        rec = json.loads(line)

for line in f 是惰性迭代,几个 GB 的日志也不会把内存打爆;json.loads 只处理单行,天然匹配 JSON Lines 格式。

2. 坏行容错:跳过并记下行号

线上日志总混着截断行和乱码,解析失败直接抛异常会中断整个任务,加一层 try/except 更稳:

import json

bad = []
with open("app.log", encoding="utf-8", errors="replace") as f:
    for i, line in enumerate(f, 1):
        try:
            rec = json.loads(line)
        except json.JSONDecodeError:
            bad.append(i)
            continue

errors="replace" 防止单个非法字节让整个文件读不下去;跑完看一眼 bad 的数量,还能顺带评估日志采集质量。

3. 用 defaultdict 做分组统计

想知道每个服务的错误分布,不用 Pandas:

import json
from collections import defaultdict

by_svc = defaultdict(int)
with open("app.log", encoding="utf-8") as f:
    for line in f:
        rec = json.loads(line)
        if rec.get("level") == "ERROR":
            by_svc[rec.get("svc", "unknown")] += 1

for svc, n in sorted(by_svc.items(), key=lambda x: -x[1]):
    print(f"{svc:12} {n}")

defaultdict(int) 省掉了"键不存在先置零"的判断;rec.get("svc", "unknown") 兜住缺字段的情况,统计脚本不会因为一行缺字段的日志崩掉。

4. 时间分桶:fromisoformat 加小时截断

排查"故障集中出现在哪个时段"时,把时间戳截断到小时再计数:

import json
from collections import defaultdict
from datetime import datetime

per_hour = defaultdict(int)
with open("app.log", encoding="utf-8") as f:
    for line in f:
        rec = json.loads(line)
        ts = datetime.fromisoformat(rec["ts"])
        bucket = ts.replace(minute=0, second=0, microsecond=0)
        per_hour[bucket] += 1

ISO 8601 格式的时间字符串(如 2026-10-26T09:15:32)用 datetime.fromisoformat 直接解析,不必再用 strptime 写格式串,代码短一截,也不容易写错。

5. 结果写回文件:注意 ensure_ascii

import json

result = {"total_errors": 137, "by_svc": dict(by_svc)}
with open("report.json", "w", encoding="utf-8") as f:
    json.dump(result, f, ensure_ascii=False, indent=2)

ensure_ascii=False 是关键:不加的话中文会全部变成 \uXXXX 转义,肉眼没法读。再加 indent=2,输出方便 diff 和 review。

三、小结

这五个片段可以自由组合:流式读 + 容错 + defaultdict 统计,就是一个 20 行左右的日志巡检脚本。日志量大、维度多的时候再上专业工具也不迟。标准库最大的好处是"到处都能跑",写一次可以用在开发机、跳板机和容器里,不需要装任何包。

文中代码在 Python 3.10 环境验证通过,均只使用标准库。

【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0)

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。