python - 如何使用Python并行化CPU密集型数据处理任务？

Question

我正在创建一些代码来对大型 pdf 数据集执行 OCR 并将提取的文本写入 csv。这是使用 Imagemagick、Pillow、PyOCR (Tesseract) 等库的组合来完成的，并且已经在小数据样本上进行了测试。

数据包含大量文件夹(约 2500 个)，每个文件夹包含大约 15 个 pdf 文件。为每个文件夹中的 pdf 创建一个 csv，处理每个文件夹大约需要 10 分钟。这意味着对所有 2500 个文件夹运行 OCR 大约需要 18 天，这实在是太长了。我需要能够在 7 天或更短的时间内运行整个过程。

因此，我正在考虑并行处理每个子文件夹，因为每个子文件夹的处理独立于任何其他子文件夹。我的第一种方法是使用 concurrent.futures 模块创建一个 ProcessPool，如下所示:

executor = concurrent.futures.ProcessPoolExecutor(4)
futures = ([executor.submit(run_pdf_to_text_ocr, folder) for folder in sub_folders])
concurrent.futures.wait(futures)

这里run_pdf_to_text_ocr()是对每个文件夹运行处理的主函数。进程使用相同的资源存在一些问题，我正在解决这些问题，以便每个文件夹(进程/线程)的资源使用都是隔离的。

当然，这个处理管道非常消耗 CPU 资源，并且在运行时会最大化处理器的性能。我将启动一个大型 AWS EC2 实例以最终完成完整运行。所以，在开始之前我想知道的是:

If this approach I'm taking is the correct one? Are there any alternate approaches I can take to do this in a better way? Is using distributed processing something I should do instead? What can I do to properly monitor this long-running pipeline so I can be aware of any problems that might arise in the middle of processing?

我对 Python 非常熟悉，并且更喜欢尽可能使用它的解决方案。

Answer 1

我认为multiprocessing模块就是你想要的。

from multiprocessing import Pool

workers = Pool()
workers.apply_async(your_task_function, args=(args_for_this_task))
workers.join()

所以你可以轻松地将整个任务分割成小任务，然后传递给多CPU。您应用的任务将被放置在workers的队列中。

此外，要监视错误，您可以传递如下错误回调:workers.apply_async(your_task_function, args=(args_for_this_task), error_callback=error_callback)

此外，您可以使用 multiprocessing.Manager 轻松地跨进程共享变量，甚至可以在不同的计算机上共享变量。