并行处理数据帧[副本]

Question

我有一个进程，它要求处理dataframe的每一行，然后向每一行追加一个新值。这是一个很大的数据帧，一次处理一个数据帧需要几个小时。

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

基本上我的代码是这样的

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

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

Answer 1

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

但是为了给出概念，您可以将多处理（pool.map)与块结合使用。读取chunk中的csv（或如答案末尾所述制作chucks）并将其映射到池，在处理每个chunk时添加新行（或将其添加到列表中并制作新chunk）并从函数返回。

最后，在执行所有池时组合数据流。

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)
            

注意：您可以通过相同的逻辑传递chuck，而不是读取csv，为了计算块大小，您可能需要类似round_of((df的长度）/（可用核心数-2))的内容，例如100000/14=round(7142.85)=7150行每块

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