python—使用ray并行化大型程序的正确方法

我有一个相当大的python程序（约800行），其结构如下：
设置指令，其中我处理用户提供的输入文件，并定义对程序执行全局的变量/对象。
main函数，它利用前面的设置阶段并调用程序的主要附加函数。
附加函数，可以是主函数直接调用的主函数，也可以是次函数，仅由主附加函数调用。
我处理主函数结果的最后几行代码。
程序是大规模并行的，因为主函数的每次执行都独立于上一次和下一次执行。因此，我使用ray在集群中的多个工作节点上并行执行main函数。操作系统为centos linux 8.2.2004版（core），集群执行pbs pro 19.2.4.20190830141245。我使用的是python3.7.4、ray0.8.7和redis3.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:44289警告工人.py:1134--警告：远程功能 __main__.foo 腌制时尺寸为220019409。它将存储在redis中，这可能会导致内存问题。这可能意味着它的定义使用了一个大数组或其他对象。
2020-08-17 17:17:10281警告worker.py:1134--此worker被要求执行其未注册的函数。你可能得重新启动雷。
如果尝试在较大的数据集上运行代码，则会出现以下错误：

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逻辑的一些方面），但它最终在大型数据集上崩溃了。仅使用光线任务，我无法调用存储在对象存储中的scipy interpolator对象( ValueError: buffer source array is read-only )，并且装饰所有函数没有意义，因为实际上只有主函数应该并发执行（同时调用其他函数）。
因此，我决定改变程序的结构，使用光线演员。安装说明现在是 __init__ 方法。特别是，scipy插值器对象在该方法中定义并设置为的属性 self ，因为全局变量是。除了通过numba编译的函数外，大多数函数（包括主函数）都变成了类方法。对于后者，它们仍然是用 @jit ，但是现在它们中的每一个在类中都有一个等价的 Package 器方法来调用jitted函数。
为了让我的程序并行执行我现在的main方法，我依赖于actorpool。我创建了尽可能多的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上成功运行，没有任何警告或错误。

python—使用ray并行化大型程序的正确方法

1条答案

相关问题

热门标签

最新问答