如何用NumPy与Pandas优化1700条数据的Excel导出性能

问题背景 #

现有Python代码处理1700条数据耗时约1分30秒，存在明显性能瓶颈。尝试直接将接口响应追加到Pandas DataFrame时出现数据覆盖问题，希望通过NumPy/Pandas优化提升效率。原代码如下：

原json_to_excel函数 #

def json_to_excel(dtAltaInicial: date, dtAltaFinal: date):
    path = '../../public/exporta_dados.xlsx'
    start = timer()
    data = getAttendance(dataAltaInicial=str(dtAltaInicial),dataAltaFinal=str(dtAltaFinal))
    print(len(data))
    result = [flatten_data(item) for item in data]
    df = pd.read_json(json.dumps(result))
    df.to_excel(path, index=False, header=True)
    end = timer()
    print(timedelta(seconds=end-start))
    headers = {'Content-Disposition': 'attachment; filename="exporta_dados.xlsx"'}

原flatten_data函数 #

def flatten_data(y):
    out = {}

    def flatten(x, name=''):
        if type(x) is dict:
            for a in x:
                flatten(x[a], name + a + '_')
        elif type(x) is list:
            i = 0
            for a in x:
                flatten(a, name + str(i) + '_')
                i += 1
        else:
            out[name[:-1]] = x

    flatten(y)
    return out

原getAttendance函数 #

def getAttendance(dataAltaInicial, dataAltaFinal):
    url = 'https://api.com.br/search'
    headers = {...}
    body={..., 'page': 1, 'dataAltaInicial': dataAltaInicial, 'dataAltaFinal': dataAltaFinal}

    response = requests.post(url, headers=headers, json=body)
    data = json.loads(response.text)
    total = data['total']
    data = data['items']
    totalPages = ceil(total / 100)

    if totalPages > 1:
        data = getAllPages(url, headers, totalPages, body, data)

    data = data if total > 0 else None
    return data

原getAllPages函数 #

def getAllPages(url, headers, totalPages, body, data):
    atendimentos = []
    atendimentos = atendimentos + data
    for i in range(2, totalPages + 1):
        body.update({'page': i})
        reponse = requests.post(url, headers=headers, json=body)
        data = json.loads(reponse.text)
        atendimentos = atendimentos + data['items']

    return atendimentos

核心优化点及实现 #

1. 数据获取：异步请求+高效列表操作 #

同步循环请求接口是主要耗时点之一，改用异步请求并行获取数据；同时用list.extend()代替列表拼接+，避免频繁创建新列表损耗性能。

优化后的异步数据获取代码

import aiohttp
import asyncio
from math import ceil

async def fetch_page(session, url, headers, body):
    async with session.post(url, headers=headers, json=body) as response:
        return await response.json()

async def get_attendance_async(dataAltaInicial, dataAltaFinal):
    url = 'https://api.com.br/search'
    headers = {...}
    body = {...,'page': 1, 'dataAltaInicial': dataAltaInicial, 'dataAltaFinal': dataAltaFinal}
    
    async with aiohttp.ClientSession() as session:
        # 获取第一页及总页数
        first_page = await fetch_page(session, url, headers, body)
        total = first_page['total']
        if total == 0:
            return []
        total_pages = ceil(total / 100)
        all_data = first_page['items']
        
        # 构造后续页面请求任务
        tasks = []
        for page in range(2, total_pages + 1):
            body_copy = body.copy()
            body_copy['page'] = page
            tasks.append(fetch_page(session, url, headers, body_copy))
        
        # 并行执行所有请求
        pages = await asyncio.gather(*tasks)
        for page_data in pages:
            all_data.extend(page_data['items'])
        
        return all_data

2. 嵌套JSON展开：用Pandas原生json_normalize替代自定义递归 #

自定义flatten_data纯Python递归效率低，Pandas的pd.json_normalize()是C扩展实现，处理嵌套JSON速度快数倍，还能自动处理列表嵌套（可通过record_path指定展开路径）。

3. DataFrame构建：避免无效的JSON序列化/反序列化 #

原代码pd.read_json(json.dumps(result))属于冗余操作，直接从字典列表创建DataFrame即可；若用json_normalize，可直接生成结构化DataFrame。

优化后的json_to_excel函数

from time import timer
from datetime import timedelta
import pandas as pd
import asyncio

def json_to_excel(dtAltaInicial: date, dtAltaFinal: date):
    path = '../../public/exporta_dados.xlsx'
    start = timer()
    
    # 异步获取数据
    data = asyncio.run(get_attendance_async(str(dtAltaInicial), str(dtAltaFinal)))
    print(len(data))
    
    # 直接用json_normalize展开嵌套数据并生成DataFrame
    df = pd.json_normalize(data)
    
    # 导出Excel（指定引擎提升导出速度）
    df.to_excel(path, index=False, header=True, engine='xlsxwriter')
    
    end = timer()
    print(f"耗时：{timedelta(seconds=end-start)}")
    headers = {'Content-Disposition': 'attachment; filename="exporta_dados.xlsx"'}

4. 解决DataFrame追加覆盖问题 #

若仍需要分批追加DataFrame，需使用pd.concat([df1, df2], ignore_index=True)，避免因索引重复导致的数据覆盖。示例：

# 分批构建DataFrame示例
dfs = []
for page_data in pages:
    df_page = pd.json_normalize(page_data['items'])
    dfs.append(df_page)
df = pd.concat(dfs, ignore_index=True)

其他辅助优化建议 #

• 导出Excel时指定engine='xlsxwriter'，比默认引擎速度更快
• 若数据量极大，可考虑分块导出或使用CSV格式（导出速度远快于Excel）
• 对接口请求添加超时设置，避免因网络问题阻塞程序

内容的提问来源于stack exchange，提问作者João Pedro