多进程中向全局DataFrame写入变量失败问题求助

问题场景 #

需要并行执行两个函数分别生成不同DataFrame，但多进程无法将结果写入全局变量，运行代码时抛出AttributeError: 'ValueProxy' object has no attribute 'append'，最终全局DataFrame仍为空。

错误代码示例 #

import pandas as pd
import multiprocess as mp
manager = mp.Manager()

df_build = pd.DataFrame()

df_0 = df_1 = manager.Value(pd.DataFrame, df_build)

df_test = pd.DataFrame({'a': range(0,9),
                        'b': range(10,19),
                        'c': range(100,109),
                        'd': range(200,209)})

def frame_0(x = df_0):
    x = x.append(df_test[0:2])
    print('result frame_0','\n', x)
    return x
    
def frame_1(y = df_1):
    y = y.append(df_test[2:4])
    print('result frame_1','\n', y)
    return y

starts_thread = mp.Process(target = frame_0, args = ())
finishes_thread = mp.Process(target = frame_1, args = ())
starts_thread.start()
finishes_thread.start()
starts_thread.join()
finishes_thread.join()    

print('DF_0',df_0, '\n'
      'DF_1', df_1)

执行输出 #

AttributeError: 'ValueProxy' object has no attribute 'append'
DF_0 Value(<class 'pandas.core.frame.DataFrame'>, Empty DataFrame
Columns: []
Index: []) 
DF_1 Value(<class 'pandas.core.frame.DataFrame'>, Empty DataFrame
Columns: []
Index: [])

问题原因 #

• manager.Value(pd.DataFrame, df_build)生成的是ValueProxy代理对象，并非真正的DataFrame，它不支持DataFrame的append等方法。
• 即使在函数中执行x = x.append(...)，也只是在子进程中创建了局部变量，完全没有修改到Manager管理的共享对象。
• 之前列表写入成功是因为manager.list()返回的是支持append的代理列表，与DataFrame的代理逻辑不同。

解决方法 #

方案1：用Manager.dict存储DataFrame #

通过共享字典保存每个进程生成的DataFrame，子进程计算完成后将结果存入字典，主进程再取出使用：

import pandas as pd
import multiprocess as mp

manager = mp.Manager()
# 用共享字典存储多进程结果
shared_dfs = manager.dict()

df_test = pd.DataFrame({'a': range(0,9),
                        'b': range(10,19),
                        'c': range(100,109),
                        'd': range(200,209)})

def frame_0():
    result = df_test[0:2].copy()
    print('result frame_0','\n', result)
    shared_dfs['df0'] = result
    
def frame_1():
    result = df_test[2:4].copy()
    print('result frame_1','\n', result)
    shared_dfs['df1'] = result

if __name__ == '__main__':
    p0 = mp.Process(target=frame_0)
    p1 = mp.Process(target=frame_1)
    p0.start()
    p1.start()
    p0.join()
    p1.join()    

# 从共享字典提取结果
df_0 = shared_dfs['df0']
df_1 = shared_dfs['df1']
print('DF_0', '\n', df_0)
print('DF_1', '\n', df_1)

方案2：用进程池+返回值收集 #

无需共享变量，直接通过进程池的apply_async获取每个进程的返回结果，主进程统一收集：

import pandas as pd
import multiprocess as mp

df_test = pd.DataFrame({'a': range(0,9),
                        'b': range(10,19),
                        'c': range(100,109),
                        'd': range(200,209)})

def frame_0():
    result = df_test[0:2].copy()
    print('result frame_0','\n', result)
    return result
    
def frame_1():
    result = df_test[2:4].copy()
    print('result frame_1','\n', result)
    return result

if __name__ == '__main__':
    with mp.Pool(2) as pool:
        res0 = pool.apply_async(frame_0)
        res1 = pool.apply_async(frame_1)
        # 阻塞获取进程执行结果
        df_0 = res0.get()
        df_1 = res1.get()

print('DF_0', '\n', df_0)
print('DF_1', '\n', df_1)

关键注意点 #

• 避免用manager.Value或manager.Array存储DataFrame这类复杂对象，它们更适合简单数据类型。
• 子进程中操作DataFrame时，建议用copy()避免意外的引用问题。
• Windows环境下必须添加if __name__ == '__main__':，否则会触发多进程启动的递归错误。

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