Наиболее эффективный способ многопроцессорной обработки отдельных функций над одним и тем же объектом.

Question

В скрипте Python у меня есть большой набор данных, к которому я хотел бы применить несколько функций. Функции отвечают за создание определенных выходных данных, которые сохраняются на жесткий диск.

Несколько замечаний:

1. функции независимы
2. ни одна из функций ничего не возвращает
3. функции будут занимать различное количество времени
4. некоторые функции могут не работать, и это нормально

Могу ли я выполнить многопроцессорную обработку таким образом, чтобы каждая функция и набор данных отправлялись в ядро ​​отдельно и выполнялись там? Таким образом, мне не нужно, чтобы первая функция заканчивалась, прежде чем вторая может сработать? Нет необходимости в их последовательной зависимости. Спасибо!

Answer 1

Я был в той же ситуации, что и вы, и использовал Процесс es со следующей функцией:

import multiprocessing as mp

def launch_proc(nproc, lst_functions, lst_args, lst_kwargs):
    n = len(lst_functions)
    r = 1 if n % nproc > 0 else 0
    for b in range(n//nproc + r):
        bucket = []
        for p in range(nproc):
             i = b*nproc + p
             if i == n:
                 break
             proc = mp.Process(target=lst_functions[i], args=lst_args[i], kwargs=lst_kwargs[i])
             bucket.append(proc)
        for proc in bucket:
            proc.start()
        for proc in bucket:
            proc.join()

Это имеет главный недостаток: все процессы в корзине должны завершитьсяпрежде чем новое ведро может начаться.Я пытался использовать JoinableQueue , чтобы избежать этого, но не смог заставить его работать.

Пример:

def f(i):
    print(i)

nproc = 2
n     = 11
lst_f      = [f] * n
lst_args   = [[i] for i in range(n)]
lst_kwargs = [{}] * n
launch_proc(nproc, lst_f, lst_args, lst_kwargs)

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

Answer 2

Поскольку ваши функции являются независимыми и читают только данные, при условии, что это не проблема, если ваши данные изменяются во время выполнения функции, они также являются поточно-ориентированными.

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

Примечание. Чтобы она работала более чем на одном ядре, вы должны использовать Python Multiprocessing .В противном случае все потоки будут работать на одном ядре.Это происходит потому, что Python имеет глобальную блокировку интерпретатора (GIL).Для получения дополнительной информации Все потоки Python выполняются на одном ядре

В качестве альтернативы вы можете использовать DASK , который дополняет данные для выполнения многопоточности.Добавляя некоторые накладные расходы, это может быть быстрее для ваших нужд.