🏠 首页 攻略 JSON 数据处理进阶:15 个实用技巧处理嵌套和大型数据

JSON 数据处理进阶:15 个实用技巧处理嵌套和大型数据

JSON 是前后端数据交换的标配,但嵌套结构和海量数据常常让人头疼。本文演示 15 个进阶技巧:路径查询、扁平化、聚合、流式处理、验证、转换等,附完整 Python 和 JavaScript 代码示例。

JSON 数据看起来简单,一嵌套三层、数据量到百万级,处理起来就没那么简单了。

我处理过不少真实场景:电商订单数据嵌套了 SKU 信息、日志 JSON 字段超过 200 个、前端要解析 5MB 的 JSON 响应。今天把这些实战技巧整理出来。

1. JSONPath 查询嵌套数据

嵌套 JSON 最常用的需求就是"找到某个深层字段"。JSONPath 比逐层 .data[key][0].name 优雅得多。

Python 用 jsonpath-ng

from jsonpath_ng import parse

data = {
    "users": [
        {"name": "张三", "orders": [{"product": "笔记本", "price": 3500}]},
        {"name": "李四", "orders": [{"product": "鼠标", "price": 129}]}
    ]
}

# 查所有订单金额
jsonpath_expr = parse('$.users[*].orders[*].price')
prices = [match.value for match in jsonpath_expr.find(data)]
print(prices)  # [3500, 129]

# 查张三的所有订单
jsonpath_expr = parse('$.users[?(@.name=="张三")].orders[*].product')
products = [match.value for match in jsonpath_expr.find(data)]
print(products)  # ['笔记本']

JavaScript 可以用 jsonpath-plus 库:

import { JSONPath } from 'jsonpath-plus';

const prices = JSONPath({ path: '$.users[*].orders[*].price', json: data });
// [3500, 129]

2. 扁平化嵌套 JSON

API 返回的嵌套结构,传到数据库或表格展示时,最好先拍平。

def flatten_json(nested, prefix='', sep='_'):
    """递归扁平化嵌套 JSON"""
    out = {}
    for key, value in nested.items():
        new_key = f"{prefix}{sep}{key}" if prefix else key
        if isinstance(value, dict):
            out.update(flatten_json(value, new_key, sep))
        elif isinstance(value, list):
            out[new_key] = str(value)  # 数组转为字符串
        else:
            out[new_key] = value
    return out

nested = {
    "user": {
        "name": "张三",
        "address": {"city": "北京", "zip": "100000"}
    }
}
print(flatten_json(nested))
# {'user_name': '张三', 'user_address_city': '北京', 'user_address_zip': '100000'}

JavaScript 版本:

function flatten(obj, prefix = '') {
    return Object.keys(obj).reduce((acc, key) => {
        const newVal = obj[key];
        const newKey = prefix ? `${prefix}_${key}` : key;
        if (typeof newVal === 'object' && !Array.isArray(newVal)) {
            Object.assign(acc, flatten(newVal, newKey));
        } else {
            acc[newKey] = newVal;
        }
        return acc;
    }, {});
}

3. 使用 jq 快速查看 JSON(命令行神器)

终端里看 JSON 别用手翻。jq 一行命令搞定:

# 格式化
cat data.json | jq '.'

# 提取字段
cat data.json | jq '.users[] | .name'

# 条件过滤(找出 price > 1000 的商品)
cat data.json | jq '.products[] | select(.price > 1000)'

# 聚合统计(计算总金额)
cat data.json | jq '[.orders[].total] | add'

# 转 CSV(配合 -r 和 @csv)
cat data.json | jq -r '.users[] | [.name, .age] | @csv'

Mac 上 brew install jq,Linux 上 apt install jq,Windows 可以用 Scoop 或 WSL。

4. 大文件流式处理

50MB 的 JSON 文件,用 json.load() 会直接爆内存。用流式解析:

import ijson

# ijson 是增量解析器,边读边解析,内存占用极小
with open('large_data.json', 'rb') as f:
    for item in ijson.items(f, 'items.item'):
        # item 是数组中的每一个对象
        process(item)

JavaScript 端可以用 stream-json

const { pipeline } = require('stream');
const { Stringify } = require('stream-json/Stringify');
const { Sigil } = require('stream-json/Sigil');
const { StreamArray } = require('stream-json/streamers/StreamArray');

const fs = require('fs');
fs.createReadStream('large_data.json')
    .pipe(Sigil.parser())
    .pipe(StreamArray.withParser())
    .on('data', ({key, value}) => {
        console.log(value); // 逐条处理
    });

5. JSON Schema 数据验证

前后端交接数据,光靠文档不够,最好用 Schema 做校验。

Python 用 jsonschema

from jsonschema import validate, ValidationError

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string", "minLength": 1},
        "age": {"type": "integer", "minimum": 0, "maximum": 150},
        "email": {"type": "string", "format": "email"},
        "tags": {
            "type": "array",
            "items": {"type": "string"}
        }
    },
    "required": ["name", "email"]
}

data = {"name": "", "age": 25, "email": "test@test.com"}
try:
    validate(instance=data, schema=schema)
except ValidationError as e:
    print(f"验证失败: {e.message}")
    # 验证失败: '' is too short

6. 将 JSON 转为 CSV 下载

前端场景常见需求:把 API 返回的 JSON 数组导出为 CSV 给用户下载。

function jsonToCsv(jsonArray, filename = 'export.csv') {
    if (!jsonArray.length) return;
    
    const headers = Object.keys(jsonArray[0]);
    const csvRows = [
        headers.join(','),
        ...jsonArray.map(row => 
            headers.map(header => {
                const val = row[header] ?? '';
                // 处理含逗号或引号的值
                const escaped = String(val).replace(/"/g, '""');
                return `"${escaped}"`;
            }).join(',')
        )
    ];
    
    const blob = new Blob([csvRows.join('\n')], { type: 'text/csv;charset=utf-8;' });
    const link = document.createElement('a');
    link.href = URL.createObjectURL(blob);
    link.download = filename;
    link.click();
}

// 使用
jsonToCsv(data, '订单数据.csv');

7. JSON Diff 比对

改完接口,想看具体改了什么?比对两个 JSON:

import json

def json_diff(old, new, path=''):
    """递归比对两个 JSON 结构"""
    changes = []
    
    if type(old) != type(new):
        changes.append(f"{path or 'root'}: 类型变化 ({type(old).__name__}{type(new).__name__})")
        return changes
    
    if isinstance(old, dict):
        all_keys = set(list(old.keys()) + list(new.keys()))
        for key in all_keys:
            new_path = f"{path}.{key}" if path else key
            if key not in old:
                changes.append(f"{new_path}: + 新增字段")
            elif key not in new:
                changes.append(f"{new_path}: - 删除字段")
            else:
                changes.extend(json_diff(old[key], new[key], new_path))
    elif old != new:
        changes.append(f"{path or 'root'}: {old}{new}")
    
    return changes

old = {"a": 1, "b": {"c": 2}}
new = {"a": 1, "b": {"c": 3, "d": 4}, "e": 5}
for change in json_diff(old, new):
    print(change)
# b.c: 2 → 3
# b.d: + 新增字段
# e: + 新增字段

8. 处理日期格式统一

JSON 里的日期格式五花八门,转成统一格式再处理:

from datetime import datetime

def normalize_date(date_str):
    formats = [
        '%Y-%m-%d',
        '%Y/%m/%d %H:%M:%S',
        '%d-%m-%Y',
        '%Y-%m-%dT%H:%M:%SZ',
        '%Y-%m-%dT%H:%M:%S%z',
    ]
    for fmt in formats:
        try:
            return datetime.strptime(date_str, fmt).strftime('%Y-%m-%d %H:%M:%S')
        except ValueError:
            continue
    return date_str  # 格式都不匹配,原样返回

print(normalize_date('2026-06-12T09:00:00Z'))
# 2026-06-12 09:00:00

9. 压缩 JSON 体积

前端加载大 JSON 响应慢?后端直接 gzip 压缩:

import gzip
import json

data = {"products": [{"id": i, "name": f"商品{i}", "price": i * 10} for i in range(10000)]}
json_str = json.dumps(data)
compressed = gzip.compress(json_str.encode())

print(f"原始: {len(json_str)} 字节")
print(f"压缩: {len(compressed)} 字节")
print(f"压缩率: {len(compressed) / len(json_str) * 100:.1f}%")
# 原始: ~500KB, 压缩: ~15KB, 压缩率: ~3%

前端收到后自动解压缩(浏览器支持 gzip 响应)。

10. JSON 与 XML 互相转换

对接老系统时经常碰到 XML,Python 用 xmltodict

import xmltodict
import json

xml_str = """
<root>
    <user id="1">
        <name>张三</name>
        <age>28</age>
    </user>
</root>
"""

# XML → JSON
data = xmltodict.parse(xml_str)
print(json.dumps(data, ensure_ascii=False, indent=2))

# JSON → XML
xml_out = xmltodict.unparse(data, pretty=True)

11. 处理 null / undefined 值

前后端交互时,null 和 undefined 是常见坑:

// 安全访问嵌套字段,空值时给默认值
const city = user?.address?.city ?? '未知城市';

// 批量替换所有 null 为默认值
function cleanNulls(obj, defaultVal = '') {
    if (obj === null || obj === undefined) return defaultVal;
    if (Array.isArray(obj)) return obj.map(item => cleanNulls(item, defaultVal));
    if (typeof obj === 'object') {
        return Object.fromEntries(
            Object.entries(obj).map(([k, v]) => [k, cleanNulls(v, defaultVal)])
        );
    }
    return obj;
}

12. JSON 缓存与本地存储

SPA 应用中,缓存 JSON 响应减少重复请求:

// 简单缓存
const cache = new Map();

async function fetchWithCache(url, ttl = 300000) {
    const cached = cache.get(url);
    if (cached && Date.now() - cached.timestamp < ttl) {
        return cached.data;
    }
    
    const res = await fetch(url);
    const data = await res.json();
    cache.set(url, { data, timestamp: Date.now() });
    return data;
}

13. JSON 字段重命名

API 返回的字段名和业务层不一致,统一映射:

FIELD_MAP = {
    'user_id': 'userId',
    'first_name': 'firstName',
    'last_name': 'lastName',
    'created_at': 'createTime',
}

def rename_fields(data):
    if isinstance(data, list):
        return [rename_fields(item) for item in data]
    if isinstance(data, dict):
        return {
            FIELD_MAP.get(k, k): rename_fields(v)
            for k, v in data.items()
        }
    return data

14. 用 jq 快速生成测试数据

开发时需要造 JSON 测试数据,jqrangemap 很实用:

# 生成 10 个模拟用户
jq -n '[range(10) | {id: ., name: "用户\(.)", active: true}]'

# 输出:
# [
#   {"id": 0, "name": "用户0", "active": true},
#   {"id": 1, "name": "用户1", "active": true},
#   ...
# ]

15. JSON Lines(NDJSON)处理

日志、大数据场景常用 JSON Lines 格式,一行一个 JSON 对象:

# 读取 JSONL
with open('events.jsonl') as f:
    events = [json.loads(line) for line in f]

# 写入 JSONL(比 JSON 数组快,因为不用一次性加载全部数据)
with open('events.jsonl', 'w') as f:
    for event in events:
        f.write(json.dumps(event) + '\n')

总结

JSON 处理就这些高频场景。掌握了上面这些技巧,基本可以覆盖 90% 的日常需求。

建议收藏 jq 的命令用法——终端里看 JSON 数据,它是最快的工具,没有之一。