如何用Python将百万级结构化嵌套JSON解析为DataFrame/CSV？

核心思路 #

嵌套JSON的关键是扁平化嵌套结构：将层级字段（如SUMMARY）拆为独立列，将数组字段（如FEATURES）展开为多行（长表）或多列（宽表）。针对百万级数据，需优先考虑内存效率，避免一次性加载全量数据。

步骤1：小样本测试（验证逻辑） #

先基于你的样本数据编写扁平化函数，确保解析逻辑正确：

导入依赖库 #

import pandas as pd
import json

编写扁平化函数 #

这个函数将单条JSON记录转换为多条结构化行（对应每个特征）：

def flatten_record(record):
    crd = record['CRD']
    # 提取顶层公共字段
    base_data = {
        'FG': crd['FG'],
        'ZPN': crd['ZPN'],
        'MATCH': crd['MATCH'],
        'COUNT': crd['COUNT'],
        'SUMMARY_ID': crd['SUMMARY']['ID'],
        'SUMMARY_PATHID': crd['SUMMARY']['PATHID']
    }
    
    feature_rows = []
    # 展开FEATURES数组，每个特征生成一行
    for feature in crd['DETAILS']['PARADATA']['FEATURES']:
        row = base_data.copy()
        # 提取特征基础字段
        row.update({
            'FEATURENAME': feature['FEATURENAME'],
            'FEATUREVALUE': feature.get('FEATUREVALUE'),
            'FEATUREUNIT': feature.get('FEATUREUNIT'),
            'FEATUREID': feature.get('FEATUREID')
        })
        # 展开FEATUREVALUEDETAILS（如果存在）
        if 'FEATUREVALUEDETAILS' in feature:
            fvd = feature['FEATUREVALUEDETAILS']
            row.update({
                'FVD_VALUE': fvd.get('VALUE'),
                'FVD_SIGN': fvd.get('SIGN'),
                'FVD_UNIT': fvd.get('UNIT'),
                'FVD_MULTIPLIER': fvd.get('MULTIPLIER'),
                'FVD_MULTIPLIERVALUE': fvd.get('MULTIPLIERVALUE')
            })
        feature_rows.append(row)
    return feature_rows

测试样本数据 #

# 你的样本JSON字符串
sample_json = '''{ 
    "CRD" : {
        "FG" : "ZVX", 
        "ZPN" : "04W05BA2A", 
        "MATCH" : "exact", 
        "COUNT" : 4, 
        "SUMMARY" : {
            "ID" : "33772", 
            "PATHID" : "10417"
        }, 
        "DETAILS" : {
            "PARADATA" : {
                "FEATURES" : [
                    {
                        "FEATURENAME" : "Laptop Value", 
                        "FEATUREVALUE" : "0.9 F", 
                        "FEATUREUNIT" : "", 
                        "FEATUREID" : "22", 
                        "FEATUREVALUEDETAILS" : {
                            "VALUE" : "0.8", 
                            "SIGN" : "", 
                            "UNIT" : "F", 
                            "MULTIPLIER" : "p", 
                            "MULTIPLIERVALUE" : "9.0E-12"
                        }
                    }, 
                    {
                        "FEATURENAME" : "Product weight", 
                        "FEATUREVALUE" : "", 
                        "FEATUREUNIT" : "mm", 
                        "FEATUREID" : "1372"
                    }, 
                    {
                        "FEATURENAME" : "Variable", 
                        "FEATUREVALUE" : "Fixed", 
                        "FEATUREUNIT" : "", 
                        "FEATUREID" : "138", 
                        "FEATUREVALUEDETAILS" : {
                            "VALUE" : "Fixed", 
                            "SIGN" : "", 
                            "UNIT" : "", 
                            "MULTIPLIER" : "", 
                            "MULTIPLIERVALUE" : "1.0"
                        }
                    }
                ]
            }
        }
    }
}'''

# 解析并转换为DataFrame
data = json.loads(sample_json)
flattened_rows = flatten_record(data)
df = pd.DataFrame(flattened_rows)
print(df.head())

转换为宽表（可选） #

如果需要将特征作为列（比如每个特征对应一列），可以用pivot转换：

wide_df = df.pivot(
    index=['FG', 'ZPN', 'MATCH', 'COUNT', 'SUMMARY_ID', 'SUMMARY_PATHID'],
    columns='FEATURENAME',
    values=['FEATUREVALUE', 'FEATUREUNIT', 'FEATUREID', 'FVD_VALUE']
).reset_index()

# 重命名多层列名
wide_df.columns = ['_'.join(col).strip() if col[0] else col[1] for col in wide_df.columns.values]
print(wide_df.head())

步骤2：百万级大文件处理（内存友好） #

直接加载百万级JSON会占用大量内存，用ijson逐记录解析并分块保存：

安装ijson #

pip install ijson

批量处理代码 #

import ijson

def process_large_json(input_path, output_csv):
    batch_size = 10000  # 每处理1万条保存一次
    all_rows = []
    
    with open(input_path, 'r') as f:
        # 注意：根据你的JSON结构调整ijson路径
        # 示例：如果JSON是顶级数组 [{"CRD":...}, {"CRD":...}]，用'item'
        # 如果是嵌套结构 {"data": [{"CRD":...}, ...]}，用'data.item'
        for record in ijson.items(f, 'item'):
            flattened = flatten_record(record)
            all_rows.extend(flattened)
            
            # 分块保存
            if len(all_rows) >= batch_size:
                temp_df = pd.DataFrame(all_rows)
                # 第一次写入加表头，后续追加不加
                temp_df.to_csv(output_csv, mode='a', header=not pd.io.common.file_exists(output_csv), index=False)
                all_rows = []
    
    # 处理剩余数据
    if all_rows:
        temp_df = pd.DataFrame(all_rows)
        temp_df.to_csv(output_csv, mode='a', header=not pd.io.common.file_exists(output_csv), index=False)
    
    # 读取最终CSV
    final_df = pd.read_csv(output_csv)
    return final_df

# 调用示例
# final_df = process_large_json('your_large_data.json', 'output.csv')

关键注意事项 #

• JSON路径调整：根据你的实际JSON结构修改ijson.items的路径参数，确保能正确遍历每条记录。
• 数据类型优化：解析完成后统一转换字段类型，比如df['COUNT'] = df['COUNT'].astype(int)，减少内存占用。
• 长表vs宽表：百万级数据优先选择长表，宽表会因列数过多导致内存压力陡增。
• 分块大小：根据你的内存配置调整batch_size，内存小就调小数值。

内容的提问来源于stack exchange，提问作者Shiv948