Имеет ли multiprocessing.pool.imap вариант (например, starmap), который допускает несколько аргументов?

Question

Я делаю некоторые вычисления в больших наборах байтов. Процесс выполняется на кусках байтов. Я пытаюсь использовать параллельную обработку с использованием многопроцессорности для повышения производительности. Первоначально я пытался использовать pool.map, но это допускает только один аргумент, затем я нашел о pool.starmap. Но pool.starmap дает результаты только после завершения всех процессов. Мне нужны результаты, когда они приходят (вроде). Я пытаюсь использовать pool.imap, который предоставляет результаты по завершении процессов, но не допускает множественные аргументы (моя функция требует 2 аргумента). Кроме того, важна последовательность результатов.

Некоторые примеры кода ниже:

pool = mp.Pool(processes=4) 
y = [] 
for x in pool.starmap(f, zip(da, repeat(db))): 
    y.append(x) 

Приведенный выше код работает, но только дает результаты, как только завершили все процессы. Я не вижу никакого прогресса. Вот почему я пытался использовать pool.imap, хорошо работает, но только с одним аргументом:

pool = mp.Pool(processes=4) 
y = [] 
for x in pool.imap(f, da)): 
    y.append(x) 

На нескольких аргументов вызывает следующее исключение:

TypeError: f() missing 1 required positional argument: 'd' 

Ищете простой способ достижения всех 3 требований :

1. параллельной обработки с использованием нескольких параметров/аргументы
2. удается увидеть прогресс в то время как процессы выполняются
3. Приказанные результаты.

Спасибо!

Answer 1

Я могу ответить на первые два вопроса довольно быстро. Я думаю, вы должны иметь возможность справиться с третьим вопросом после понимания первых двух.

1. Parrallel Обработка с несколькими аргументами

Я не уверен, что в целом «Starmap» равнозначны, но вот альтернатива. То, что я делал в прошлом, сводит мои аргументы к одному объекту данных, например списку. Например, если вы хотите передать три аргумента в свой map_function, вы можете добавить эти аргументы в список, а затем использовать список с помощью функции или .imap().

def map_function(combo): 
    a = combo[0] 
    b = combo[1] 
    c = combo[2] 
    return a + b + c 

if '__name__' == '__main__': 
    combo = [] 
    combo[0] = arg_1 
    combo[1] = arg_2 
    combo[2] = arg_3 

    pool = Pool(processes=4) 
    pool.map(map_function, combo) 

2. Отслеживание прогресса

Хороший способ сделать это с помощью multiprocessing «S общая стоимость. Я действительно спросил об этом (almost) same exact question около месяца назад. Это позволяет вам манипулировать одной и той же переменной из разных процессов, созданных вашей функцией map. Ради обучения, я позволю вам прочитать и разобраться в решении общего состояния самостоятельно. Если у вас все еще есть проблемы после нескольких попыток, я буду более чем счастлив помочь вам, но я верю, что научить себя понимать что-то гораздо ценнее, чем я, давая вам ответ.

Надеюсь, что это поможет!

Answer 2

Я думаю, что это решение точно соответствует вашим требованиям 3: https://stackoverflow.com/a/28382913/2379433

Короче говоря, p = Pool(); p.imap позволит вам видеть прогресс и поддерживать порядок. Если вам нужны функции map с несколькими аргументами, вы можете использовать вилку multiprocessing, которая обеспечивает лучшую сериализацию и несколько аргументов. См. Ссылку для примера.