並行python迭代

Question

我想根據pandas.DataFrame中的值創建一個類的實例。這我已經失望了。

import itertools 
import multiprocessing as mp 
import pandas as pd 

class Toy: 
    id_iter = itertools.count(1) 

    def __init__(self, row): 
     self.id = self.id_iter.next() 
     self.type = row['type'] 

if __name__ == "__main__": 

    table = pd.DataFrame({ 
     'type': ['a', 'b', 'c'], 
     'number': [5000, 4000, 30000] 
     }) 

    for index, row in table.iterrows(): 
     [Toy(row) for _ in range(row['number'])] 

多重試圖

我已經能夠並行本（排序）加入以下內容：

pool = mp.Pool(processes=mp.cpu_count()) 
m = mp.Manager() 
q = m.Queue() 

for index, row in table.iterrows(): 
    pool.apply_async([Toy(row) for _ in range(row['number'])]) 

看來，這將是如果row['number']數字是快大大長於table的長度。但在我的實際情況中，table是幾千行長，而每個row['number']是比較小的。

嘗試將table分解爲cpu_count()塊並在表內迭代似乎更爲明智。但現在我們處於我的蟒蛇技能的邊緣。

我試過的東西，Python解釋器尖叫我要，如：

pool.apply_async(
     for index, row in table.iterrows(): 
     [Toy(row) for _ in range(row['number'])] 
     ) 

同樣的事情， 「不能醃」

Parallel(n_jobs=4)(
    delayed(Toy)([row for _ in range(row['number'])]) \ 
      for index, row in table.iterrows() 
) 

編輯

這可能會讓我更接近一點，但仍然不在那裏。我在一個單獨的函數中創建類實例，但我被告知'NoneType'對象是不可迭代的。

Answer 1

這對我來說有點不清楚，你期待的輸出是什麼。你只想形式

[Toy(row_1) ... Toy(row_n)] 

其中每個Toy(row_i)出現了多重性row_i.number的大名單？

基於由@JD龍提到我想你可以做這樣的事情的答案：

def process(df): 
    L = [] 
    for index, row in table.iterrows(): 
     L += [Toy(row) for _ in range(row['number'])] 
    return L 

table = pd.DataFrame({ 
    'type': ['a', 'b', 'c']*10, 
    'number': [5000, 4000, 30000]*10 
    }) 

p = mp.Pool(processes=8) 
split_dfs = np.array_split(table,8)  
pool_results = p.map(process, split_dfs) 
p.close() 
p.join() 

# merging parts processed by different processes 
result = [a for L in pool_results for a in L]