многопроцессорная обработка Python .get () никогда не заканчивается - PullRequest
Новая
       8

многопроцессорная обработка Python .get () никогда не заканчивается

0 голосов
/ 26 июня 2018

Я пришел сюда, потому что у меня есть проблема с многопроцессорностью в моем скрипте Python.мой код: 

def filter_list_of_list_values(myList,myFilterList):
    for index in range(len(myList)):
        print(index)
        sub_array = myList[index]
        for stopword in myFilterList :
            sub_array = list(filter(lambda a: a != stopword, sub_array))
        sub_array = [w for w in sub_array if not w in myFilterList]
        myList[index] = sub_array
    return myList  

import multiprocessing
import numpy as np
#We are going to execute a multiprocessing and split the list in as many parts than processors used :
N_PROCS = 6
#Lists splitting : 
L_sub_lists  = np.array_split(tokenized_text, N_PROCS)


final_List = []
start_time = time.time()


print('Creating pool with %d processes\n' % N_PROCS)
with multiprocessing.Pool(N_PROCS) as pool:
    #We initialize a list of tasks which each call the same function, but
    #with a diffrent list
    TASKS = [(sub_list, english_stopwords) for sub_list in L_sub_lists]
    print("TASK OK")

    results = [pool.apply_async(filter_list_of_list_values, t) for t in TASKS]
    print("results OK")

    final_results = [r.get() for r in results]
    print("final_results OK")

    for sub_list_res in final_results:
        print("appending")
        final_List+= sub_list_res
        print("list_append")

print("--- %s seconds ---" % (time.time() - start_time))

Сценарий застрял на: 

final_results = [r.get() for r in results]

Я действительно не понимаю, почему, потому что я использовал тот же сценарий (с некоторыми небольшими отличиями) с другимконтекст (другая функция и применяется к DataFrame, а не списку списков) и все работает очень хорошо

Пример: 

L = [['Paris', 'Monaco', 'Washington', 'Lyon', 'Venise', 'Marseille'],
 ['New-York', 'NapleWashington', 'Turin', 'Chicago', 'Las Vegas'],
 ['Lyon', 'Rome', 'Chicago', 'Venise', 'Naple', 'Turin']]

filter_list_of_list_values(L,['Lyon','Turin','Chicago'])

приведет к: 

[['Paris', 'Monaco', 'Washington', 'Venise', 'Marseille'], ['New-York', 'NapleWashington', 'Las Vegas'], ['Rome', 'Venise', 'Naple']]

1 Ответ

0 голосов
/ 26 июня 2018

Итак, я попытался взглянуть на это, и похоже, что ваш код почти прав.Вам не хватает того, как вы на самом деле формируете свой вклад, хотя.Ваш пример кода терпит неудачу, потому что у вас нет импортированного времени, и вы не определили tokenized_text, и я понятия не имею, каким на самом деле должен быть ввод.НО!Основываясь на ваших примерах, ваш код работает, поэтому я подозреваю, что все, что вы делаете для формирования ввода, неверно

Вот в основном функциональная версия вашего кода 

import time
import multiprocessing
import numpy as np

N_PROCS = 6
# L_sub_lists = np.array_split(tokenized_text, N_PROCS)

final_List = []
start_time = time.time()

L = [['Paris', 'Monaco', 'Washington', 'Lyon', 'Venise', 'Marseille'],
     ['New-York', 'NapleWashington', 'Turin', 'Chicago', 'Las Vegas'],
     ['Lyon', 'Rome', 'Chicago', 'Venise', 'Naple', 'Turin']]

filter_list = ['Lyon', 'Turin', 'Chicago']


def filter_list_of_list_values(myList, myFilterList):
    for index in range(len(myList)):
        sub_array = myList[index]
        for stop_word in myFilterList:
            sub_array = list(filter(lambda a: a != stop_word, sub_array))
        sub_array = [w for w in sub_array if w not in myFilterList]
        myList[index] = sub_array
    return myList  


print(filter_list_of_list_values(L, filter_list))

print('Creating pool with %d processes\n' % N_PROCS)
with multiprocessing.Pool(N_PROCS) as pool:
    TASKS = [([sub_list], filter_list) for sub_list in L]
    print("TASK OK")

    results = [pool.apply_async(filter_list_of_list_values, t) for t in TASKS]
    print("results OK")

    print("Getting final results")
    final_results = [r.get() for r in results]
    print("final_results OK")

print("Printing final_results %s" % final_results)
print("--- %s seconds ---" % (time.time() - start_time))

По сути, он просто разбивает ваш большой список списков на более мелкие и обрабатывает их в подпроцессах.Я тестирую вашу основную функцию один раз перед проверкой, чтобы проверить ожидаемый результат и результаты, чтобы убедиться, что распределенная обработка возвращает то же самое.Я думаю, что в этом суть, хотя я не уверен, потому что вопрос не ясен, кроме «Этот код не работает» и «Я ожидаю, что эти результаты»

Вот вывод сценария: 

[['Paris', 'Monaco', 'Washington', 'Venise', 'Marseille'], ['New-York', 'NapleWashington', 'Las Vegas'], ['Rome', 'Venise', 'Naple']]
Creating pool with 6 processes

TASK OK
results OK
Getting final results
final_results OK
Printing final_results [[['Paris', 'Monaco', 'Washington', 'Venise', 'Marseille']], [['New-York', 'NapleWashington', 'Las Vegas']], [['Rome', 'Venise', 'Naple']]]
...