在 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. 主程序读取文件，将行写入管道，执行将行处理到适当的子进程所需的最小过滤。

每个子进程可能应该是从标准输入读取和写入的不同进程的管道。

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

Answer 2

的一种策略是分配给每个工人的偏移，所以如果你有八个工作进程分配然后号码0〜7员号码0读取第一记录处理它然后跳过7和接着处理所述第八记录等，工人数1读取那么第二记录跳过7，并处理该第九记录.........

有该方案具有许多优点。它不将文件不管有多大的工作总是平均分配，在同一台机器上的进程将在大致相同的速率来处理，所以你不承担任何过多的I / O开销使用相同的缓冲区域。只要文件处理不当被更新可以重新运行各个线程以从故障中恢复。

Answer 3

您不要提起你是如何处理的线;可能是最重要的一块信息。

时每行独立的？是计算依赖于一个前行的下一个未来？他们必须以块进行处理？多久每行的处理时间？是否有必须在最后收编“所有”数据的处理步骤？或者可中间结果被扔掉，保持只是一个运行总数？可以将文件通过线程数除以文件大小最初分裂？抑或当你处理它它成长？

如果该行是独立的和文件不长，只有你所需要的协调是种田去“开始的地址”和“长”到每个劳动者的;他们可以independantly开放，并寻求到该文件，然后你必须简单地协调它们的结果;或许通过等待N个结果来返回到队列中。

如果线路的不独立的，答案将高度依赖于文件的结构。

Answer 4

我知道你具体问了一下Python，但我会鼓励你看的Hadoop（的http：// Hadoop的。 apache.org/ ）：它实现了Map和Reduce这是专为解决这类问题的算法。

好运

Answer 5

这在很大程度上取决于你的文件的格式。

是否有意义的任何地方拆呢？或者你需要把它在一个新行分割？或者你需要确保你的对象定义的最后拆呢？

而不是分裂的文件，你应该在同一个文件中使用多个阅读器，使用os.lseek跳转到该文件的相应部分。

更新：海报补充说，他想拆就新的生产线。然后，我提出以下建议：

让我们说你有4个过程。然后简单的解决方法是os.lseek为0％，25％，50％和文件的75％，和读取的字节，直到你遇到第一个新的生产线。这对每一个过程的起点。你并不需要分割的文件，要做到这一点，只是寻求到正确的位置在每个过程中的大量文件，并从那里开始阅读。

Answer 6

弗雷德里克Lundh的上蒂姆·布雷的广泛搜索基准一些注意事项是一个有趣的阅读，约一个非常类似用途的情况下，用了很多好的建议。各种其他作者也实施了同样的事情，有些是从文章链接，但你可能想尝试google搜索“蟒蛇宽取景器”，或有事找一些。 （也有某处基于multiprocessing模块上的解决方案，但似乎并不可用了）

Answer 7

如果运行时间长，而不是具有各处理通过Queue读取其下一行，具有进程读取线的批次。这样的开销分摊在若干行（例如数千或更多）。