具有分布式集群的 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 和pathos fork 的pp。此外，对于 ssh，您需要运行 ssh-agent 以允许使用 ssh 进行无密码登录。

但是，希望一切正常……如果您的功能代码可以使用dill.source.importable 传输到远程主机。

仅供参考，pathos 早就应该发布了，基本上，在新的稳定版本被删除之前，有一些错误和界面更改需要解决。

Answer 2

我建议看看 Ray，它的目的就是为了做到这一点。

Ray 在单机多核设置中使用与在分布式设置中相同的语法来并行化代码。如果您愿意使用 for 循环而不是 map 调用，那么您的示例将如下所示。

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。

Answer 3

在这里聚会有点晚了，但由于我也在寻找类似的解决方案，而这个问题仍未标记为已回答，我想我会贡献我的发现。

我最终使用了SCOOP。它提供了一个并行映射实现，可以跨多个内核、跨多个主机工作。如果在调用期间需要，它还可以回退到 Python 的串行 map 函数。

从SCOOP的介绍页面中，它引用了以下特点：

SCOOP 相对于期货、多处理和 类似的模块如下：

• 通过网络利用多台计算机的功能；
• 能够在一个任务中生成多个任务；
• API 与PEP-3148 兼容；
• 只需少量修改即可并行化串行代码；
• 高效的负载平衡。

它确实有一些怪癖（函数/类必须是可腌制的），如果它们不都共享相同的文件系统架构，那么让事情在多个主机上顺利运行的设置可能会很乏味，但总的来说我很高兴结果。为了我们的目的，做了很多 Numpy 和 Cython，它提供了出色的性能。

希望这会有所帮助。

Answer 4

你看过disco吗？

特点：

• Map / Reduce 范例
• Python 编程
• 分布式共享磁盘
• ssh 底层传输
• Web 和控制台界面
• 轻松添加/阻止/删除节点
• master 在没有用户干预的情况下启动 slaves 节点
• slave 节点在发生故障时会自动重启
• 很好的文档。在Install Guide 之后，我能够在几分钟内启动一个 2 机集群（我唯一需要做的就是创建 $DISCO_HOME/root 文件夹以连接到 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)