Python将JSON转换为CSV:数据格式转换脚本
为什么一个”简单的格式转换”会写成工程问题?绝大多数技术文档把JSON转CSV讲成三步:json.loads()解析、提取字段、csv.writer()写出。逻辑没错,但只对结构干净、字段固定、编码统一的JSON成立。
实际业务中的JSON来自批量数据采集,每一条记录的嵌套深度、字段完整度、编码方式都可能不同。一个商品列表接口返回的JSON,第一页是{"price": 29.9},第二页变成{"price": {"current": 29.9, "original": 39.9}},第三页直接把price字段丢了。脚本要兜住这三种情况,复杂度不是线性增长,是指数级的。
技术团队的第一反应通常是”换更高级的库”或”写更精巧的解析逻辑”。但真正的工程卡点不在Python写法,而在上游数据结构本身。下面逐个展开三个卡点。
JSON结构不可控的三个卡点,根源在哪?
卡点
表现
根源
嵌套深度不定
同一字段在不同记录中嵌套层级不同,flatten逻辑被迫递归
数据源返回的JSON格式在分页或不同时段有差异
字段随机缺失
部分记录缺少某些字段,CSV列对齐失败
数据源在部分条件下省略可选字段,采集层未做默认值填充
编码混杂
UTF-8中混入GBK或Latin-1字符,解析报错或乱码
数据源返回的Content-Type与实际编码不一致,采集层未做编码归一化
三个卡点的共性:问题发生在转换阶段,但根源在采集阶段。采集层拿到的JSON越干净、结构越一致,转换脚本就越简单。
这不是”Python技巧不够”,是上游数据质量不可控的代价在下游集中爆发。
Python脚本怎么写,才能兜住结构不可控的JSON?即使明确了根源在采集层,转换脚本仍然需要兜底。以下是一个可复用的框架,覆盖嵌套flatten、字段对齐、编码预处理三个环节。
环节一:递归flatten
嵌套JSON需要展平为一维键值对,键名用下划线拼接父子层级。
def flatten_json(obj, parent_key='', sep='_'):
items = {}
if isinstance(obj, dict):
for k, v in obj.items():
new_key = f"{parent_key}{sep}{k}" if parent_key else k
items.update(flatten_json(v, new_key, sep))
elif isinstance(obj, list):
for i, v in enumerate(obj):
items.update(flatten_json(v, f"{parent_key}{sep}{i}", sep))
else:
items[parent_key] = obj
return items
关键判断:list类型的展平策略要按业务决定。商品标签["电子", "手机"]展平为tags_0、tags_1是可以的;但如果list里嵌套了完整对象,展平后CSV列数会失控,这时应该把该字段序列化为JSON字符串存入单列。
环节二:字段对齐
不同记录的字段集合不一致时,需要先扫一遍所有记录收集全量字段,再用全量字段做CSV表头,缺失字段填默认值。
import csv
import json
def json_to_csv(json_list, output_path):
flat_rows = [flatten_json(record) for record in json_list]
all_keys = set()
for row in flat_rows:
all_keys.update(row.keys())
all_keys = sorted(all_keys)
with open(output_path, 'w', newline='', encoding='utf-8-sig') as f:
writer = csv.DictWriter(f, fieldnames=all_keys, extrasaction='ignore')
writer.writeheader()
for row in flat_rows:
writer.writerow({k: row.get(k, '') for k in all_keys})
注意encoding='utf-8-sig':加BOM头是为了让Excel正确识别UTF-8编码。这个细节在数据分析团队拿CSV做下游处理时经常踩坑。
环节三:编码预处理
采集返回的原始字节流,在json.loads()之前做一次编码归一化:
import chardet
def safe_decode(raw_bytes):
detected = chardet.detect(raw_bytes)
encoding = detected.get('encoding', 'utf-8')
return raw_bytes.decode(encoding, errors='replace')
errors='replace'保证即使编码检测不准,也不会因为单个非法字符阻断整个流程。但replace会丢失信息,如果业务对字段精确度要求高(比如药品数据采集、法律大数据),应该记录替换日志,人工复核。
三个环节串起来就是一个可复用的转换框架。但框架能解决的是”兜底”,不能解决”根因”。
大文件和流式处理怎么做?上面的框架把所有记录加载到内存再写CSV,对万条级别的数据够用。但企业级数据采集的JSON文件动辄几百MB甚至GB级别,全量加载会直接打爆内存。pandas加载JSON时内存占用约为原始文件大小的3-5倍,一个500MB的JSON文件需要1.5-2.5GB内存。
两条路:
方案
适用场景
Python库
内存占用
处理速度
流式解析
JSON文件是一个大数组,需要逐条处理
ijson
低(MB级)
中等
逐行读取
JSON已是JSONL格式(每行一条独立JSON)
内置readline
低(MB级)
快
流式解析示例(ijson):
import ijson
def stream_json_to_csv(json_path, output_path):
all_keys = set()
buffer = []
with open(json_path, 'rb') as f:
for record in ijson.items(f, 'item'):
flat = flatten_json(record)
all_keys.update(flat.keys())
buffer.append(flat)
if len(buffer) >= 5000:
flush_to_csv(buffer, all_keys, output_path)
buffer.clear()
if buffer:
flush_to_csv(buffer, all_keys, output_path)
这里有一个工程取舍:流式解析时,第一批5000条记录里收集到的字段集合可能不完整,后续批次出现新字段时CSV表头需要补列。实际操作中,建议先做一次轻量级的字段扫描(只提取key不存值),拿到全量字段后再做第二遍流式写入。两遍扫描的时间成本大约增加30%-50%,但能避免CSV列错位。这个成本在生产环境里是值得的。
如果数据采集阶段可以控制输出格式,建议直接输出JSONL(每行一条独立JSON),转CSV时只需逐行json.loads() + 写入,不依赖ijson等第三方库。
采集层的结构可控性,能省多少转换成本?回到本篇的判断轴:转换脚本的复杂度,本质上是采集层返回数据结构不可控的代价。
把上面四个环节的工程量拆开看:
环节
结构可控时的工作量
结构不可控时的工作量
差异来源
递归flatten
固定一层,不需要递归
递归+list展平策略判断
嵌套深度是否稳定
字段对齐
字段固定,直接写表头
全量扫描+缺失填充
字段集合是否一致
编码预处理
统一UTF-8,不需要chardet
逐条检测+替换+日志
编码是否归一
大文件处理
按固定schema分块写入
两遍扫描+动态补列
schema是否可预测
我们青果网络在服务网站采集器、APP大数据分析这类场景时(来源:青果实践观测,2024-2025,样本=数百家),观察到一个反复出现的规律:当采集层的IP调度稳定(可用率99.9%、平均延迟