并行处理数据帧[重复]答案

【问题标题】：processing a dataframe in parallel [duplicate]并行处理数据帧[重复]
【发布时间】：2021-01-29 09:30:22
【问题描述】：

我有一个过程，需要处理数据帧的每一行，然后将一个新值附加到每一行。这是一个大型数据帧，一次处理一个数据帧需要花费数小时。

如果我有一个将每一行发送到函数的 iterrow 循环，我可以并行化我的处理以加快速度吗？该行的结果不相关

基本上我的代码是这样的

for index, row in df.iterrows():
   row['data'] = function[row]

有没有一种简单的方法可以加快处理速度？

【问题讨论】：

大概看看Dask吧。
根据我的经验，如果你正在循环一个 DataFrame，你可能不是用 Pandas 的方式来做的。
正如@Jarad 所说，您可能不想迭代。查看是否存在矢量化替代方案（不适用，取决于function 是否有更好的选择）——请参阅this answer by me 以及when (not) to use apply 上的这个。

标签： python pandas

【解决方案1】：

虽然迭代行不是好的做法，并且可以使用 grouby/transform 聚合等替代逻辑，但如果在最坏的情况下您确实需要这样做，请遵循答案。此外，您可能不需要在这里重新实现所有内容，您可以使用 Dask 之类的库，它构建在 pandas 之上。

但只是为了提供想法，您可以将multiprocessing (Pool.map) 与chunking 结合使用。读取块中的csv（或在答案末尾提到的制作夹头）并将其映射到池中，在处理每个块时添加新行（或将它们添加到列表并制作新块）并从函数中返回。

最后在所有池执行完毕后合并数据帧。

import pandas as pd
import numpy as np
import multiprocessing


def process_chunk(df_chunk):
        
        for index, row in df_chunk.reset_index(drop = True).iterrows():
                    #your logic for updating this chunk or making new chunk here
                         
                    print(row)
                    
                    print("index is " + str(index))
        #if you can added to same df_chunk, return it, else if you appended
        #rows to have list_of_rows, make a new df with them and return
        #pd.Dataframe(list_of_rows)  

        return df_chunk   


if __name__ == '__main__':
            #use all available cores , otherwise specify the number you want as an argument,
            #for example if you have 12 cores,  leave 1 or 2 for other things
            pool = multiprocessing.Pool(processes=10) 
            
            results = pool.map(process_chunk, [c for c in pd.read_csv("your_csv.csv", chunksize=7150)])
            pool.close()
            pool.join()
            
            #make new df by concatenating
            
            concatdf = pd.concat(results, axis=0, ignore_index=True)

注意：您可以通过相同的逻辑传递chucks，而不是读取csv，来计算块大小，您可能需要round_of( (length of df) / (number of core available-2)) 例如100000/14 = round(7142.85) = 7150 rows 每个块

 results = pool.map(process_chunk,
        [df[c:c+chunk_size] for c in range(0,len(df),chunk_size])

【讨论】：

【解决方案2】：

为什么不使用df.iterrows() 而不是使用apply() 之类的矢量化方法？

df.apply(function, axis=1)

.apply() 是 Pandas 对列/行执行迭代的方法。它利用矢量化技术，将简单和复杂操作的执行速度提高了很多倍。

查看这篇Reference 文章，了解它有何不同。

其他选项正在查看Dask、Vaex 或只是旧时尚Multiprocessing。

【讨论】：