python - 为什么 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 中的每一行，我查找数据框 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 (每一行都是一个独立的问题)，所以这些包应该真正发光。

我在这里遗漏了什么吗？

谢谢!

Answer 1

让我将我发表的评论总结为类似答案的内容。我希望这些信息有用，因为这里汇集了许多问题。

首先，我想向您指出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 中优化良好的方法。与往常一样，您应该始终分析您的代码...