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

Question

我有以下问题

我有一个包含句子的数据框主文件,例如

master
Out[8]: 
                  original
0  this is a nice sentence
1      this is another one
2    stackoverflow is nice

对于Master中的每一行,我使用查找到另一个Dataframe 从站以获得最佳匹配fuzzywuzzy.我使用fuzzywuzzy,因为两个数据帧之间的匹配句子可能有点不同(额外的字符等).

例如,奴隶可能是

slave
Out[10]: 
   my_value                      name
0         2               hello world
1         1           congratulations
2         2  this is a nice sentence 
3         3       this is another one
4         1     stackoverflow is nice

这是一个功能齐全,精彩,紧凑的工作示例:)

from fuzzywuzzy import fuzz
import pandas as pd
import numpy as np
import difflib


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': [2,1,2,3,1]})

def fuzzy_score(str1, str2):
    return fuzz.token_set_ratio(str1, str2)

def helper(orig_string, slave_df):
    #use fuzzywuzzy to see how close original and name are
    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.ix[slave_df.score.idxmax(),'my_value']

master['my_value'] = master.original.apply(lambda x: helper(x,slave))

100万美元的问题是:我可以将我的应用代码并行化吗？

毕竟,每一行都master与所有行进行比较slave(slave是一个小数据集,我可以将多个数据副本保存到RAM中).

我不明白为什么我不能进行多次比较(即同时处理多行).

问题:我不知道该怎么做或者甚至可能.

任何帮助非常感谢!

Answer 1

您可以使用Dask.dataframe将其并行化.

>>> dmaster = dd.from_pandas(master, npartitions=4)
>>> dmaster['my_value'] = dmaster.original.apply(lambda x: helper(x, slave), name='my_value'))
>>> dmaster.compute()
                  original  my_value
0  this is a nice sentence         2
1      this is another one         3
2    stackoverflow is nice         1

另外,你应该考虑在这里使用线程与进程之间的权衡.您的模糊字符串匹配几乎肯定不会释放GIL,因此您不会从使用多个线程中获得任何好处.但是,使用进程会导致数据序列化并在您的计算机上移动,这可能会使事情变慢.

您可以通过管理方法的get=关键字参数来尝试使用线程和进程或分布式系统compute().

import dask.multiprocessing
import dask.threaded

>>> dmaster.compute(get=dask.threaded.get)  # this is default for dask.dataframe
>>> dmaster.compute(get=dask.multiprocessing.get)  # try processes instead