Распараллеливание цикла for в Python

35

Есть ли в Python инструменты, которые похожи на парфор Matlab? Я нашел эту ветку , но ей четыре года. Я подумал, что, может быть, кто-то здесь может иметь более свежий опыт

Вот пример типа вещи, которую я хотел бы распараллелить:

X = np.random.normal(size=(10, 3))
F = np.zeros((10, ))
for i in range(10):
    F[i] = my_function(X[i,:])

где my_functionпринимает ndarrayразмер (1,3)и возвращает скаляр.

По крайней мере, я бы хотел использовать несколько ядер одновременно - как parfor. Другими словами, предположим, что система с общей памятью с 8-16 ядрами.

Пол Г. Константин
источник
Много результатов на Google. Это кажется довольно простым: blog.dominodatalab.com/simple-parallelization quora.com/What-is-the-Python-equivalent-of-MATLABs-parfor
Дуг Липински
Спасибо, @ Дуг-Липински. Эти примеры, как и другие, которые я нашел во время поиска в Google, имеют несколько тривиальных вычислений на основе индекса итерации. И они всегда утверждают, что код «невероятно прост». Мой пример определяет массивы (выделяет память) вне цикла for. Я в порядке, делаю это другим способом; это просто, как я делаю это в Matlab. Сложная часть, которая, кажется, разбивает эти примеры, заключается в получении части данного массива для функции внутри цикла.
Пол Г. Константин

Ответы:

19

Joblib делает то, что вы хотите. Основной шаблон использования:

from joblib import Parallel, delayed

def myfun(arg):
     do_stuff
     return result

results = Parallel(n_jobs=-1, verbose=verbosity_level, backend="threading")(
             map(delayed(myfun), arg_instances))

где arg_instancesсписок значений, для которых myfunвычисляется параллельно. Основное ограничение - myfunэто функция верхнего уровня. backendПараметр может быть либо "threading"или "multiprocessing".

Вы можете передать дополнительные общие параметры в распараллеленную функцию. Тело myfunтакже может ссылаться на инициализированные глобальные переменные, значения которых будут доступны детям.

Аргументы и результаты могут быть практически любыми с помощью потокового бэкэнда, но результаты должны быть сериализуемыми с многопроцессорным бэкэндом.


Dask также предлагает аналогичную функциональность. Это может быть предпочтительным, если вы работаете с основными данными или пытаетесь распараллелить более сложные вычисления.

Дэниел Малер
источник
Я вижу нулевую добавленную стоимость для использования батареи, включая многопроцессорность. Держу пари, что Joblib использует его под капотом.
Ксавье Комбель
1
Следует отметить, что joblib не является волшебством, серверная часть threadingстрадает от узкого места в GIL, а multiprocessingсерверная часть несет большие накладные расходы из-за сериализации всех параметров и возвращаемых значений. Смотрите этот ответ для низкоуровневых деталей параллельной обработки в Python.
Якуб Клинковский
Я не могу найти сочетание сложности функции и количества итераций, для которых joblib был бы быстрее, чем цикл for. Для меня он имеет такую ​​же скорость, если n_jobs = 1, и намного медленнее во всех остальных случаях
Алексей Фоминс
@AleksejsFomins Параллелизм, основанный на потоках, не поможет для кода, который не выпускает GIL, но делает это значительное количество, особенно для науки о данных или числовых библиотек. В противном случае вам потребуется мультипроцессинг, Jobli поддерживает оба. Многопроцессорный модуль теперь также имеет параллель, mapкоторый вы можете использовать напрямую. Также, если вы используете скомпилированный mkl numpy, он будет автоматически распараллеливать векторизованные операции без каких-либо действий. Numpy в Ananconda по умолчанию включен в MKL. Хотя не существует универсального решения. У Joblib очень низкий уровень шума, и в 2015 году было меньше опций.
Даниэль Малер
Спасибо за совет. Я помню, как пробовал многопроцессорность раньше и даже писал несколько постов, потому что они не масштабировались, как я ожидал. Может быть, мне стоит еще раз взглянуть
Алексей Фоминь
9

То, что вы ищете, это Numba , которая может автоматически распараллеливать цикл for. Из их документации

from numba import jit, prange

@jit
def parallel_sum(A):
    sum = 0.0
    for i in prange(A.shape[0]):
        sum += A[i]

    return sum
LKlevin
источник
8

Не предполагая что-то особенное в my_functionвыборе, multiprocessing.Pool().map()можно предположить распараллеливание таких простых циклов. joblib, dask, mpiВычисления или numbaкак предложено в других ответах выглядит не приносит никакой пользы для таких случаев использования и добавлять ненужные зависимости (суммировать их избыточна). Использование многопоточности, как предложено в другом ответе, вряд ли будет хорошим решением, потому что вы должны быть тесно связаны с GIL-взаимодействием вашего кода, или ваш код должен выполнять в основном ввод / вывод.

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

import multiprocessing
import numpy as np

if __name__ == "__main__":
   #the previous line is necessary under windows to not execute 
   # main module on each child under windows

   X = np.random.normal(size=(10, 3))
   F = np.zeros((10, ))

   pool = multiprocessing.Pool(processes=16)
   # if number of processes is not specified, it uses the number of core
   F[:] = pool.map(my_function, (X[i,:] for i in range(10)) )

Однако есть некоторые предостережения (но они не должны влиять на большинство приложений):

  • под окнами нет поддержки fork, поэтому при запуске каждого дочернего элемента запускается интерпретатор с основным модулем, поэтому он может иметь дополнительные издержки (и это является причиной if __name__ == "__main__"
  • Аргументы и результаты функции my_function зарезервированы и зарезервированы, это может быть слишком большой нагрузкой, посмотрите этот ответ, чтобы уменьшить его https://stackoverflow.com/a/37072511/128629 . Это также делает непринимаемые предметы непригодными для использования
  • my_functionне должен зависеть от общих состояний, таких как общение с глобальными переменными, потому что состояния не разделяются между процессами. чистые функции (функции в математическом смысле) являются примером функций, которые не разделяют состояния
Ксавье Комбель
источник
6

Мое впечатление от parfor заключается в том, что MATLAB инкапсулирует детали реализации, поэтому он может использовать как параллелизм разделяемой памяти (что вам нужно), так и параллелизм распределенной памяти (если вы используете сервер распределенных вычислений MATLAB ).

Если вам нужен параллелизм разделяемой памяти, и вы выполняете какой-то параллельный цикл задач, вероятно, вам нужен пакет стандартной библиотеки многопроцессорной обработки , возможно, с хорошим внешним интерфейсом , таким как joblib , как упомянуто в посте Дуга. Стандартная библиотека не собирается уходить, и она поддерживается, поэтому она с низким уровнем риска.

Есть и другие варианты, такие как параллельный Python и возможности IPython . Беглый взгляд на Parallel Python заставляет меня думать, что это ближе к духу parfor, поскольку библиотека инкапсулирует детали для распределенного случая, но цена этого заключается в том, что вам необходимо принять их экосистему. Стоимость использования IPython аналогична; Вы должны принять IPython способ ведения дел, который может или не может стоить этого для вас.

Если вы заботитесь о распределенной памяти, я рекомендую mpi4py . Lisandro Dalcin отлично работает, а mpi4py используется в оболочках PETSc Python, поэтому я не думаю, что он скоро исчезнет. Как и многопроцессорная, это интерфейс низкого (er) уровня параллелизма, чем parfor, но тот, который может длиться некоторое время.

Джефф Оксберри
источник
Спасибо, @Geoff. Есть ли у вас опыт работы с этими библиотеками? Может быть, я попробую использовать mpi4py на машине с общей памятью / многоядерном процессоре.
Павел Г. Константин
@PaulGConstantine Я успешно использовал mpi4py; это довольно безболезненно, если вы знакомы с MPI. Я не использовал многопроцессорность, но я порекомендовал его коллегам, которые сказали, что у них это хорошо работает. Я тоже использовал IPython, но не функции параллелизма, поэтому не могу сказать, насколько хорошо он работает.
Джефф Оксберри
1
У Арона есть хорошее руководство по mpi4py, которое он подготовил для курса PyHPC в Supercomputing: github.com/pyHPC/pyhpc-tutorial
Мэтт Кнепли,
4

Прежде чем искать инструмент «черного ящика», который можно использовать для параллельного выполнения «общих» функций Python, я бы предложил проанализировать, как my_function()можно распараллеливать вручную.

Во-первых, сравните время выполнения my_function(v)с forнакладными расходами цикла Python : [C] forЦиклы Python довольно медленные, поэтому затраченное время my_function()может быть незначительным.

>>> timeit.timeit('pass', number=1000000)
0.01692986488342285
>>> timeit.timeit('for i in range(10): pass', number=1000000)
0.47521495819091797
>>> timeit.timeit('for i in xrange(10): pass', number=1000000)
0.42337894439697266

Вторая проверка, если есть простая векторная реализация my_function(v), которая не требует циклов:F[:] = my_vector_function(X)

(Эти два первых пункта довольно тривиальны, простите, если я упомянул их здесь только для полноты.)

В- третьих , и самое главное, по крайней мере , для реализации CPython, чтобы проверить , является ли my_functionпроводит большую часть времени , это внутри или снаружи от глобальной блокировки интерпретатора или GIL . Если время проводится за пределами GIL, то следует использовать threadingстандартный библиотечный модуль . ( Вот пример). Кстати, можно было бы написать my_function()как расширение C просто для выпуска GIL.

Наконец, если my_function()не выпустить GIL, можно использовать multiprocessingмодуль .

Ссылки: документы Python по параллельному выполнению и введение numpy / scipy при параллельной обработке .

Стефано М
источник
2

Вы можете попробовать Юлю. Он довольно близок к Python и имеет множество конструкций MATLAB. Перевод здесь:

F = @parallel (vcat) for i in 1:10
    my_function(randn(3))
end

Это также делает случайные числа параллельными и просто объединяет результаты в конце во время сокращения. При этом используется многопроцессорная обработка (поэтому вам необходимо addprocs(N)добавить процессы перед использованием, и это также работает на нескольких узлах HPC, как показано в этом сообщении в блоге ).

Вы также можете использовать pmapвместо:

F = pmap((i)->my_function(randn(3)),1:10)

Если вам нужен параллелизм потоков, вы можете использовать его Threads.@threads(хотя убедитесь, что вы делаете алгоритм потокобезопасным). Перед открытием Джулии установите переменную среды JULIA_NUM_THREADS, а затем:

Ftmp = [Float64[] for i in Threads.nthreads()]
Threads.@threads for i in 1:10
    push!(Ftmp[Threads.threadid()],my_function(randn(3)))
end
F = vcat(Ftmp...)

Здесь я делаю отдельный массив для каждого потока, чтобы они не конфликтовали при добавлении в массив, а затем просто объединяли массивы. Потоки довольно новы, так что прямо сейчас есть только прямое использование потоков, но я уверен, что многопоточные сокращения будут добавлены в потоки и карты.

Крис Ракауцкас
источник
0

я рекомендую использовать параллельные и отложенные функции библиотеки joblib, использовать модуль «tempfile» для создания временной разделяемой памяти для больших массивов, примеры и использование которых можно найти здесь https://pythonhosted.org/joblib/parallel.html

Рамкумар
источник