为什么多处理比熊猫中的简单计算慢?

Question

这与如何在Pandas中使用apply并行化许多(模糊)字符串比较有关？

再考虑这个简单(但很有趣)的例子：

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中的每一行，我使用fuzzywuzzy计算的字符串相似性得分查找数据帧slave以获得最佳匹配。

现在，让我们将这些数据帧放大一点：

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

问题：为什么与 Pandas 相比，Dask和multiprocessing的多处理速度如此之慢？假设我的真实数据远大于此。我能得到更好的结果吗？

毕竟，我在这里考虑的问题是embarassingly parallel(每一行都是一个独立的问题)，所以这些包应该真正闪耀。

我在这里错过了什么吗？

谢谢！

Answer 1

让我把我所做的评论总结成一个答案。我希望这些信息证明是有用的，因为这里有许多问题合二为一。

首先，我想向您指出 distributed.readthedocs.io/en/latest/efficiency.html，其中讨论了许多 dask 性能主题。请注意，这都是关于分布式调度程序的，但由于它可以在进程内启动，使用线程或进程，或这些的组合，它确实取代了以前的 dask 调度程序，并且通常建议在所有情况下。

1)创建流程需要时间。这总是正确的，在窗户上尤其如此。如果您对实际性能感兴趣，则只需创建一次具有固定开销的流程，并运行许多任务。在 dask 中，有很多方法可以创建集群，甚至可以在本地创建集群。

2) dask(或任何其他调度程序)处理的每个任务都会产生一些开销。在分布式调度程序的情况下，这是<1ms，但在任务本身的运行时非常短的情况下，这可能很重要。

3) dask 中的一种反模式是在客户端中加载整个数据集并将其传递给工作线程。相反，您希望使用dask.dataframe.read_csv等函数，其中数据由工作线程加载，从而避免昂贵的序列化和进程间通信。Dask 非常擅长将计算移动到数据所在的位置，从而最大限度地减少通信。

4)当进程之间的通信时，序列化的方法很重要，这是我对为什么非dask多处理对你来说如此缓慢的猜测。

5)最后，并非所有工作都会在dask下获得绩效提升。这取决于许多因素，但通常主要的是：数据是否适合内存？如果是，可能很难匹配numpy和pandas中优化良好的方法。与往常一样，您应该始终分析您的代码...