Python批量解析JSON生成DataFrame越跑越慢问题排查

API 探路者

2026-8-29

核心性能问题

你的代码越跑越慢的根本原因是逐行调用df.loc[len(df)]追加数据的写法有严重的性能缺陷：

pandas DataFrame是基于连续内存块存储的结构，不支持高效的原地逐行追加。每次执行追加操作时，pandas都会重新申请一块更大的内存，把当前DataFrame里的所有数据完整复制到新内存块，再写入新行。
这个操作的时间复杂度是O(n²)：数据量小的时候复制开销可以忽略，随着行数增加，每次追加需要复制的数据量线性增长，处理速度会越来越慢，到后期单次追加的耗时可能是初期的上百倍。

除此之外代码里还有几个影响性能/稳定性的小问题：

缺少import pandas as pd导入语句，直接运行会触发名称错误
用字符串加法拼接文件路径不跨平台，Windows系统下会出现路径识别错误
逐字段重复判断字典键、重复索引i["properties"]，存在不必要的性能损耗
裸except会捕获所有异常（包括键盘中断、内存不足这类和文件读取无关的异常），重抛异常时会丢失原始错误栈，不利于排查问题
读文件时先调用f.read()再转JSON，比直接给json.load()传文件对象多了一次全量字符串拷贝，浪费内存。

修正后代码

核心优化思路是：解析过程中先把所有记录存入普通Python列表（Python列表的追加是O(1)均摊复杂度，无全量复制开销），等所有文件解析完成后，一次性把列表转成DataFrame，整体处理速度会提升数十到上百倍。

import os
import json
import pandas as pd

# 提前定义字段和对应JSON路径的映射，减少重复代码
FIELD_MAP = {
    "id": ("id", None),
    "amount": ("properties", "amount"),
    "createdate": ("properties", "createdate"),
    "dealname": ("properties", "dealname"),
    "dealstage": ("properties", "dealstage"),
    "last_source": ("properties", "hs_analytics_latest_source"),
    "first_source": ("properties", "hs_analytics_source"),
    "is_closed": ("properties", "hs_is_closed"),
    "is_closed_won": ("properties", "hs_is_closed_won"),
    "last_modified_date": ("properties", "hs_lastmodifieddate"),
    "landing_site": ("properties", "ip__shopify__landing_site"),
    "type_of_deal": ("properties", "type_of_deal"),
    "pipeline": ("properties", "pipeline"),
}

def read_json(file_path: str) -> dict:
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            # 直接传入文件对象，不需要先read()
            return json.load(f)
    except (OSError, json.JSONDecodeError) as e:
        raise Exception(f"Reading {file_path} file encountered an error: {str(e)}") from e

if __name__ == "__main__":
    # 用普通列表存所有解析出的行，不要提前建空df逐行加
    rows = []
    json_dir = "json"
    for file in os.listdir(json_dir):
        print(file)
        file_path = os.path.join(json_dir, file)
        # 跳过文件夹，避免读子目录时报错
        if not os.path.isfile(file_path):
            continue
        data = read_json(file_path)
        results = data.get("results", [])
        for item in results:
            row = {}
            for field, path in FIELD_MAP.items():
                if len(path) == 1:
                    # id字段直接在item根级别
                    row[field] = item.get(path[0])
                else:
                    # properties下的字段
                    props = item.get("properties", {})
                    row[field] = props.get(path[1])
            rows.append(row)
    
    # 所有数据解析完一次性转DataFrame
    df = pd.DataFrame(rows)
    df.to_csv("historical.csv", index=False)

额外优化说明

用字段映射表代替大段重复的if-else判断，代码更简洁易维护
用字典的.get()方法直接取键对应的值，不存在时默认返回None，省去重复的键存在判断
导出csv时加index=False，避免把pandas自动生成的行索引写到输出文件里
加了文件类型判断，遍历到json目录下的子文件夹时不会报错
指定了文件读取编码为utf-8，避免不同系统默认编码不一致导致的中文乱码问题

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