Ray 比 Python 和 .multiprocessing 慢得多

Question

我上传了 13 万个 json 文件。

我用Python 做这个：

import os
import json
import pandas as pd

path = "/my_path/"

filename_ending = '.json'


json_list = []

json_files = [file for file in os.listdir(f"{path}") if file.endswith(filename_ending)]

import time
start = time.time()

for jf in json_files:
    with open(f"{path}/{jf}", 'r') as f:

        json_data = json.load(f)

        json_list.append(json_data)

end = time.time()

这需要 60 秒。

我用multiprocessing 做这个：

import os
import json
import pandas as pd
from multiprocessing import Pool
import time

path = "/my_path/"

filename_ending = '.json'

json_files = [file for file in os.listdir(f"{path}") if file.endswith(filename_ending)]


def read_data(name):
    with open(f"/my_path/{name}", 'r') as f:
        json_data = json.load(f)

    return json_data


if __name__ == '__main__':

    start = time.time()

    pool = Pool(processes=os.cpu_count())                       
    x = pool.map(read_data, json_files)     

    end = time.time()

这需要 53 秒。

我用ray 做这个：

import os
import json
import pandas as pd
from multiprocessing import Pool
import time
import ray


path = "/my_path/"

filename_ending = '.json'

json_files = [file for file in os.listdir(f"{path}") if file.endswith(filename_ending)]

start = time.time()

ray.shutdown()
ray.init(num_cpus=os.cpu_count()-1)

@ray.remote    
def read_data(name):
    with open(f"/my_path/{name}", 'r') as f:
        json_data = json.load(f)

    return json_data

all_data = []
for jf in json_files:
    all_data.append(read_data.remote(jf))


final = ray.get(all_data)

end = time.time()

这需要 146 秒。

我的问题是为什么ray 需要这么多时间？

是不是因为：

1) 对于相对少量的数据，ray 相对较慢？

2) 我的代码做错了什么？

3) ray 不是很有用吗？

Answer 1

我从未使用过 ray，但我很有信心，我的解释应该是正确的。

原始代码做了一个简单的 json 反序列化。该代码主要需要文件 IO 和一点点 CPU。 （json反序列化比较快，这也是json成为流行交换格式的原因之一）

Ray 必须将数据从一个进程推送到另一个进程（如果通过网络分布在多台机器上）。为了做到这一点，它自己执行一些序列化/反序列化（也许它使用 pickle 和一个健壮的 TCP 协议来推送参数和收集结果）。并且可能这个开销比实际任务所花费的工作量更大。

如果您对 json 数据进行更多计算（任何更占用 CPU 的数据），那么您将能够看到差异。

我的猜测是，您的示例问题太简单了，因此 ray 的开销超过了使用多个工作人员的好处。

换句话说。分配任务和收集结果所花费的时间/精力比实际执行计算结果所花费的时间/精力更多。

Answer 2

我会说假设 1) 可能是最接近事实的。 Ray 看起来像一个强大的库，但你所做的只是读取一堆文件。您的代码只是为了进行基准测试，还是某个更大程序的一部分？如果是后者，那么让您的基准代码反映这一点可能会很有趣。

这没什么大不了的，但我调整了你的 3 个程序，所以它们至少应该更高效一些。

---

import os
import json


folder_path = "/my_path/"
filename_ending = '.json'

json_files = (os.path.join(folder_path, fp) for fp in os.listdir(f"{folder_path}") if fp.endswith(filename_ending))


def load_json_from_file(file_path):
    with open(file_path, 'r') as file_1:
        return json.load(file_1)


json_list = [load_json_from_file(curr_fp) for curr_fp in json_files]

---

import os
import json
import multiprocessing as mp


folder_path = "/my_path/"
filename_ending = '.json'

json_files = (os.path.join(folder_path, fp) for fp in os.listdir(f"{folder_path}") if fp.endswith(filename_ending))


def load_json_from_file(file_path):
    with open(file_path, 'r') as file_1:
        return json.load(file_1)


with mp.Pool() as pool:       
    json_list = pool.map(load_json_from_file, json_files)  

---

import os
import json
import ray

folder_path = "/my_path/"
filename_ending = '.json'


@ray.remote
def load_json_from_file(file_path):
    with open(file_path, 'r') as file_1:
        return json.load(file_1)


json_files = (os.path.join(folder_path, fp) for fp in os.listdir(f"{folder_path}") if fp.endswith(filename_ending))

ray.init()

futures_list = [load_json_from_file.remote(curr_fp) for curr_fp in json_files]

json_list = ray.get(futures_list)

---

如果您有任何问题，请告诉我。如果您可以再次运行基准测试，我很想知道有什么区别（如果有的话）。