多處理組應用python

Question

我有兩組，其中一組的行要作爲組處理，另一組有待查看。

test = pd.DataFrame({'Address1':['123 Cheese Way','234 Cookie Place','345 Pizza Drive','456 Pretzel Junction'],'city':['X','U','X','U']}) 
test2 = pd.DataFrame({'Address1':['123 chese wy','234 kookie Pl','345 Pizzza DR','456 Pretzel Junktion'],'city':['X','U','Z','Y'] , 'ID' : ['1','3','4','8']}) 

gr1 = test.groupby('city') 
gr2 = test2.groupby('city') 

目前我申請我的功能組中的每一行，

gr1.apply(lambda x: custom_func(x.Address1, gr2.get_group(x.name))) 

但是我不知道該怎麼做多處理這一點。請指教。

編輯： - 我試圖使用dask，但我不能將整個數據幀傳遞給dask函數 - 因爲它的apply函數有一個限制。我嘗試在我的gr1（group）上使用dask apply，但由於我在自定義函數中設置索引，dask會拋出一個錯誤，指出「索引器太多」。

這裏有DASK，這給了我一個錯誤 - 「Pandas' object has no attribute 'city'

ddf1 = dd.from_pandas(test, 2) 
ddf2 = dd.from_pandas(test2, 2) 

dgr1 = ddf1.groupby('city') 
dgr2 = ddf2.groupby('city') 

meta = pd.DataFrame(columns=['Address1', 'score', 'idx','source_index']) 
ddf1.map_partitions(custom_func, x.Address1, dgr2.get_group(x.city).Address1,meta=meta).compute() 

Answer 1

我提供了另一種解決方案，在這裏使用DASK，

import pandas as pd 
from multiprocessing import Pool 
test = pd.DataFrame({'Address1':['123 Cheese Way','234 Cookie Place','345 Pizza Drive','456 Pretzel Junction'],'city':['X','U','X','U']}) 
test2 = pd.DataFrame({'Address1':['123 chese wy','234 kookie Pl','345 Pizzza DR','456 Pretzel Junktion'],'city':['X','U','Z','Y'] , 'ID' : ['1','3','4','8']}) 

test=test.assign(dataset = 'test') 
test2=test2.assign(dataset = 'test2') 

newdf=pd.concat([test2,test],keys = ['test2','test']) 
gpd=newdf.groupby('city') 
def my_func(mygrp): 
    test_data=mygrp.loc['test'] 
    test2_data=mygrp.loc['test2'] 
    #do something specific 
    #if needed print something 
    return {'Address':test2_data.Address1.values[0],'ID':test2_data.ID.values[0]} #return some other stuff 

mypool=Pool(processes=2) 
ret_list=mypool.imap(my_func,(group for name, group in gpd)) 

pd.DataFrame(ret_list) 

回報像

ID address 
0 3 234 kookie Pl 
1 1 123 chese wy 
2 8 456 Pretzel Junktion 
3 4 345 Pizzza DR 

PS：在OP的問題中，兩個相似的數據集在一個專門的函數中進行比較，這裏的解決方案是u ses pandas.concat。根據問題，還可以想象一個pd.merge。