多进程场景下合并多个Pandas DataFrame并输出的问题求解

问题原因 #

1. pd.DataFrame.to_csv()默认开启表头写入（header=True），每个子进程追加写入时都会将当前DataFrame的列名作为新行写入文件，而pd.read_csv()仅会将第一行识别为全局表头，后续写入的表头会被判定为普通数据，因此只有第一个name对应的列名能被正常识别。
2. 原代码存在两处可优化点：for循环语句后缺少冒号导致语法错误；仅对最后一个进程调用join()，会出现主进程提前退出、子进程未执行完成的问题。
3. 你当前的实现是行追加写入，如果需求是将各子进程的单列DataFrame按列合并为多列DataFrame（最终列包含bob、steve），行追加的写入逻辑本身不匹配需求，csv的追加操作是在文件末尾新增行，无法新增列。

方案1：适配行合并场景 #

如果多个子进程生成的DataFrame结构一致，需要按行合并为同一个文件，可以调整to_csv的表头写入规则，仅首次写入时输出表头：

import pandas as pd
from multiprocessing import Process, Lock
import os

def foo(name, lock):
    d = {f'{name}': [1, 2]}
    df = pd.DataFrame(data=d)

    lock.acquire()
    try:
        # 仅文件不存在时写入表头
        is_first_write = not os.path.exists('output.txt')
        df.to_csv('output.txt', mode='a', header=is_first_write, index=False)
    finally:
        lock.release()

if __name__ == '__main__':
    lock = Lock()
    processes = []
    for name in ['bob','steve']:
        p = Process(target=foo, args=(name, lock))
        processes.append(p)
        p.start()
    # 等待所有子进程执行完成
    for p in processes:
        p.join()

方案2：主进程统一合并（适配列合并场景，更推荐） #

如果需要将各子进程的不同列DataFrame合并为多列结果，推荐用队列收集所有子进程的输出，主进程统一合并后再写入，无需加锁控制IO，逻辑更稳定：

import pandas as pd
from multiprocessing import Process, Queue

def foo(name, queue):
    d = {f'{name}': [1, 2]}
    df = pd.DataFrame(data=d)
    # 将生成的DataFrame传入队列
    queue.put(df)

if __name__ == '__main__':
    res_queue = Queue()
    processes = []
    name_list = ['bob','steve']
    
    for name in name_list:
        p = Process(target=foo, args=(name, res_queue))
        processes.append(p)
        p.start()
    # 等待所有子进程执行完成
    for p in processes:
        p.join()
    
    # 取出所有结果按列合并
    df_list = [res_queue.get() for _ in range(len(name_list))]
    final_df = pd.concat(df_list, axis=1)
    # 统一写入文件
    final_df.to_csv('output.txt', index=False)

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