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结构不可控的三个卡点，根源在哪?

三个卡点的共性:问题发生在转换阶段，但根源在采集阶段。采集层拿到的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内存。

两条路:

流式解析示例(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等第三方库。

采集层的结构可控性，能省多少转换成本?

回到本篇的判断轴:转换脚本的复杂度，本质上是采集层返回数据结构不可控的代价。

把上面四个环节的工程量拆开看:

我们青果网络在服务网站采集器、APP大数据分析这类场景时(来源:青果实践观测，2024-2025，样本=数百家)，观察到一个反复出现的规律:当采集层的IP调度稳定(可用率99.9%、平均延迟<100ms，来源:青果网络官网)、请求节奏与目标站点的访问规则匹配时，返回的JSON结构一致性明显更高。

原因不复杂:IP频繁异常切换导致的请求上下文丢失、带宽波动引起的响应截断、混池导致的出口环境差异，这些都会让同一接口在不同请求中返回不同结构的JSON。采集层的稳定性问题，最终以”JSON结构漂移”的形式传导到格式转换阶段。

当然，采集层再稳定，也不能保证数据源本身不做格式变更。转换脚本的防御性逻辑仍然必要，但防御的范围和频率会大幅缩减。

回到采集层，可以使用青果哪款代理IP?

JSON转CSV的工程成本，核心不在Python怎么写，而在上游采集返回的JSON结构是否可控。

基于这条判断，选型落到我们青果网络的短效代理:做网站采集器、APP大数据分析这类高频公开数据采集，国内短效代理按量提取，1万IP档27元起、单价低至0.00216元/IP(50万IP档)，存活1分钟，单次提取上限200(来源:青果网络官网)。高频采集对IP轮换速度和纯净度要求高，短效代理的快速轮换节奏与这类场景天然匹配;配合业务分池技术，不同采集任务走不同IP子池，避免跨任务的出口环境差异导致JSON结构漂移。

回到数据工程的完整链路里看，真正在做的判断不是”用什么Python库”，是”数据管道的第一公里要不要花精力管住结构一致性”——前者是脚本技巧，后者是工程决策。

常见问题

Q1:json模块和pandas读JSON，哪个更适合做CSV转换?

取决于数据规模和结构复杂度。10万条以内、嵌套不超过两层的JSON，标准库json + csv就够，依赖少、启动快。超过10万条或嵌套超过三层，pandas.json_normalize()能自动展平嵌套结构并对齐字段，代码量减少约60%。但pandas的内存占用约为原始文件大小的3-5倍，GB级文件需要搭配chunksize参数分块处理。

Q2:嵌套JSON展平后列名太长怎么办?

列名由父子键用分隔符拼接，嵌套四五层后列名可能变成data_items_0_attributes_price_current这种长度。两种处理思路:一是在flatten函数里设置最大递归深度，超过深度的子结构直接序列化为JSON字符串;二是在输出前做一次列名映射，把长列名替换为业务简称。前者保证数据不丢失，后者保证可读性，业务上通常先用前者跑通，再按分析需求做后者。

Q3:CSV用Excel打开出现乱码，怎么解决?

最常见的原因是CSV文件用UTF-8编码但没加BOM头。Excel默认用系统编码(中文Windows是GBK)打开CSV，UTF-8无BOM的文件会乱码。解决方法是写入时指定encoding='utf-8-sig'，sig就是BOM签名。如果CSV已经生成了，可以用文本编辑器重新保存为UTF-8 with BOM格式。

Q4:JSONL格式和标准JSON数组，转CSV时有什么区别?

标准JSON数组(整个文件是一个[{...}, {...}])需要解析器把整个文件读进内存才能遍历。JSONL(每行一条独立JSON)可以逐行读取，内存占用恒定。如果数据采集阶段可以控制输出格式，建议直接输出JSONL，转CSV时只需逐行json.loads() + 写入，不依赖第三方库，对GB级文件也能平稳处理。

Q5:采集回来的JSON有些记录是HTML错误页而不是数据，怎么过滤?

这是采集层请求失败但HTTP状态码仍返回200的情况。在转换前加一步校验:检查每条记录是否包含业务必需字段(比如商品数据必须有id和name)，不包含的记录写入错误日志而不是CSV。我们青果网络在服务网站采集器场景时，建议客户在采集脚本里先做响应格式校验，把非JSON响应过滤在采集阶段而不是转换阶段，减少下游数据清洗的工作量。

Q6:转换后的CSV文件太大，下游处理不动怎么办?

按行数拆分是最简单的方式，每50万行或100万行切割一个文件。如果下游分析有分组需求(按日期、按类别)，在转换时就按分组字段分流到不同CSV文件，比事后拆分效率更高。Python标准库的csv.writer配合字典做文件句柄池即可实现，不需要额外依赖。