pandas 数据帧上的函数的多重处理

Question

我有一个大型 pandas 数据框，其中包含多个 "records"，其中包含 2 个或更多行项目。我正在尝试使用多处理对每条记录高效地执行 CPU 密集计算。这是一个简单的示例，其中包含一个函数，该函数仅向每条记录添加一个随机数：

import pandas as pd
from random import randrange
from multiprocessing import Pool

#Trivial example function 
def my_func(record): 
    df.loc[((df.Record == record), 'Result')] = randrange(0,100)
    print (df)

d = {'Record': ['A', 'A', 'B', 'B'], 'Values': [100, 200, 50, 70]}
df = pd.DataFrame(d)
all_records = df['Record'].unique()

if __name__ == '__main__':
    pool = Pool(processes=2)
    pool.map(my_func,all_records)
    df.to_csv('output.csv')

所需的输出是原始数据框，其中包含一个标题为 "Result" 的新列，其中包含每个记录的随机数。例如：

    Record  Values  Result
0      A     100    63.0
1      A     200    63.0
2      B      50    22.0
3      B      70    22.0

实际结果是我的 CSV 输出未使用结果列进行更新。我可以通过函数中的 print 语句告诉进程正在工作。根据我的研究，这些过程作用于 df 的副本并且没有被重新组合在一起。如何将每个过程的结果反映在单个数据框中？

Answer 1

这可能对你有用：

import pandas as pd
from random import randrange
from multiprocessing import Pool

#Trivial example function 
def my_func(record):
    sub_df = df.loc[df['Record'] == record]
    sub_df['Result'] = randrange(0,100)
    # return results for the record as pd.Series
    return sub_df['Result']  

d = {'Record': ['A', 'A', 'B', 'B'], 'Values': [100, 200, 50, 70]}
df = pd.DataFrame(d)
all_records = df['Record'].unique()

if __name__ == '__main__':
    pool = Pool(processes=2)
    results = pool.map(my_func, all_records)
    pool.close()
    pool.join()

    # concatenate results into a single pd.Series
    results = pd.concat(results)

    # join results with original df
    joined_df = df.join(results)

    print(joined_df)
    #       Record  Values  Result
    # 0      A     100      90
    # 1      A     200      90
    # 2      B      50      62
    # 3      B      70      62

pandas 数据帧上的函数的多重处理

Multiprocessing of a function on a pandas dataframe

python

python-3.x

pandas

multiprocess