如何解决使用Ray并行化大型程序的正确方法
我有一个相当大的Python程序(约800行),其结构如下:
- 设置说明,我在其中处理用户提供的输入文件,并定义将对程序执行全局的变量/对象。
- 主要功能,它利用上一个设置阶段并调用程序的主要附加功能。
- 从主函数直接调用它们的意义上讲,附加函数可以是主要函数,或者在仅由主附加函数调用的意义上讲,可以是辅助函数。
- 我在最后几行代码中处理main函数的结果。
该程序在很大程度上是并行的,因为主要功能的每次执行均独立于上一个和下一个。因此,我使用Ray在群集中的多个工作程序节点上并行执行主要功能。操作系统是CentOS Linux版本8.2.2004(核心),并且集群执行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警告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",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 thread中已经指出的那样,我应该在它们上调用ray.put
。那里的问题是我碰到了这些ValueError: buffer source array is read-only
,我不知道如何诊断。另外,我不确定是否应该用@ray.remote
装饰所有功能,还是只用main装饰。我想我可以对所有其他功能执行@ray.remote(num_cpus=1)
,因为它实际上应该只是并行执行的主要功能,但是我不知道这是否有意义。
我们非常感谢您的帮助,如果需要,我很乐意提供更多信息。
解决方法
我潜在地解决了我的问题,但是我不介意别人的意见,因为我的Ray知识确实有限。另外,我想这可能会给遇到类似问题的其他人提供帮助(希望我并不孤单!)。
正如我在问题中提到的那样,该程序对于足够小的数据集运行得很好(尽管它似乎绕过了Ray逻辑的某些方面),但最终在大型数据集上崩溃了。仅使用Ray任务,我没有设法调用存储在对象存储区(ValueError: buffer source array is read-only
中的Scipy Interpolator对象,并且装饰所有函数都没有意义,因为实际上只有主要的一个函数应同时执行(在调用时其他功能)。
因此,我决定更改程序的结构以使用Ray Actors。设置说明现在是__init__
方法的一部分。特别是,在此方法中定义了Scipy插值器对象,并将其设置为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上运行,没有任何警告或错误。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。