How to implement dask mappartitions on multiple dataframe lambda function?

Question

I've implemented a fuzzy string matching algo between two dataframes just using pandas. My issue is how do I convert this to a dask operation using multiple cores? My program runs about 3-4 days on pure python, and I want to parallelize the operations to optimize time cost. I've already used the multiprocessing package to extract the number of cores using the code below:

numCores = multiprocessing.cpu_count()

fields = ['id','phase','new']
emb = pd.read_csv('my_csv.csv', skipinitialspace=True, usecols=fields)

Then I had to subdivide the dataframe emb into two dataframes (emb1, emb2) based on numeric values associated per string. As in I'm matching a dataframe with all elements having value to 3 to their corresponding value 2 in the other dataframe by matched string.The code for pure pandas operation is below.

emb1 = emb[emb.phase.isin([3.0])]
emb1.set_index('id',inplace=True)

emb2 = emb[emb.phase.isin([2.0,1.5])]
emb2.set_index('id',inplace=True)

def fuzzy_match(x, choices, scorer, cutoff):
return process.extractOne(x, choices=choices, scorer=scorer, score_cutoff=cutoff)

FuzzyWuzzyResults = pd.DataFrame(emb1.sort_index().loc[:,'strings'].apply(fuzzy_match, args = (emb2.loc[:,'strings'],fuzz.ratio,90)))

I sort of tried doing a dask implementation using this code:

emb1 = dd.from_pandas(emb1, npartitions=numCores)
emb2 = dd.from_pandas(emb2, npartitions=numCores)

But running the lambda function for two dataframes is confusing me. Any ideas?

guru guru · Accepted Answer · 2018-05-18T22:52:16

So I just fixed my code to remove the manual partition of the dataframe and used groupby instead.

Here's the code:

for i in [2.0,1.5]:
    FuzzyWuzzyResults = emb.map_partitions(lambda df: df.groupby('phase').get_group(3.0)['drugs'].apply(fuzzy_match, args=(df.groupby('phase').get_group(i)['drugs'],fuzz.ratio,90)), meta=('results')).compute()

Not sure whether it's accurate, but at least it's running, and that too on all CPU cores.

How to implement dask mappartitions on multiple dataframe lambda function?

1 Answers