python - 使用 Ray 并行化大型程序的正确方法
问题描述
我有一个相当大的 Python 程序(约 800 行),其结构如下:
- 设置说明,我在其中处理用户提供的输入文件并定义对程序执行全局的变量/对象。
- Main 函数,它利用前面的设置阶段并调用程序的主要附加函数。
- 附加功能,可以是主要的,因为它们被主要功能直接调用,也可以是次要的,因为它们只被主要的附加功能调用。
- 最后几行代码,我在其中处理 main 函数的结果。
该程序是大规模并行的,因为主函数的每次执行都独立于前一个和下一个。因此,我使用 Ray 在集群中的多个工作节点上并行执行主要功能。操作系统为 CentOS Linux release 8.2.2004 (Core),集群执行 PBS Pro 19.2.4.20190830141245。我正在使用 Python 3.7.4、Ray 0.8.7 和 Redis 3.4.1。
我在 Python 脚本中有以下内容,foo
主要功能在哪里:
@ray.remote(memory=2.5 * 1024 * 1024 * 1024)
def foo(locInd):
# Main function
if __name__ == '__main__':
ray.init(address='auto', redis_password=args.pw,
driver_object_store_memory=10 * 1024 * 1024 * 1024)
futures = [foo.remote(i) for i in zip(*np.asarray(indArr == 0).nonzero())]
waitingIds = list(futures)
while len(waitingIds) > 0:
readyIds, waitingIds = ray.wait(
waitingIds, num_returns=min([checkpoint, len(waitingIds)]))
for r0, r1, r2, r3, r4, r5, r6, r7 in ray.get(readyIds):
# Process results
indArr[r0[::-1]] = 1
nodesComplete += 1
ray.shutdown()
以下是我用来启动 Ray 的说明
# Head node
/path/to/ray start --head --port=6379 \
--redis-password=$redis_password \
--memory $((120 * 1024 * 1024 * 1024)) \
--object-store-memory $((20 * 1024 * 1024 * 1024)) \
--redis-max-memory $((10 * 1024 * 1024 * 1024)) \
--num-cpus 48 --num-gpus 0
# Worker nodes
/path/to/ray start --block --address=$1 \
--redis-password=$2 --memory $((120 * 1024 * 1024 * 1024)) \
--object-store-memory $((20 * 1024 * 1024 * 1024)) \
--redis-max-memory $((10 * 1024 * 1024 * 1024)) \
--num-cpus 48 --num-gpus 0
只要我在足够小的数据集上工作,一切都会按预期运行。尽管如此,执行会产生以下警告
- 2020-08-17 17:16:44,289 警告 worker.py:1134 -- 警告:远程函数
__main__.foo
在腌制时的大小为 220019409。它将存储在 Redis 中,这可能会导致内存问题。这可能意味着它的定义使用了一个大数组或其他对象。 - 2020-08-17 17:17:10,281 WARNING worker.py:1134 -- 该工作者被要求执行一个它没有注册的函数。您可能需要重新启动 Ray。
如果我尝试在更大的数据集上运行代码,我会收到以下错误:
Traceback (most recent call last):
File "/home/157/td5646/.local/lib/python3.7/site-packages/redis/connection.py", line 700, in send_packed_command
sendall(self._sock, item)
File "/home/157/td5646/.local/lib/python3.7/site-packages/redis/_compat.py", line 8, in sendall
2020-08-21 14:22:34,226 WARNING worker.py:1134 -- Warning: The remote function __main__.foo has size 898527351 when pickled. It will be stored in Redis, which could cause memory issues. This may mean that its definition uses a large array or other object.
return sock.sendall(*args, **kwargs)
ConnectionResetError: [Errno 104] Connection reset by peer
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "./Program.py", line 1030, in <module>
for i in zip(*np.asarray(indArr == 0).nonzero())]
File "./Program.py", line 1030, in <listcomp>
for i in zip(*np.asarray(indArr == 0).nonzero())]
File "/home/157/td5646/.local/lib/python3.7/site-packages/ray/remote_function.py", line 95, in _remote_proxy
return self._remote(args=args, kwargs=kwargs)
File "/home/157/td5646/.local/lib/python3.7/site-packages/ray/remote_function.py", line 176, in _remote
worker.function_actor_manager.export(self)
File "/home/157/td5646/.local/lib/python3.7/site-packages/ray/function_manager.py", line 152, in export
"max_calls": remote_function._max_calls
File "/home/157/td5646/.local/lib/python3.7/site-packages/redis/client.py", line 3023, in hmset
return self.execute_command('HMSET', name, *items)
File "/home/157/td5646/.local/lib/python3.7/site-packages/redis/client.py", line 877, in execute_command
conn.send_command(*args)
File "/home/157/td5646/.local/lib/python3.7/site-packages/redis/connection.py", line 721, in send_command
check_health=kwargs.get('check_health', True))
File "/home/157/td5646/.local/lib/python3.7/site-packages/redis/connection.py", line 713, in send_packed_command
(errno, errmsg))
redis.exceptions.ConnectionError: Error 104 while writing to socket. Connection reset by peer.
关于我如何向 Ray 描述程序,我显然做错了。我有我认为是全局的 Scipy Interpolator 对象,但是,正如在这个 GitHub线程中已经指出的那样,我应该调用ray.put
它们。那里的问题是我遇到了这些ValueError: buffer source array is read-only
我不知道如何诊断的问题。另外,我不确定是否应该使用@ray.remote
或仅使用主要功能来装饰所有功能。我想我可以@ray.remote(num_cpus=1)
为所有附加功能做,因为它真的应该是并行执行的主要功能,但我不知道这是否有意义。
非常感谢任何帮助,如果需要,我很乐意提供更多信息。
解决方案
我可能已经解决了我的问题,但我不介意别人的意见,因为我的 Ray 知识确实有限。另外,我想这可能会帮助遇到类似问题的其他人(希望我并不孤单!)。
正如我在问题中提到的,该程序对于足够小的数据集运行得相当好(即使它似乎绕过了 Ray 逻辑的某些方面),但它最终在大型数据集上崩溃了。仅使用 Ray 任务,我没有设法调用存储在 Object Store ( ValueError: buffer source array is read-only
) 中的 Scipy Interpolator 对象,并且装饰所有函数没有意义,因为实际上只有主要的函数应该同时执行(同时调用其他函数)。
因此,我决定更改程序的结构以使用 Ray Actor。设置说明现在是该__init__
方法的一部分。特别是,Scipy Interpolator 对象在此方法中定义并设置为 的属性self
,就像全局变量一样。大多数函数(包括主函数)都变成了类方法,但通过 Numba 编译的函数除外。对于后者,它们仍然是用 装饰的独立函数@jit
,但现在它们中的每一个在调用 jitted 函数的类中都有一个等效的包装方法。
为了让我的程序并行执行我现在的 main 方法,我依赖于 ActorPool。我创建了与可用 CPU 一样多的 actor,每个 actor 执行 main 方法,成功调用方法和 Numba 编译的函数,同时还设法访问 Interpolator 对象。我只适用@ray.remote
于定义的 Python 类。所有这些都转化为以下结构:
@ray.remote
class FooClass(object):
def __init__(self, initArgs):
# Initialisation
@staticmethod
def exampleStaticMethod(args):
# Processing
return
def exampleMethod(self, args):
# Processing
return
def exampleWrapperMethod(self, args):
return numbaCompiledFunction(args)
def mainMethod(self, poolMapArgs):
# Processing
return
@jit
def numbaCompiledFunction(args):
# Processing
return
ray.init(address='auto', redis_password=redPass)
actors = []
for actor in range(int(ray.cluster_resources()['CPU'])):
actors.append(FooClass.remote(initArgs))
pool = ActorPool(actors)
for unpackedTuple in pool.map_unordered(
lambda a, v: a.mainMethod.remote(v),
poolMapArgs):
# Processing
ray.shutdown()
这在分布在 4 个节点上的 192 个 CPU 上成功运行,没有任何警告或错误。
推荐阅读
- django - Django表单未在模板中呈现输入字段
- postgresql - 如何更新postgresql中的jsonb列,它只是一个值数组,没有键
- mysql - 如何获取列数
- ios - iOS 中的 kCMTimeIndefinite 是什么?
- android - 如何在 viewpager 中添加片段和活动,以便片段出现在滑动时?
- c# - 如何修复异常“请求 System.Security.Permissions.EnvironmentPermission ... 类型的权限失败”
- javascript - 即使在使用 javascript 在 wordpress 中单击 Ok 后,如何停止进入下一个 POP 框?
- ios - iOS - 在后台模式下不扫描 BLE 外围设备
- c# - Close() 方法调用 WPF C# 时不关闭窗口
- html - 奇怪在 Laravel 中嵌套了一个标签问题