Парализуйте функцию, используя многопроцессорность в python

Question

У меня есть функция с именем new_function(df,y) с двумя аргументами df и y, и я пытаюсь применить ее к списку из 100 фреймов данных, используя multiprocessing, и я хочу добавить результаты в список. Я прошел предыдущие посты, но не мог понять, я пытаюсь этот метод, но не могу обработать.

import multiprocessing as mp
pool = mp.Pool(processes = mp.cpu_count())
results = [pool.map(new_function, df) for df in data]
pool.close() 
pool.join()

Answer 1

В вашем примере у вас есть

results = [pool.map(new_function, df) for df in data]

Это не будет работать так, как вы ожидаете, так как второй аргумент pool.map должен быть итеративным для элементов, которые вы хотите выполнить new_function.

Вы должны иметь возможность запустить new_function параллельно, используя следующий подход. Этот код использует functools.partial для генерации «частичной функции», то есть с константным параметром y, заполненным с использованием CONSTANT. pool.map выполнит эту функцию для каждого из DataFrame s, содержащихся в data.

import functools
import multiprocessing as mp

if __name__ == "__main__":
    with mp.Pool(processes = mp.cpu_count()) as pool:
        results = pool.map(functools.partial(new_function, y=CONSTANT), data)

    print(results)

Поскольку у меня нет ваших входных данных, вы можете увидеть, как это работает с использованием фиктивных входов и простого new_function:

def new_function(x, y):                                                         
    return x, y                                                                 

if __name__ == "__main__":                                                      
    data = [1, 2, 3, 4]                                                         
    CONSTANT = 12                                                               

    with mp.Pool(processes = mp.cpu_count()) as pool:                           
        results = pool.map(functools.partial(new_function, y=CONSTANT), data)   

    print(results)

Вывод:

[(1, 12), (2, 12), (3, 12), (4, 12)]

Если параметр y изменяется с data, т.е. у вас есть список значения, соответствующие различным DataFrame s в data, вместо этого можно использовать pool.starmap:

def new_function(x, y):                                                         
    return x, y

if __name__ == "__main__":
    # Sample data                                                    
    data = [1, 2, 3, 4]                                                         
    y_list = [5, 6, 7, 8]                                                       

    with mp.Pool(processes = mp.cpu_count()) as pool:                           
        results = pool.starmap(new_function, zip(data, y_list))                 

    print(results)

Выход:

[(1, 5), (2, 6), (3, 7), (4, 8)]