如何在进程之间共享 pandas DataFrame 对象？答案

【问题标题】：How to share pandas DataFrame object between processes?如何在进程之间共享 pandas DataFrame 对象？
【发布时间】：2013-11-10 06:52:31
【问题描述】：

这个问题与我之前发布的链接有相同的点。

(Is there a good way to avoid memory deep copy or to reduce time spent in multiprocessing?)

因为我遇到了“DataFrame”对象共享问题，所以我对此一无所知。

我简化了示例代码。

如果有任何专业人士修改我的代码以在没有 Manager.list、Manager.dict、numpy sharedmem 的进程之间共享“DataFrame”对象，我会非常感谢她或他。

这是代码。

#-*- coding: UTF-8 -*-'
import pandas as pd
import numpy as np
from multiprocessing import *
import multiprocessing.sharedctypes as sharedctypes
import ctypes

def add_new_derived_column(shared_df_obj):
    shared_df_obj.value['new_column']=shared_df_obj.value['A']+shared_df_obj.value['B'] / 2
    print shared_df_obj.value.head()
    '''
    "new_column" Generated!!!

          A         B  new_column
0 -0.545815 -0.179209   -0.635419
1  0.654273 -2.015285   -0.353370
2  0.865932 -0.943028    0.394418
3 -0.850136  0.464778   -0.617747
4 -1.077967 -1.127802   -1.641868
    '''

if __name__ == "__main__":

    dataframe = pd.DataFrame(np.random.randn(100000, 2), columns=['A', 'B'])

    # to shared DataFrame object, I use sharedctypes.RawValue
    shared_df_obj=sharedctypes.RawValue(ctypes.py_object, dataframe )

    # then I pass the "shared_df_obj" to Mulitiprocessing.Process object
    process=Process(target=add_new_derived_column, args=(shared_df_obj,))
    process.start()
    process.join()

    print shared_df_obj.value.head()
    '''
    "new_column" disappeared.
    the DataFrame object isn't shared.

          A         B
0 -0.545815 -0.179209
1  0.654273 -2.015285
2  0.865932 -0.943028
3 -0.850136  0.464778
4 -1.077967 -1.127802
    '''

【问题讨论】：

你打算做什么？我认为您希望数据帧可用于多处理是有原因的
有没有办法通过不需要共享对象的管道传递您的数据帧？另一种解决方案可能是将您的数据框作为一个数组传递，其中包含 data + 2 数组作为标签，并在修改后重建它。
@CoMartel 这基本上是Queue 发生的情况 - 问题是（对于大帧）与通过管道移动帧相关的开销破坏了并行处理它的任何好处（取决于你到底在做什么）

标签： python multithreading pandas multiprocessing object-sharing

【解决方案1】：

您可以使用命名空间管理器，以下代码按您的预期工作。

#-*- coding: UTF-8 -*-'
import pandas as pd
import numpy as np
from multiprocessing import *
import multiprocessing.sharedctypes as sharedctypes
import ctypes

def add_new_derived_column(ns):
    dataframe2 = ns.df
    dataframe2['new_column']=dataframe2['A']+dataframe2['B'] / 2
    print (dataframe2.head())
    ns.df = dataframe2

if __name__ == "__main__":

    mgr = Manager()
    ns = mgr.Namespace()

    dataframe = pd.DataFrame(np.random.randn(100000, 2), columns=['A', 'B'])
    ns.df = dataframe
    print (dataframe.head())

    # then I pass the "shared_df_obj" to Mulitiprocessing.Process object
    process=Process(target=add_new_derived_column, args=(ns,))
    process.start()
    process.join()

    print (ns.df.head())

【讨论】：