Многопроцессор Python с распределенным кластером

Question

Я ищу пакет python, который может выполнять многопроцессорную обработку не только на разных ядрах на одном компьютере, но и на кластере, распределенном на нескольких компьютерах. Существует множество различных пакетов python для распределенных вычислений, но для большинства, похоже, требуется изменение кода для запуска (например, префикс, указывающий, что объект находится на удаленной машине). В частности, я хотел бы как можно ближе к многопроцессорной функции pool.map. Так, например, если на одном компьютере сценарий:

from multiprocessing import Pool 
pool = Pool(processes = 8) 
resultlist = pool.map(function, arglist) 

Тогда псевдокод для распределенного кластера будет:

from distprocess import Connect, Pool, Cluster 

pool1 = Pool(processes = 8) 
c = Connect(ipaddress) 
pool2 = c.Pool(processes = 4) 
cluster = Cluster([pool1, pool2]) 
resultlist = cluster.map(function, arglist) 

Answer 1

Если вы хотите очень простое решение, не существует один.

Однако есть решение, имеющее интерфейс multiprocessing - pathos - который имеет возможность устанавливать соединения с удаленными серверами через параллельную карту и выполнять многопроцессорную обработку.

Если вы хотите подключиться к ssh-туннелированию, вы можете это сделать ... или если вы в порядке с менее безопасным методом, вы тоже можете это сделать.

>>> # establish a ssh tunnel 
>>> from pathos.core import connect 
>>> tunnel = connect('remote.computer.com', port=1234) 
>>> tunnel  
Tunnel('-q -N -L55774:remote.computer.com:1234 remote.computer.com') 
>>> tunnel._lport 
55774 
>>> tunnel._rport 
1234 
>>> 
>>> # define some function to run in parallel 
>>> def sleepy_squared(x): 
... from time import sleep 
... sleep(1.0) 
... return x**2 
... 
>>> # build a pool of servers and execute the parallel map 
>>> from pathos.pp import ParallelPythonPool as Pool 
>>> p = Pool(8, servers=('localhost:55774',)) 
>>> p.servers 
('localhost:55774',) 
>>> y = p.map(sleepy_squared, x) 
>>> y 
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81] 

Или, вместо того, чтобы вы могли настроить для прямого подключения (без SSH)

>>> p = Pool(8, servers=('remote.computer.com:5678',)) 
# use an asynchronous parallel map 
>>> res = p.amap(sleepy_squared, x) 
>>> res.get() 
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81] 

Это все немного привередливы, для удаленного сервера, чтобы работать, вы должны начать сервер, работающий на remote.computer.com в указанном порту заранее - и вы должны убедиться, что оба параметра на вашем локальном хосте и удаленном хосте позволят либо прямое соединение, либо соединение с ssh-туннелем. Кроме того, у вас должна быть та же версия pathos и из вилки pathospp, работающей на каждом хосте. Кроме того, для ssh вам нужно запустить ssh-agent, чтобы разрешить вход без пароля с помощью ssh.

Но тогда, надеюсь, все это сработает ... если ваш код функции можно передать на удаленный хост с помощью dill.source.importable.

FYI, pathos давно назрел выпуск, и в основном есть несколько ошибок и изменений интерфейса, которые необходимо решить до того, как будет разрезан новый стабильный релиз.

Answer 2

Немного поздно на вечеринку здесь, но так как я тоже искал аналогичное решение, и этот вопрос по-прежнему не отмечен как ответ, я думал, что внес свои выводы.

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

С вводной странице совок, он приводит следующие особенности:

особенности совок и преимущества по сравнению с фьючерсами, многопроцессорной и подобные модули являются следующие:

• Harness мощность нескольких компьютеров через сеть;
• Возможность запуска нескольких задач внутри задачи;
• API, совместимый с PEP-3148;
• Параллельный последовательный код с незначительными изменениями;
• Эффективная балансировка нагрузки.

Он имеет некоторые особенности (функции/классы должны быть pickleable), и установка, чтобы получить вещи работают плавно через несколько хостов могут быть утомительными, если они не все придерживаются той же файловой системы схемы, но в целом Я очень доволен результатами. Для наших целей, делая немало Numpy & Cython, он обеспечивает отличную производительность.

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

Answer 3

Вы искали disco?

Особенности:

• Map/Reduce парадигмы
• Python программирования
• Распределенная общий диск
• SSH нижележащий транспорт
• веб и консольные интерфейсы
• легко добавлять/блок/удалить узел
• мастер запуск раб узлов без вмешательства пользователя
• раб узлов автоматически перезапускаются в случае сбоя
• хорошей документации. После Install Guide я смог запустить кластер с двумя машинами за несколько минут (единственное, что мне нужно было сделать, это создать $ DISCO_HOME/корневую папку, чтобы подключиться к WebUI, я думаю, из-за создания ошибки в файле журнала).

Простой пример из документации дискотеки:

from disco.core import Job, result_iterator 

def map(line, params): 
    for word in line.split(): 
     yield word, 1 

def reduce(iter, params): 
    from disco.util import kvgroup 
    for word, counts in kvgroup(sorted(iter)): 
     yield word, sum(counts) 

if __name__ == '__main__': 
    job = Job().run(input=["http://discoproject.org/media/text/chekhov.txt"], 
        map=map, 
        reduce=reduce) 
    for word, count in result_iterator(job.wait(show=True)): 
     print(word, count) 

Answer 4

я предлагаю взглянуть на Ray, целью которого является сделать именно это.

Ray использует тот же синтаксис для параллелизации кода в многоуровневой настройке одной машины, как в распределенной настройке. Если вы хотите использовать цикл for вместо вызова карты, ваш пример будет выглядеть следующим образом.

import ray 
import time 

ray.init() 

@ray.remote 
def function(x): 
    time.sleep(0.1) 
    return x 

arglist = [1, 2, 3, 4] 

result_ids = [function.remote(x) for x in arglist] 
resultlist = ray.get(result_ids) 

Это будет выполнять четыре задачи параллельно, используя, как и многие другие ядра, локальные. Чтобы запустить тот же пример в кластере, единственной строкой, которая будет изменена, будет вызов ray.init(). Соответствующая документация может быть найдена here.

Обратите внимание, что я помогаю разработать Ray.