如何解析嵌套JSON文件并自动转换为无嵌套结构的Pandas DataFrame

通用递归展平函数 #

无需手动指定任何嵌套列名，自动处理所有层级的字典、列表结构，生成单层键值对：

import pandas as pd
import os
import json

def flatten_json(nested_json, parent_key='', sep='.'):
    """
    递归展平嵌套JSON，返回单层字典
    :param nested_json: 待展平的嵌套JSON对象（字典/列表）
    :param parent_key: 上层键名，用于拼接生成最终键
    :param sep: 嵌套层级的分隔符，默认用点
    :return: 展平后的单层字典
    """
    items = []
    if isinstance(nested_json, dict):
        for k, v in nested_json.items():
            new_key = f"{parent_key}{sep}{k}" if parent_key else k
            items.extend(flatten_json(v, new_key, sep=sep).items())
    elif isinstance(nested_json, list):
        # 列表内如果是嵌套结构按索引拼接键名，普通值直接转逗号分隔字符串，可按需调整逻辑
        if all(isinstance(i, (dict, list)) for i in nested_json):
            for idx, v in enumerate(nested_json):
                new_key = f"{parent_key}{sep}{idx}" if parent_key else str(idx)
                items.extend(flatten_json(v, new_key, sep=sep).items())
        else:
            items.append((parent_key, ','.join(map(str, nested_json))))
    else:
        items.append((parent_key, nested_json))
    return dict(items)

单文件处理示例 #

你提供的测试JSON用该函数处理后直接得到无嵌套结构的DataFrame：

test_json = {"vehicle_type":"Car","car_info":{"count":3,"infos":[{"car":"BMW","name":"5","description":"","production_time":"2017-07-16","tags":["car","BMW","sedan"],"references":[],"country":["germany"],"fuel_type":["electrical"],"color":"black","price":null,"technic":{"0-100":"-","horsepower":"165Nm"},"mileage":{"mile":14004}}]}}
flattened_data = flatten_json(test_json)
df = pd.DataFrame([flattened_data])

多文件批量处理代码 #

支持批量读取指定文件夹下所有JSON文件，可选择合并所有结果输出为单个文件，或每个JSON对应输出单独文件：

def batch_process_json(input_dir, output_dir, merge_result=True, output_file='merged_result.csv'):
    """
    批量处理文件夹下所有JSON文件
    :param input_dir: 存放JSON文件的文件夹路径
    :param output_dir: 结果输出文件夹路径
    :param merge_result: 是否将所有JSON的处理结果合并为一个DataFrame
    :param output_file: 合并结果的导出文件名
    """
    os.makedirs(output_dir, exist_ok=True)
    all_dfs = []
    
    # 遍历所有.json后缀的文件
    for filename in os.listdir(input_dir):
        if filename.lower().endswith('.json'):
            file_path = os.path.join(input_dir, filename)
            # 读取JSON文件
            with open(file_path, 'r', encoding='utf-8') as f:
                try:
                    json_data = json.load(f)
                except Exception as e:
                    print(f"读取文件{filename}失败：{e}")
                    continue
            # 展平JSON并转为DataFrame
            flattened_data = flatten_json(json_data)
            df = pd.DataFrame([flattened_data])
            # 新增来源文件名列方便溯源
            df['source_file'] = filename
            
            if merge_result:
                all_dfs.append(df)
            else:
                # 单文件单独导出
                output_path = os.path.join(output_dir, f"{os.path.splitext(filename)[0]}.csv")
                df.to_csv(output_path, index=False, encoding='utf-8-sig')
                print(f"文件{filename}处理完成，已导出到{output_path}")
    
    if merge_result and all_dfs:
        merged_df = pd.concat(all_dfs, ignore_index=True)
        merged_output_path = os.path.join(output_dir, output_file)
        merged_df.to_csv(merged_output_path, index=False, encoding='utf-8-sig')
        print(f"所有文件处理完成，合并结果已导出到{merged_output_path}")

# 调用示例，替换为你自己的路径即可
if __name__ == '__main__':
    INPUT_FOLDER = './your_json_folder'  # JSON文件存放路径
    OUTPUT_FOLDER = './processed_result' # 处理结果输出路径
    batch_process_json(INPUT_FOLDER, OUTPUT_FOLDER, merge_result=True)

补充说明 #

• 列表处理逻辑可按需调整：如果希望列表内的每个对象单独生成一行，可在展平前先遍历列表元素逐个处理
• 分隔符可自定义：修改flatten_json函数的sep参数即可调整嵌套列名的分隔符，比如改成下划线_
• 编码适配：如果你的JSON文件用的是其他编码，修改open函数的encoding参数即可

内容的提问来源于stack exchange，提问作者Atakan Aydınbaş