在Python中划分大型文件以进行多处理的最佳方法是什么？

Question

我遇到了很多"令人尴尬的并行"项目,我想与multiprocessing模块并行化.但是,它们通常涉及读取大文件(大于2GB),逐行处理,运行基本计算,然后写入结果.使用Python的多处理模块拆分文件并处理文件的最佳方法是什么？如若Queue或JoinableQueue在multiprocessing使用？还是Queue模块本身？或者,我应该使用multiprocessing？在一个进程池上映射可迭代文件？我已经尝试了这些方法,但是在逐行分配数据方面的开销是巨大的.我已经使用了一个轻量级的管道过滤器设计cat file | process1 --out-file out1 --num-processes 2 | process2 --out-file out2,它将第一个进程的一定比例的输入直接传递给第二个输入(参见这篇文章),但是我想要一个完全包含在Python中的解决方案.

令人惊讶的是,Python文档没有提出这样做​​的规范方法(尽管multiprocessing文档中有关编程指南的冗长部分).

谢谢,文斯

附加信息:每行的处理时间各不相同.有些问题很快,几乎没有I/O限制,有些是受CPU限制的.CPU绑定的非依赖任务将从并行化获得后期,使得即使是低效的将数据分配给处理功能的方式在挂钟时间方面仍然是有益的.

一个主要的例子是一个脚本,它从行中提取字段,检查各种按位标志,并将具有某些标志的行以全新格式写入新文件.这似乎是一个I/O限制问题,但是当我使用带有管道的廉价并发版本运行它时,速度提高了大约20%.当我使用池和映射运行它时,或者队列中的队列multiprocessing总是超过100%.

Answer 1

最好的架构之一已经是Linux操作系统的一部分.不需要特殊的库.

你想要一个"扇出"的设计.

1. "主"程序创建了许多通过管道连接的子进程.

2. 主程序读取文件,将行写入管道,执行将行处理到适当的子进程所需的最小过滤.

每个子进程可能应该是从stdin读取和写入的不同进程的管道.

您不需要队列数据结构,这正是内存中的管道 - 两个并发进程之间的字节队列.

Answer 2

一种策略是为每个工作人员分配一个偏移,所以如果你有八个工人进程你分配然后编号0到7.工人编号0读取第一个记录处理它然后跳过7然后继续处理第8个记录等,工人编号1读取第二条记录然后跳过7并处理第9条记录.........

该方案有许多优点.无论文件有多大,工作总是均匀分配,同一台机器上的进程将以大致相同的速率处理,并使用相同的缓冲区,因此不会产生任何过多的I/O开销.只要文件没有更新,您就可以重新运行单个线程以从故障中恢复.