python - 如何使用 python Ray 并行处理一个大列表？

Question

我想使用 ray 在列表的每个元素上并行化函数的操作.下面是一个简化的片段

import numpy as np
import time

import ray
import psutil
num_cpus = psutil.cpu_count(logical=False)
ray.init(num_cpus=num_cpus)


@ray.remote
def f(a, b, c):
    return a * b - c


def g(a, b, c):
    return a * b - c


def my_func_par(large_list):
    # arguments a and b are constant just to illustrate
    # argument c is is each element of a list large_list
    [f.remote(1.5, 2, i) for i in large_list]


def my_func_seq(large_list):
    # arguments a anf b are constant just to illustrate
    # argument c is is each element of a list large_list
    [g(1.5, 2, i) for i in large_list]

my_list = np.arange(1, 10000)

s = time.time()
my_func_par(my_list)
print(time.time() - s)
>>> 2.007

s = time.time()
my_func_seq(my_list)
print(time.time() - s)
>>> 0.0372

问题是，当我计时 my_func_par ，它比 my_func_seq 慢得多(如上所示~54 倍) . ray 的一位作者确实回答了关于 this blog 的评论这似乎解释了我正在做的是设置 len(large_list)不同的任务，这是不正确的。
如何使用 ray 并修改上面的代码以并行运行它？ (也许通过将 large_list 分成块，块的数量等于 CPU 的数量)
编辑:这个问题有两个重要的标准

函数f需要接受多个参数

可能需要使用 ray.put(large_list)使larg_list变量可以存储在共享内存中而不是复制到每个处理器

Answer 1

补充一下 Sang 上面所说的:
雷Distributed multiprocessing.Pool支持固定大小的 Ray Actors 池，以便于并行化。

import numpy as np
import time

import ray
from ray.util.multiprocessing import Pool
pool = Pool()

def f(x):
    # time.sleep(1)
    return 1.5 * 2 - x

def my_func_par(large_list):
    pool.map(f, large_list)

def my_func_seq(large_list):
    [f(i) for i in large_list]

my_list = np.arange(1, 10000)

s = time.time()
my_func_par(my_list)
print('Parallel time: ' + str(time.time() - s))

s = time.time()
my_func_seq(my_list)
print('Sequential time: ' + str(time.time() - s))

使用上面的代码， my_func_par运行速度更快(约 0.1 秒)。如果您玩弄代码并制作 f(x)慢一些，比如 time.sleep ，您可以看到多处理的明显优势。