pandas处理速度过慢如何分批执行？Flask渐进式返回前端实现咨询

你这个需求本质是后端分批异步处理 + 前端渐进式加载，用你当前已有的Flask + Pandas + 前端JS技术栈就能实现，不需要额外引入复杂组件，具体步骤如下：

1 先解决当前代码的核心性能问题 #

你当前函数慢的核心原因是df.apply逐行调用tdp、get_main_task两个数据库查询函数，437条数据就会执行874次独立的SQL查询，这是耗时的根源，先做这个优化就能大幅提升速度：

def get_project_details(pids: list, conn: pymssql._pymssql.Connection) -> dict:
    pidq = ",".join([f"'{x}'" for x in pids])
    query = f"SELECT * FROM VW_CSMS_PMT_ProjectDetails WHERE Project_Id in ({pidq})"
    df = pd.read_sql(
        query,
        con=conn,
        parse_dates=["StartDate", "TargetDate"],
    )
    # 新增：批量查询所有pid对应的关联数据，仅执行2次SQL
    all_timesheet = get_batch_tdp(pids, conn) # 自行实现该函数，一次性查询所有pid的timesheet数据，返回{project_id: 详情}的字典
    all_main_task = get_batch_main_task(pids, conn) # 同上，一次性查询所有pid的主任务数据
    
    out = df.apply(
        lambda x: {
            "id": x["Project_Id"],
            "type": "project",
            "name": x["Project_Name"],
            "start_date": get_date_string(x["StartDate"]),
            "target_date": get_date_string(x["TargetDate"]),
            "estimate": x["EST_HRS"],
            "status": x["status"],
            # 直接从预查的字典取数据，不用每次查库
            "timesheet_details": all_timesheet.get(x["Project_Id"]),
            "main_task": all_main_task.get(x["Project_Id"]),
        },
        axis=1,
    )

    if isinstance(out, pd.core.series.Series):
        out = out.tolist()
    else:
        out = None
    return out

2 实现分批处理+前端渐进渲染逻辑 #

如果优化后还是需要实现先展示前10条、后台逐步返回剩余数据的需求，用Flask原生支持的SSE（服务器推送事件）实现即可，不需要前端轮询，效率更高：

2.1 后端Flask接口实现 #

首先写一个分组工具函数，再写SSE推送接口：

from flask import Response, request
import json

# 按固定大小切分列表
def split_batch(lst, batch_size=10):
    return [lst[i:i+batch_size] for i in range(0, len(lst), batch_size)]

@app.route('/process_projects', methods=['POST'])
def process_projects():
    all_pids = request.json.get('pids', [])
    batches = split_batch(all_pids)
    conn = 你的数据库连接
    
    def generate_stream():
        # 先返回第一批次数据，前端立即渲染
        first_batch = get_project_details(batches[0], conn)
        yield f'data: {json.dumps({"batch_index": 0, "total_batches": len(batches), "data": first_batch})}\n\n'
        # 后台逐批处理剩余数据，每处理完一批推送一次
        for idx in range(1, len(batches)):
            batch_data = get_project_details(batches[idx], conn)
            yield f'data: {json.dumps({"batch_index": idx, "total_batches": len(batches), "data": batch_data})}\n\n'
    
    return Response(generate_stream(), mimetype='text/event-stream')

2.2 前端接收渲染逻辑 #

用原生JS监听SSE推送的消息，每收到一批数据就追加到页面列表中：

// 你的全量项目ID列表
const allPids = [你的437个pid数组];
const projectListDom = document.getElementById('project-list'); // 页面上的列表容器

// 建立SSE连接
const evtSource = new EventSource('/process_projects', {
    method: 'POST',
    body: JSON.stringify({pids: allPids}),
    headers: {'Content-Type': 'application/json'}
});

// 监听推送消息
evtSource.onmessage = function(event) {
    const res = JSON.parse(event.data);
    // 渲染当前批次的10条数据，追加到列表末尾
    res.data.forEach(project => {
        const item = document.createElement('div');
        item.innerText = project.name; // 按你的需求渲染字段
        projectListDom.appendChild(item);
    });
    // 所有批次处理完成，关闭连接
    if (res.batch_index === res.total_batches - 1) {
        evtSource.close();
    }
}

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