为什么多处理比 Pandas 中的简单计算慢？

Question

这与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 中的每一行，我使用fuzzywuzzy 计算的字符串相似度得分在Dataframe 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

让我将我制作的 cmets 总结为类似答案的内容。我希望这些信息对您有用，因为这里汇集了许多问题。

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

1) 创建流程需要时间。这总是正确的，尤其是在 Windows 上。如果您对现实生活中的性能感兴趣，您将希望只创建一次流程，并使用其固定开销并运行许多任务。在 dask 中，有 many ways 制作您的集群，甚至是在本地。

2) dask（或任何其他调度程序）处理的每个任务都会产生一些开销。在分布式调度器的情况下，这是

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

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

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