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

Ответ 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 и вилка pathos pp работающая на каждом хосте. Кроме того, для ssh вам нужно запустить ssh-agent, чтобы разрешить вход без пароля с помощью ssh.

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

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

Ответ 2

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

Я закончил использовать SCOOP. Он обеспечивает параллельную реализацию карты, которая может работать через несколько ядер, через несколько хостов. При желании он также может вернуться к функции последовательной map Python во время вызова.

На странице введения SCOOP приводятся следующие возможности:

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

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

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

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

Ответ 3

Я бы посоветовал взглянуть на Рэя, цель которого - сделать именно это.

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(). Соответствующую документацию можно найти здесь.

Заметьте, что я помогаю развить Рэй.

Ответ 4

Вы смотрели на дискотеку?

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

• Карта/Уменьшить парадигму
• Программирование на Python
• Распределенный общий диск
• Транспортировка ssh
• веб-интерфейс и консольные интерфейсы
• легко добавлять/блокировать/удалять узел
• мастер запускает ведомые узлы без вмешательства пользователя
• узлы ведомых автоматически перезапускаются в случае отказа
• хорошая документация. Следуя Руководству по установке, я смог запустить кластер с двумя компьютерами за несколько минут (мне нужно было создать $ 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)