Python 解析 JSON 日志：从一行数据到一份报告
写后端的人，天天跟日志打交道。现在大部分应用的日志都是 JSON 格式——每条日志是一个 JSON 对象，方便机器解析和分析。
Python 自带的 json 模块就是干这个的：把 JSON 字符串转成字典，或者把字典转成 JSON 字符串。用法就几个函数，但真正写脚本分析日志的时候，有些细节还是得注意。
核心就两个函数：loads 和 dumps
| 函数 |
干啥的 |
啥时候用 |
json.loads(s) |
JSON 字符串 → Python 字典 |
读日志行的时候用 |
json.dumps(obj) |
Python 字典 → JSON 字符串 |
写结果文件的时候用 |
JSON 字符串转字典：
import json
line = '{"url": "/api/user", "timeSpent": 45, "status": 200}'
data = json.loads(line)
print(data["url"])
print(data["timeSpent"])
字典转 JSON 字符串：
data = {"name": "Bob", "age": 25, "city": "上海"}
json_str = json.dumps(data, ensure_ascii=False)
ensure_ascii=False 这个参数经常被忽略——不加的话，中文会变成 \u4e0a\u6d77，虽然能解析，但人没法看。
日志分析实战：筛选慢请求
假设你有一份 API 日志 req_resp.log，每行是一个 JSON 对象：
{"url": "/api/user", "timeSpent": 45, "status": 200, "method": "GET"}
{"url": "/api/order", "timeSpent": 230, "status": 200, "method": "POST"}
{"url": "/api/product", "timeSpent": 89, "status": 500, "method": "GET"}
{"url": "/api/report", "timeSpent": 310, "status": 200, "method": "GET"}
{"url": "/api/search", "timeSpent": 120, "status": 200, "method": "POST"}
需求很简单：找出耗时超过 150ms 的请求，打印出来。
完整实现
import json
LOG_PATH = "req_resp.log"
with open(LOG_PATH, "r", encoding="utf-8") as f:
for line_num, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
try:
data = json.loads(line)
except json.JSONDecodeError as e:
print(f"第 {line_num} 行 JSON 解析失败: {e}")
continue
time_spent = data.get("timeSpent", 0)
if time_spent > 150:
print(f"[慢请求] {data.get('method')} {data.get('url')} 耗时 {time_spent}ms")
运行结果
[慢请求] POST /api/order 耗时 230ms
[慢请求] GET /api/report 耗时 310ms
这个脚本虽然短，但包含了几个关键点：
- 逐行读取：文件大也不怕，不会一次性加载到内存
- 跳过空行：日志文件末尾经常有空行
- 捕获
JSONDecodeError：日志里可能有格式错误的一行，不能因为一条坏数据让整个脚本崩溃
- 用
get() 而不是 []：万一某条日志缺了某个字段，get() 返回默认值，[] 直接报 KeyError
把结果写到文件里
光打印不够，通常需要把筛选结果存下来，方便后续分析或者发给别人。
import json
LOG_PATH = "req_resp.log"
OUTPUT_PATH = "slow_requests.json"
slow_requests = []
error_count = 0
with open(LOG_PATH, "r", encoding="utf-8") as f:
for line_num, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
try:
data = json.loads(line)
except json.JSONDecodeError:
error_count += 1
continue
if data.get("timeSpent", 0) > 150:
filtered = {
"url": data.get("url"),
"method": data.get("method"),
"timeSpent": data.get("timeSpent"),
"status": data.get("status")
}
slow_requests.append(filtered)
with open(OUTPUT_PATH, "w", encoding="utf-8") as out_f:
json.dump(slow_requests, out_f, ensure_ascii=False, indent=2)
print(f"慢请求数量: {len(slow_requests)}")
print(f"解析失败: {error_count} 行")
print(f"结果已保存到 {OUTPUT_PATH}")
indent=2 让输出的 JSON 有缩进，方便人看。如果文件很大，去掉 indent 能省空间。
处理嵌套 JSON
有些日志的字段是嵌套的，比如这样：
{
"request": {
"url": "/api/user",
"method": "GET",
"headers": {"User-Agent": "Mozilla/5.0"}
},
"response": {"status": 200, "timeSpent": 45}
}
访问嵌套字段用连续的 [] 或者 get()：
url = data["request"]["url"]
method = data["request"]["method"]
time_spent = data["response"]["timeSpent"]
如果某个中间字段可能不存在，用 get() 加上空字典兜底：
url = data.get("request", {}).get("url")
这种写法在数据格式不完全统一的时候很好用，一行代码就把嵌套访问和默认值都处理了。
几个经常遇到的问题
1. JSON 解析失败
json.decoder.JSONDecodeError: Expecting property name enclosed in double quotes
最常见的几个原因：
- 用了单引号而不是双引号（JSON 标准要求双引号）
- 末尾多了逗号
- 行里有非 JSON 的前缀，比如
[INFO] 2024-01-15 {"url": "/api"}
解决办法： 捕获异常，打印出问题行：
try:
data = json.loads(line)
except json.JSONDecodeError:
print(f"解析失败的行: {line[:100]}")
continue
如果日志行有固定前缀，用正则把 JSON 部分提取出来再解析。
2. 中文变成 Unicode
不加 ensure_ascii=False 的时候：
json.dumps({"name": "张三"})
加上：
json.dumps({"name": "张三"}, ensure_ascii=False)
3. 字段类型不一致
有的日志里 timeSpent 是数字，有的可能是字符串 "230"。解析的时候统一转换：
def safe_get_int(data, key, default=0):
value = data.get(key)
if value is None:
return default
try:
return int(value)
except (ValueError, TypeError):
return default
time_spent = safe_get_int(data, "timeSpent", 0)
文件读写：load 和 dump
如果整个文件就是一个 JSON 对象（不是每行一个），用 json.load() 和 json.dump()：
with open("config.json", "r", encoding="utf-8") as f:
config = json.load(f)
with open("output.json", "w", encoding="utf-8") as f:
json.dump(config, f, ensure_ascii=False, indent=2)
但日志分析通常用逐行处理（每行一个 JSON），因为文件可能很大，一次性加载会撑爆内存。