pandas 数据帧上的函数的多重处理
Multiprocessing of a function on a pandas dataframe
我有一个大型 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 的副本并且没有被重新组合在一起。如何将每个过程的结果反映在单个数据框中?
这可能对你有用:
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 数据框,其中包含多个 "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 的副本并且没有被重新组合在一起。如何将每个过程的结果反映在单个数据框中?
这可能对你有用:
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