大型 Pandas Dataframe 并行处理

Posted

技术标签:

【中文标题】大型 Pandas Dataframe 并行处理【英文标题】:Large Pandas Dataframe parallel processing 【发布时间】:2016-02-10 07:46:16 【问题描述】:

我正在访问一个非常大的 Pandas 数据框作为全局变量。此变量通过joblib 并行访问。

例如。

df = db.query("select id, a_lot_of_data from table")

def process(id):
    temp_df = df.loc[id]
    temp_df.apply(another_function)

Parallel(n_jobs=8)(delayed(process)(id) for id in df['id'].to_list())

以这种方式访问​​原始 df 似乎会跨进程复制数据。这是出乎意料的,因为原始 df 在任何子进程中都没有改变? (或者是吗?)

【问题讨论】:

【参考方案1】:

对于joblib创建的每个进程,需要对整个DataFrame进行pickle和unpickled。在实践中,这非常慢,并且还需要每个内存的许多倍。

一种解决方案是使用表格格式将数据存储在 HDF (df.to_hdf) 中。然后,您可以使用select 选择数据子集以进行进一步处理。在实践中,这对于交互式使用来说太慢了。它也非常复杂,您的员工需要存储他们的工作,以便在最后一步合并。

另一种方法是使用target='parallel' 探索numba.vectorize。这将需要使用 NumPy 数组而不是 Pandas 对象,因此它也有一些复杂性成本。

从长远来看,dask 有望为 Pandas 带来并行执行,但这并不是指望很快。

【讨论】:

我从***.com/questions/10721915/… 假设除非原始对象被更改,否则子进程不会收到完整副本。 joblib 是否会破坏写时复制语义? 只有少数类型可以使用共享内存传递。 Pandas 对象不在此列表中。 joblib 在调用Parallel 时使用关键字参数max_nbytes 自动处理numpy 数组的内存共享,具体取决于数组的大小。见joblib's site。另见this answer。当然,您可以使用 NumPy 数组代替 Pandas,您可能会看到加速。【参考方案2】:

如您所述,Python 多处理通常使用单独的进程完成,这意味着这些进程不共享内存。如果您可以使用 np.memmap 来解决问题,则有一个潜在的解决方法,正如在 joblib 文档后面提到的那样,尽管转储到磁盘显然会增加一些开销:https://pythonhosted.org/joblib/parallel.html#working-with-numerical-data-in-shared-memory-memmaping

【讨论】:

以上是关于大型 Pandas Dataframe 并行处理的主要内容,如果未能解决你的问题,请参考以下文章

Pandas 和多处理内存管理:将 DataFrame 拆分为多个块

Python multiprocess.Pool.map 无法处理大型数组。

如何使用 pandas 聚合大型 DataFrame 中的多个列?

dataframe最大存储多少数据?

更新不在磁盘上的大型 DataFrame 对象。

如何制作从大型 xlsx 文件加载 pandas DataFrame 的进度条?