如何使用Python multiprocessing并行处理Pandas DataFrame每行数据

修正后可直接运行的代码 #

import pandas as pd
from multiprocessing import Pool, cpu_count

# 保留你原有get_price的实现逻辑
def get_price(row):
    # 原有HTTP请求处理逻辑
    pass

def check_option(row):
    get_price(row)

def main():
    # 保留你原有读取CSV的逻辑
    df_master = read_bb_csv(file)
    if df_master.empty:
        return
    # 自动适配CPU核心数，也可手动指定固定值如process_num = 2
    process_num = cpu_count()
    # 将DataFrame所有行转为可迭代的tuple列表
    row_iter = list(df_master.itertuples(index=True, name='Pandas'))
    
    # 上下文管理器自动管理进程生命周期，无需手动close/join
    with Pool(process_num) as pool:
        # map自动将行迭代器的每个元素分发到不同进程调用check_option处理
        pool.map(check_option, row_iter)

# Windows系统必须加该入口保护，否则会出现进程无限启动报错
if __name__ == "__main__":
    main()

原代码错误点说明 #

• 缩进逻辑错误，Pool初始化、循环处理的缩进层级不符合Python语法规范
• 错误在外层加for循环遍历行，pool.map本身会遍历传入的可迭代对象，外层循环会导致逻辑完全错乱
• 不需要使用partial封装参数，check_option本身仅接收row单参数，直接传入行迭代器即可

可选优化建议 #

• 你的场景属于IO密集型（大量HTTP请求等待），可以替换为多线程实现降低进程开销，仅需修改导入语句为from multiprocessing.dummy import Pool，其余代码无需改动，执行效率相当
• 无需处理返回结果时，不用接收pool.map的返回值，不影响各进程任务执行

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