如何并行化向Pandas DataFrame追加数据的循环操作？

你的场景属于IO密集型（包含SQL查询操作），用线程池是最优选择。需要注意的是Pandas DataFrame并非线程安全，不能让多个线程直接修改同一个DataFrame，正确的做法是让每个线程独立返回结果，最后统一合并。

实现步骤与代码示例 #

import pandas as pd
from concurrent.futures import ThreadPoolExecutor

# 你的原有业务函数（根据实际逻辑调整）
def myFunction(i, A, B, C):
    # 这里是你的计算逻辑+SQL查询操作
    # 返回一个Pandas Series格式的行数据
    ...

if __name__ == "__main__":
    # 初始化业务参数（根据实际情况定义A、B、C）
    A = ...
    B = ...
    C = ...
    total_iterations = 10000
    thread_count = 5  # 设定4-5个并行线程

    # 用线程池批量执行任务，收集所有返回的Series
    with ThreadPoolExecutor(max_workers=thread_count) as executor:
        # 将循环任务映射到线程池，传递必要参数
        results = list(executor.map(lambda i: myFunction(i, A, B, C), range(total_iterations)))

    # 过滤掉可能的空结果（如果有任务执行失败返回空Series）
    valid_results = [series for series in results if not series.empty]

    # 一次性合并所有Series为最终DataFrame，效率远高于循环append
    df_final = pd.concat(valid_results, ignore_index=True)

关键说明 #

• 线程池选择：SQL查询属于IO密集型操作，ThreadPoolExecutor比进程池更轻量、开销更小，适合这类场景。
• 线程安全规避：绝对不要让多个线程直接调用df_final.append()，DataFrame的修改操作不是原子性的，多线程同时操作会导致数据错乱、丢失甚至程序崩溃。
• 合并效率优化：pd.concat一次性合并所有Series，比循环调用append的效率提升非常明显（尤其是数据量较大时），ignore_index=True可重置最终DataFrame的索引。
• 异常处理（可选）：如果担心个别任务执行失败，可以在任务函数中加入异常捕获，返回空Series后再过滤，避免单个任务失败导致整个流程中断。
• SQL连接注意：如果myFunction中使用数据库连接，建议每个线程创建独立的连接（不要复用全局连接），避免连接池的线程安全问题。

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