如何通过多线程编辑dataframe行以提升REST API批量处理效率

原方案可行性判定 #

你提出的「拆分DataFrame为等量分块→多线程并行处理→合并结果」的思路完全可行。该场景属于典型的IO密集型任务，耗时主要集中在等待REST API的网络响应阶段，CPU处于空闲状态，使用多线程可以充分利用空闲的CPU时间，大幅压缩整体处理耗时。
实现时需注意两个核心点：

• 拆分分块时保留原DataFrame的行索引，合并时按索引拼接，避免最终结果顺序错乱
• 线程数不要设置过高，建议控制在10~30区间，过高会触发API提供方的限流策略，反而会提升请求失败率、增加整体耗时

更优实现方案 #

不需要手动处理分块、合并逻辑，可以直接使用concurrent.futures.ThreadPoolExecutor实现，代码更简洁，稳定性更高，且不存在多线程修改DataFrame的线程安全问题，核心实现代码如下：

import pandas as pd
from concurrent.futures import ThreadPoolExecutor
import requests

# 单条数据的API请求逻辑，可根据实际需求调整
def fetch_metric(row):
    api_id = row["id"]
    try:
        # 替换为实际API请求逻辑
        resp = requests.get(f"https://your-api-endpoint/{api_id}", timeout=10)
        resp.raise_for_status()
        # 按实际返回结构提取metric值
        return resp.json()["data"]["metric"]
    except Exception as e:
        # 异常时可返回默认值或None，也可增加重试逻辑
        return None

if __name__ == "__main__":
    # 替换为你的实际DataFrame初始化逻辑
    df = pd.read_csv("your_data_source.csv")
    # 线程数可根据API限流规则调整，2000条数据建议设置为10~20
    MAX_WORKERS = 15

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        # 自动维持原行顺序，结果直接赋值到新列
        df["metric"] = list(executor.map(fetch_metric, df.to_dict("records")))

该方案的优势如下：

• 无需手动拆分分块、合并结果，ThreadPoolExecutor会自动完成任务调度
• 自动保留原行的顺序，不会出现结果与行错位的问题
• 多线程仅负责API请求和结果返回，最终列赋值操作在主线程完成，不存在多线程同时修改DataFrame的线程安全问题

如果后续数据量级持续提升，还可以替换为协程方案（如aiohttp+asyncio），资源占用率更低，并发能力更强。

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