这与 how to parallelize many (fuzzy) string comparisons using apply in Pandas? 相关
再次考虑这个简单(但有趣)的例子:
import dask.dataframe as dd
import dask.multiprocessing
import dask.threaded
from fuzzywuzzy import fuzz
import pandas as pd
master= pd.DataFrame({'original':['this is a nice sentence',
'this is another one',
'stackoverflow is nice']})
slave= pd.DataFrame({'name':['hello world',
'congratulations',
'this is a nice sentence ',
'this is another one',
'stackoverflow is nice'],'my_value': [1,2,3,4,5]})
def fuzzy_score(str1, str2):
return fuzz.token_set_ratio(str1, str2)
def helper(orig_string, slave_df):
slave_df['score'] = slave_df.name.apply(lambda x: fuzzy_score(x,orig_string))
#return my_value corresponding to the highest score
return slave_df.loc[slave_df.score.idxmax(),'my_value']
master
Out[39]:
original
0 this is a nice sentence
1 this is another one
2 stackoverflow is nice
slave
Out[40]:
my_value name
0 1 hello world
1 2 congratulations
2 3 this is a nice sentence
3 4 this is another one
4 5 stackoverflow is nice
我需要做的很简单:
- 对于
master
中的每一行,我查找数据框slave
使用由fuzzywuzzy
计算的字符串相似度得分来获得最佳匹配.
现在让我们把这些数据框放大一点:
master = pd.concat([master] * 100, ignore_index = True)
slave = pd.concat([slave] * 10, ignore_index = True)
首先,我尝试过 dask
#prepare the computation
dmaster = dd.from_pandas(master, npartitions=4)
dmaster['my_value'] = dmaster.original.apply(lambda x: helper(x, slave),meta=('x','f8'))
现在时间安排如下:
#multithreaded
%timeit dmaster.compute(get=dask.threaded.get)
1 loop, best of 3: 346 ms per loop
#multiprocess
%timeit dmaster.compute(get=dask.multiprocessing.get)
1 loop, best of 3: 1.93 s per loop
#good 'ol pandas
%timeit master['my_value'] = master.original.apply(lambda x: helper(x,slave))
100 loops, best of 3: 2.18 ms per loop
其次,我尝试过使用旧的 multiprocessing
包
from multiprocessing import Pool, cpu_count
def myfunc(df):
return df.original.apply(lambda x: helper(x, slave))
from datetime import datetime
if __name__ == '__main__':
startTime = datetime.now()
p = Pool(cpu_count() - 1)
ret_list = p.map(myfunc, [master.iloc[1:100,], master.iloc[100:200 ,],
master.iloc[200:300 ,]])
results = pd.concat(ret_list)
print datetime.now() - startTime
这给出了大约相同的时间
runfile('C:/Users/john/untitled6.py', wdir='C:/Users/john')
0:00:01.927000
问题:为什么同时使用 Dask
进行多处理和multiprocessing
与这里的 Pandas 相比这么慢?假设我的真实数据比这大得多。我能得到更好的结果吗?
毕竟我这里考虑的问题是embarassingly parallel
(每一行都是一个独立的问题),所以这些包应该真正发光。
我在这里遗漏了什么吗?
谢谢!
最佳答案
让我将我发表的评论总结为类似答案的内容。我希望这些信息有用,因为这里汇集了许多问题。
首先,我想向您指出distributed.readthedocs.io/en/latest/efficiency.html,其中讨论了许多 dask 性能主题。请注意,这都是就分布式调度程序而言的,但由于它可以使用线程或进程或它们的组合在进程内启动,因此它确实取代了以前的 dask 调度程序,并且通常在所有情况下都推荐使用。
1) 创建进程需要时间。这始终是正确的,尤其是在 Windows 上。如果您对现实生活中的性能感兴趣,您将希望仅创建一次进程(具有固定的开销)并运行许多任务。白天有many ways创建集群,甚至在本地。
2) dask(或任何其他调度程序)处理的每个任务都会产生一些开销。在分布式调度程序的情况下,该值<1ms,但在任务本身的运行时间非常短的情况下,这可能很重要。
3) dask 中的反模式是在客户端加载整个数据集并将其传递给工作人员。相反,您希望使用诸如 dask.dataframe.read_csv 之类的函数,其中数据由工作线程加载,从而避免昂贵的序列化和进程间通信。 Dask 非常擅长将计算转移到数据所在的位置,从而最大限度地减少通信。
4)当进程之间进行通信时,序列化方法很重要,这就是我对非 dask 多处理速度如此慢的猜测。
5) 最后,并非所有作业都会在 dask 下获得性能提升。这取决于很多因素,但通常主要的因素是:数据是否适合内存?如果是的话,可能很难匹配 numpy 和 pandas 中优化良好的方法。与往常一样,您应该始终分析您的代码...
关于python - 为什么 Pandas 中的多重处理比简单计算慢?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/49837539/