Параллельный ручной классификатор OneVsRest

Question

Могу ли я получить помощь, параллельную этому коду? Я преобразовываю проблему классификации с несколькими метками в проблему OneVsRest (двоичная релевантность). Из-за проблем с памятью, упомянутых here, я делаю это вручную.

clf_label = {} 

for i, label in enumerate(label_index.keys()): 
    print 'Fitting', i, 'label out of', len(label_index) 
    clf = SGDClassifier(loss='hinge', shuffle=True, alpha=0.000001, verbose=0, n_iter=5, n_jobs=4) 
    temp_y = np.zeros(trainY.shape) 
    temp_y[label_index[label]] = 1 

    clf.fit(trainX, temp_y) 
    clf_label[label] = clf 

Я перекручивание через keys из label_index и построения классификаторов для каждой метки. После того, как каждый классификатор поместился, я сохраняю его в другом dict, где ключ снова является меткой, но это значение является классификатором. Из-за длительного времени работы я хочу распараллелить этот код. Вот моя попытка с multiprocessing'sPool.map:

def fit_label(label, trainX, trainY, label_index): 
    # print 'Fitting', i, 'label out of', len(label_index) 
    clf = SGDClassifier(loss='hinge', shuffle=True, alpha=0.000001, verbose=0, n_iter=5) 
    temp_y = np.zeros(trainY.shape) 
    temp_y[label_index[label]] = 1 

    clf.fit(trainX, temp_y) 
    return clf 

def linear_svm(): 
    p = Pool(2) 
    func = partial(fit_label, trainX=trainX, trainY=trainY, label_index=label_index) 
    res = p.map(func, label_index.keys()[1:6]) 
    clf_label = dict(zip(label_index.keys()[1:6], res)) 

и я получаю эту ошибку:

Exception in thread Thread-3: 
Traceback (most recent call last): 
    File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/threading.py", line 808, in __bootstrap_inner 
    self.run() 
    File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/threading.py", line 761, in run 
    self.__target(*self.__args, **self.__kwargs) 
    File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/pool.py", line 342, in _handle_tasks 
    put(task) 
SystemError: NULL result without error in PyObject_Call 

Это кажется довольно простой задачей для кого-то, кто знает, как сделать параллельное программирование в Python, так что я было бы очень полезно, если бы кто-то мог переписать это параллельно, вместо того чтобы модифицировать мой (изворотливый) код. Спасибо.

Answer 1

Попробуйте определение вашей функции распараллелить вне функции linear_svm() следующим образом:

def func(fit_label, trainX=None, trainY=None, label_index=None): 
    return partial(fit_label, trainX=trainX, trainY=trainY, label_index=label_index) 


def linear_svm(): 
    numProcessors = multiprocessing.cpu_count() 
    p = Pool(processes=numProcessors) 
    res = p.map_async(func, label_index.keys()[1:6]) 
    poolres = res.get() 
    clf_label = dict(zip(label_index.keys()[1:6], poolres))