使用 ProcessPoolExecutor 将数据拆分为较小的部分时的一些问题

Question

我正在尝试将大文件拆分为较小的部分。我将首先从大文件中读取所有数据，然后使用多处理将它们写入不同的较小数据文件。

这里是方法一，这里是使用multiprocessing.Process来初始化一个进程，效果很好

def split_data_with_process(filepath, chunk_num):
    def write(write_data, index, name, prev_path, suffix):
        print("enter")
        start_time = time.time()
        with open(prev_path + f"/{name}_{index}.{suffix}", "w", encoding="utf-8") as f:
            f.writelines(write_data)
        print(time.time()-start_time)

    prev_path, filename = filepath.rsplit("/", maxsplit=1)
    name, suffix = filename.split(".")
    with open(filepath, "r", encoding="utf-8") as f:
        totalList = f.readlines()
    chunksize = math.ceil(len(totalList) / chunk_num)
    data = [(totalList[start:start + chunksize], index) for index, start in
            enumerate(range(0, len(totalList), chunksize))]
    tasks = []
    start_time = time.time()
    for each in data:
        task = multiprocessing.Process(target=write, args=(each[0], each[1], name, prev_path, suffix))
        task.start()
        tasks.append(task)
    for each in tasks:
        each.join()
    end_time = time.time()
    print(end_time - start_time)

输出是

enter
enter
enter
enter
enter
7.192562818527222
8.827389001846313
9.067991018295288
9.476916313171387
7.729929208755493
15.109729290008545

然后我尝试用ProcessPoolExecutor重写代码，代码就像

def write(input):
    list, index, prev_path, name, suffix = input
    print("enter")
    start_time = time.time()
    with open(prev_path + f"/{name}_{index}.{suffix}", "w", encoding="utf-8") as f:
        f.writelines(list)
    print(time.time() - start_time)
    return len(list)


def split_data_with_process_2(filepath, chunk_num):
    prev_path, filename = filepath.rsplit("/", maxsplit=1)
    name, suffix = filename.split(".")
    with open(filepath, "r", encoding="utf-8") as f:
        totalList = f.readlines()
    chunksize = math.ceil(len(totalList) / chunk_num)
    data = [(totalList[start:start + chunksize], index, prev_path, name, suffix) for index, start in
            enumerate(range(0, len(totalList), chunksize))]
    start_time = time.time()
    with ProcessPoolExecutor(max_workers=chunk_num) as pool:
        result = pool.map(write, data)
    print(sum(result))
    end_time = time.time()
    print(end_time - start_time)

第二种方式比第一种方式需要更长的时间。我发现看起来不同的进程是串行工作而不是并行工作。输出是

enter
3.416102170944214
enter
3.3221476078033447
enter
3.198657989501953
enter
3.453885316848755
enter
3.261833429336548
16149274
42.55615472793579

那么这里的问题是什么？

Answer 1

更新的答案：

ProcessPoolExecutor 比单独使用 Process 的开销要多得多，因为 ProcessPoolExecutor 使用 Futures 并且总体上提供了更多功能，例如取消、检查状态、获取结果等的能力。

与此相关的是，Process 不会腌制返回值。虽然这不是您看到的大部分时间增加，但它并非微不足道。例如，如果您使用队列来返回一个值，您会发现使用 Process 会增加开销。

因此，总体而言，Process 在许多方面都更精简、更直接地成为 OS 构造，但没有 Futures 本身需要开销来管理它们的特性。

以下其他 Stack Overflow 问题/答案是相关的。

Stack Overflow #18671528

其他 Stack Overflow 答案中的建议是有道理的，即在需要 Futures 的好处时使用 ProcessPoolExecutor，也许是为了比您的运行时间更长的任务。请注意，另一篇文章提交的任务比您的要多得多，并且正在产生更大/显着的性能影响（它有 1000 和 1000 多个未完成的期货）。

一些您可能会感兴趣的额外细节

一些（不是全部）开销可以从 CPython Libconcurrentuturesprocess.py 文件顶部指定的设计中推断出来......

ProcessPoolExecutor design comments

它表明您的提交线程（线程调用映射）将一个工作项排入队列，该工作项由一个内部工作线程出队，该工作线程本身将一个项目排入“Call Q”......所有这些都会增加进程未观察到的延迟。

沿着这些思路，如果您查看 Future 的详细信息...

class Future

...您将看到条件变量的使用以及管理它们的大量代码。

为了具体概述在 CPython 中发生开销的位置，需要进行更详细的性能分析，我将省略。通常，您会为使用 ProcessPoolExecutor 获得的功能付出更多的时间成本。您选择哪种方法取决于您的需求、子流程将执行多长时间、您是否需要结果、Futures 的功能以及其他考虑因素。

原答案：

最初的建议是通过在对 ProcessPoolExecutor() 的调用中省略 chunk_num 来消除使用过多工作人员的可能性。 @pythonhua (OP) 已经确认操作系统是 Linux，块的数量只有 5 个，所以太多的工人不是问题。此外，错字已修复...在评论中使用了不正确的 chunksizes 而不是 chunk_num。

您的问题不完整，因为它没有指示文件大小、行大小和块大小等。

我假设您使用的是具有大块大小的 Linux。我怀疑您不必要地用更多的工人使 ProcessPoolExecutor 饱和。

chunk_num 和你应该选择的工人数量之间没有直接的关系。您必须考虑您的硬件或让 ProcessPoolExecutor 为您做出选择。

因此，我发现您使用 ProcessPoolExecutor 存在潜在问题。如果你有很多块，你最终会得到太多的工人，这会不必要地减慢速度。

尝试删除您的 max_workers 初始化程序（即，使用 None 并让 ProcessPoolExecutor 为您选择默认值），或尝试将其更改为您系统上的逻辑处理器数量或更少。

即，改变这个......

ProcessPoolExecutor(max_workers=chunk_num)

...类似于以下之一...

ProcessPoolExecutor()

ProcessPoolExecutor(max_workers=<num_logical_processors_or_less>)

在我的一个项目中，我使用以下方法来限制 max_workers。虽然这是基于我的项目需求，但您可以应用适用于您的项目的逻辑，同时牢记硬件。就我而言，无论 cpu_count()/2 如何，我都希望不超过 15 个工作人员...

min(os.cpu_count() // 2, 15)

请注意，我假设您使用的是 Linux，因为 Windows 没有 fork，并且它的 Process 实现当前将最大工作人员上限限制为 61。由于 max_workers 过多，我看不到 Windows 上的性能受到影响，但在 Linux 上肯定会受到影响。因此，我敢打赌，您基于不正确的假设指定了过多的工作人员，您拥有的工作项目的数量之间存在直接相关性，而与处理器或硬件无关，这将是不正确的。 ProcessPoolExecutor 可以为您管理一个可用的子流程池，这通常比您随着时间的推移必须执行的工作总数少。

如果这没有帮助，请提供更多信息以阐明问题的未定义特征。