Многопроцессор в python использует только один процесс

Question

Я пытаюсь изучить многопроцессорность с помощью python. Я написал простой код, который должен кормить каждый процесс 1000 строк из входного файла txt. Моя основная функция читает строку, разбивает ее, а затем выполняет некоторые очень простые операции с элементами в строке. В конечном итоге результаты должны быть записаны в выходной файл.

Когда я запустил его, 4 процесса были правильно порождены, но только один процесс фактически работает с минимальным процессором. В результате код очень медленный и бросает вызов цели использования многопроцессорности в первую очередь. Я думаю, что у меня нет глобальной проблемы с списком, как в этом вопросе (python multiprocessing apply_async only uses one process), и я не думаю, что моя функция слишком тривиальна, как в этом случае (Python multiprocessing.Pool() doesn't use 100% of each CPU).

Я не могу понять, что я делаю неправильно, любая помощь/предложение приветствуются. Вот основной код:

import multiprocessing 
import itertools 

def myfunction(line): 
     returnlist=[] 
     list_of_elem=line.split(",") 
     elem_id=list_of_elem[1] 
     elem_to_check=list_of_elem[5] 

     ids=list_of_elem[2].split("|") 

     for x in itertools.permutations(ids,2): 
       if x[1] == elem_to_check: 
          returnlist.append(",".join([elem_id,x,"1\n"])) 
       else: 
          returnlist.append(",".join([elem_id,x,"0\n"])) 

     return returnlist  

def grouper(n, iterable, padvalue=None): 
    return itertools.izip_longest(*[iter(iterable)]*n, fillvalue=padvalue) 

if __name__ == '__main__': 
    my_data = open(r"my_input_file_to_be_processed.txt","r") 
    my_data = my_data.read().split("\n") 

    p = multiprocessing.Pool(4) 

    for chunk in grouper(1000, my_data): 
      results = p.map(myfunction, chunk) 
      for r in results: 
       with open (r"my_output_file","ab") as outfile: 
        outfile.write(r) 

EDIT Я изменил свой код, следуя предложениям (удаление избыточных данных, предварительной обработки). Однако проблема все еще существует.

import multiprocessing 
import itertools 

def myfunction(line): 
     returnlist=[] 
     list_of_elem=line.split(",") 
     elem_id=list_of_elem[1] 
     elem_to_check=list_of_elem[5] 

     ids=list_of_elem[2].split("|") 

     for x in itertools.permutations(ids,2): 
       if x[1] == elem_to_check: 
          returnlist.append(",".join([elem_id,x,"1\n"])) 
       else: 
          returnlist.append(",".join([elem_id,x,"0\n"])) 

     return returnlist  

if __name__ == '__main__': 
    my_data = open(r"my_input_file_to_be_processed.txt","r") 

    p = multiprocessing.Pool(4) 

    results = p.map(myfunction, chunk, chunksize=1000) 
     for r in results: 
      with open (r"my_output_file","ab") as outfile: 
       outfile.write(r) 

Answer 1

По вашему фрагмент кода, который я предполагаю, что я хотел бы сделать что-то вроде этого, чтобы кусок файла в 8 частей, а затем сделать вычисление было сделано 4 рабочих (почему 8 ломти и 4 рабочих? Просто случайный выбор, который я сделал для примера).

from multiprocessing import Pool 
import itertools 

def myfunction(lines): 
    returnlist = [] 
    for line in lines: 
     list_of_elem = line.split(",") 
     elem_id = list_of_elem[1] 
     elem_to_check = list_of_elem[5] 
     ids = list_of_elem[2].split("|") 

     for x in itertools.permutations(ids,2): 
      returnlist.append(",".join(
       [elem_id,x,"1\n" if x[1] == elem_to_check else "0\n"])) 

    return returnlist 

def chunk(it, size): 
    it = iter(it) 
    return iter(lambda: tuple(itertools.islice(it, size)),()) 

if __name__ == "__main__": 
    my_data = open(r"my_input_file_to_be_processed.txt","r") 
    my_data = my_data.read().split("\n") 

    prep = [strings for strings in chunk(my_data, round(len(my_data)/8))] 
    with Pool(4) as p: 
     res = p.map(myfunction, prep) 

    result = res.pop(0) 
    _ = list(map(lambda x: result.extend(x), res)) 
    print(result)  # ... or do something with the result 

Edit: Это предполагает, что вы уверены, все строки форматируются таким же образом и не вызовет ошибку.

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

def myfunction(lines): 
    returnlist = [] 
    for line in lines: 
     try: 
      list_of_elem = line.split(",") 
      elem_id = list_of_elem[1] 
      elem_to_check = list_of_elem[5] 
      ids = list_of_elem[2].split("|") 

      for x in itertools.permutations(ids,2): 
       returnlist.append(",".join(
        [elem_id,x,"1\n" if x[1] == elem_to_check else "0\n"])) 
     except Exception as err: 
      returnlist.append('I encountered error {} on line {}'.format(err, line)) 

    return returnlist