Распределенные вычисления в питоне

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

ssh-keygen -trsa

После чего зайдите в свой подкаталог .ssh и скопируйте два файла - id_rsa.pub и authorized_keys - на удаленную машину.

Начнем мы с библиотеки:

DistributedPython

В этой библиотеке всего несколько файлов:

testSubmitMaster.py - исполняемый скрипт верхнего уровня, включающий в себя тот самый список команд. Он при своем выполнении будет генерировать логи в виде outputX.dat.

testScript - это shell-скрипт, который используется testSubmitMaster.py. Он содержит несколько переменных - id-шник процесса, имя хоста, время.

submitMaster.py - включает две функции - processCommandsInParallel и submitMaster.
Функция processCommandsInParallel вызывается из testSubmitMaster.py и делает следующее:
1. Создает процесс submitMaster
2. Назначает команды для submitMaster
3. Ожидает окончания работы submitMaster.
Коннект между submitMaster и функцией осуществляется с помощью канала Pipe, создается JobDistributor и очередь queue. В цикле:
прослушиваются каналы
проверяется очередь, если она освобождается, назначаются новые задачи

jobDistributor.py - класс содержит список доступных машин и назначает им задачи. Создается один инстанс JobDistributor. Он использует subprocess, с помощью которого создает новый процесс (Popen) для выполнения команды ssh. Информация о процессе хранится в обьекте Job, JobDistributor формирует словарь, в котором ключ - это имя хоста, а значение обьект Job. Этот скрипт нужно запустить отдельно.

listQueue.py - класс, отвечающий за работу очереди задач.

Batchlib

Модуль exec_proxy использует протокол ssh для коннекта с удаленной машиной, после чего позволяет выполнить на ней питоновскую программу. У этого модуля есть следующие две глобальные переменные:

 
 connection_program -  "ssh -x" (по умолчанию)
 remote_program - по умолчанию это полный путь к скрипту remote.py на удаленной машине
 
 

 
 newdir() - создает временную директорию для хранения файлов
 mkdir() - создает новый каталог на удаленной машине
 chdir()
 rmdir()
 getcwd()
 isdir()
 isfile()
 exists()
 listdir()
 chmod()
 stat()
 mkfifo() - создает именованный канал на удаленной машине
 symlink()
 unlink() - удаляет файл
 upload() - загружает файл с локальной на удаленную машину
 download() - загружает файл судаленной на локальную машину
 write() - создает файл
 read()
 run() - выполняет команду, которая выступает в качестве параметра 
 
 

 
 is_ok() - возвращает 1 в случае ошибки
 get_error() - возвращает саму ошибку
 reset_error() - 
 
 

 import exec_proxy
 import os,stat
 
 exec_proxy.remote_program=os.path.join(os.getcwd(),'.','remote.py')
 ep=exec_proxy.Exec_proxy('user@remote_host')
 print ep.listdir('./');
 

dispy

Этот модуль можно использовать на локальной много-процессорной машине(SMP), а также в кластере. Хорошо подходит для распараллеленных вычислений, где вычислительные задачи не связаны между собой. Dispy использует asyncoro - фреймворк, построенный на асинхронных сообщениях, использующий неблокирующие сокеты, epoll, Поддерживается передача вычислений, файлов, сообщений. В качестве вычислений могут выступать как функции, так и отдельные программы. Ноды могут располагаться как в локальной сети, так и в удаленной, в последнем случае можно использовать ssh. Ноды можно открывать динамически: в модуле есть шедулер, который контролирует работу нод в кластере и может перераспределять задания. Доступен механизм callback. Установить dispy можно, скачав пакет исходников, либо с помощью команды pip3 install dispy.

Dispy состоит из 4 основных компонентов или собственно модулей :
1. dispy.py - базовая библиотека классов для cоздания кластеров. Есть 2 механизма создания кластеров - JobCluster либо SharedJobCluster. В первом случае используется встроенный шедулер, во втором случае должен быть создан отдельный шедулер с помощью другого модуля - dispyscheduler.py

2. dispynode.py - устанавливается на каждой кластерной ноде и выполняет там всю работу

3. dispyscheduler.py - шедулер, который может управлять работой нод самостоятельно.

4. dispynetrelay.py - нужен в случае использования сложной сетевой конфигурации. Если все ноды кластера перчислены в качестве параметра на этапе его создания, этот модуль не нужен. Он может получать информацию о дополнительных нодах в процессе работы.

Теперь рассмотрим простой пример. Создадим на локальной машине кластер из 5 нод, каждая из которых будет выполнять одну и ту же функцию compute. Под локальной нодой мы понимаем отдельный питоновский процесс. Каждой ноде будет передаваться случайное число в диапазоне от 1 до 10 секунд, в течение которых каждая нода будет находиться в паузе, после чего каждая нода будет возвращать собственный id-шник. В конце основной сценарий будет распечатывать протокол ответов. Основной сценарий.

 def compute(n):
     import time, socket
     time.sleep(n)
     host = socket.gethostname()
     return (host, n)
 
 if __name__ == '__main__':
     import dispy, random
     cluster = dispy.JobCluster(compute, nodes=['localhost'])
     jobs = []
     for n in range(5):
         job = cluster.submit(random.randint(1,10))
         job.id = n
         jobs.append(job)
     # cluster.wait()
     for job in jobs:
         host, n = job()
         print '%s executed job %s at %s with %s' % (host, job.id, job.start_time, n)
     cluster.stats()
 

dispynode.py -d -i localhost

После этого запускаем основной сценарий, который выведет что-то типа:

 localhost executed job 0 at 1414068881.77 with 19
 localhost executed job 1 at 1414068881.77 with 12
 localhost executed job 2 at 1414068893.8 with 20
 localhost executed job 3 at 1414068900.81 with 11
 localhost executed job 4 at 1414068911.84 with 18
 ()
                            Node |  CPUs |    Jobs |    Sec/Job | Node Time Sec
 ------------------------------------------------------------------------------
  127.0.0.1 (localhost)          |     2 |       5 |     16.020 |        80.100
 
 

Оставьте свой комментарий !
Ваше имя: Комментарий: Оба поля являются обязательными  Автор   Комментарий к данной статье