逐行处理非常大(> 20GB)的文本文件

Question

我有一些非常大的文本文件需要处理,最大的是大约60GB.

每行在七个字段中有54个字符,我想从前三个字段中删除最后三个字符 - 这应该将文件大小减少大约20%.

我是Python的新手,并且有一个代码可以按照每小时大约3.4 GB的速度完成我想做的事情,但是我需要获得至少10 GB /小时的有价值的练习 - 有没有办法加速这个？这段代码并没有接近挑战我的处理器,所以我做了一个没有受过教育的猜测,它受到内部硬盘读写速度的限制？

def ProcessLargeTextFile():
    r = open("filepath", "r")
    w = open("filepath", "w")
    l = r.readline()
    while l:
        x = l.split(' ')[0]
        y = l.split(' ')[1]
        z = l.split(' ')[2]
        w.write(l.replace(x,x[:-3]).replace(y,y[:-3]).replace(z,z[:-3]))
        l = r.readline()
    r.close()
    w.close()

任何帮助将非常感激.我在Windows 7上使用IDLE Python GUI并拥有16GB的内存 - 也许一个不同的操作系统会更有效率？

编辑:这是要处理的文件的摘录.

70700.642014 31207.277115 -0.054123 -1585 255 255 255
70512.301468 31227.990799 -0.255600 -1655 155 158 158
70515.727097 31223.828659 -0.066727 -1734 191 187 180
70566.756699 31217.065598 -0.205673 -1727 254 255 255
70566.695938 31218.030807 -0.047928 -1689 249 251 249
70536.117874 31227.837662 -0.033096 -1548 251 252 252
70536.773270 31212.970322 -0.115891 -1434 155 158 163
70533.530777 31215.270828 -0.154770 -1550 148 152 156
70533.555923 31215.341599 -0.138809 -1480 150 154 158

Answer 1

编写这样的代码更加惯用

def ProcessLargeTextFile():
    with open("filepath", "r") as r, open("outfilepath", "w") as w:
        for line in r:
            x, y, z = line.split(' ')[:3]
            w.write(line.replace(x,x[:-3]).replace(y,y[:-3]).replace(z,z[:-3]))

这里的主要优点是只执行split一次,但如果CPU没有征税,这可能会产生很小的差异

它可能有助于一次保存几千行并一次写入它们以减少硬盘的颠簸.一百万行只有 54MB的RAM!

def ProcessLargeTextFile():
    bunchsize = 1000000     # Experiment with different sizes
    bunch = []
    with open("filepath", "r") as r, open("outfilepath", "w") as w:
        for line in r:
            x, y, z = line.split(' ')[:3]
            bunch.append(line.replace(x,x[:-3]).replace(y,y[:-3]).replace(z,z[:-3]))
            if len(bunch) == bunchsize:
                w.writelines(bunch)
                bunch = []
        w.writelines(bunch)

@Janne建议,一种生成线的替代方法

def ProcessLargeTextFile():
    bunchsize = 1000000     # Experiment with different sizes
    bunch = []
    with open("filepath", "r") as r, open("outfilepath", "w") as w:
        for line in r:
            x, y, z, rest = line.split(' ', 3)
            bunch.append(' '.join((x[:-3], y[:-3], z[:-3], rest)))
            if len(bunch) == bunchsize:
                w.writelines(bunch)
                bunch = []
        w.writelines(bunch)

Answer 2

测量!你有一些有用的提示如何改进你的python代码,我同意他们.但你应该首先弄清楚,你真正的问题是什么.我找到瓶颈的第一步是:

• 从代码中删除任何处理.只需读取和写入数据并测量速度.如果只是读取和写入文件太慢,那么这不是代码问题.
• 如果只是读写速度很慢,请尝试使用多个磁盘.你正在同时阅读和写作.在同一张光盘上？如果是,请尝试使用不同的光盘,然后重试.
• 一些异步io库(Twisted？)也可能有所帮助.

如果您发现了确切的问题,请再次询问该问题的优化.

Answer 3

由于您似乎不受CPU的限制,而是受I/O的限制,您是否尝试过对第三个参数的一些变化open？

实际上,第三个参数可用于给出用于文件操作的缓冲区大小!

open( "filepath", "r", 16777216 )从文件读取时,简单地写入将使用16 MB缓冲区.它必须有所帮助.

对输出文件使用相同的文件,然后测量/比较相同的文件.

注意:这是其他人建议的同类优化,但您可以在此免费获取,无需更改代码,无需自行缓冲.

Answer 4

我将添加这个答案来解释为什么缓冲是有意义的,并提供一个更多的解决方案

你的表现令人惊叹.这篇文章有可能加速python IO吗？表明10 gb读取应该在3分钟左右.顺序写入速度相同.所以你错过了30倍,你的表现目标仍然比应该可能的慢10倍.

几乎可以肯定,这种差异在于磁盘正在寻找的磁头数量.头部搜索需要几毫秒.单个搜索对应于几兆字节的顺序读写.非常昂贵.在同一磁盘上复制操作需要在输入和输出之间进行搜索.如上所述,减少搜索的一种方法是以这样的方式缓冲,即在写入磁盘之前读取许多兆字节,反之亦然.如果你可以说服python io系统做到这一点,那太好了.否则,您可以读取并处理行到字符串数组中,然后在准备好50 MB的输出后写入.这个大小意味着搜索将导致数据传输本身的性能损失<10%.

消除输入和输出文件之间寻求的另一种非常简单的方法是使用具有两个物理磁盘的机器,并为每个物理磁盘完全分离io通道.从一个输入.输出到其他.如果您正在进行大量的大文件转换,那么拥有具有此功能的计算机是件好事.

Answer 5

ProcessLargeTextFile():
    r = open("filepath", "r")
    w = open("filepath", "w")
    l = r.readline()
    while l:

正如已经建议的那样，您可能希望使用 for 循环使其更优化。

    x = l.split(' ')[0]
    y = l.split(' ')[1]
    z = l.split(' ')[2]

您在这里执行了 3 次拆分操作，这取决于每行的大小，这将对性能产生不利影响。您应该拆分一次并将 x,y,z 分配给返回的数组中的条目。

    w.write(l.replace(x,x[:-3]).replace(y,y[:-3]).replace(z,z[:-3]))

您正在阅读的每一行，都会立即写入文件，这是非常 I/O 密集型的。您应该考虑将输出缓冲到内存并定期推送到磁盘。像这样的东西：

BUFFER_SIZE_LINES = 1024 # Maximum number of lines to buffer in memory

def ProcessLargeTextFile():
    r = open("filepath", "r")
    w = open("filepath", "w")
    buf = ""
    bufLines = 0
    for lineIn in r:

        x, y, z = lineIn.split(' ')[:3]
        lineOut = lineIn.replace(x,x[:-3]).replace(y,y[:-3]).replace(z,z[:-3])
        bufLines+=1

        if bufLines >= BUFFER_SIZE:
            # Flush buffer to disk
            w.write(buf)
            buf = ""
            bufLines=1

        buf += lineOut + "\n"

    # Flush remaining buffer to disk
    w.write(buf)
    buf.close()
    r.close()
    w.close()

您可以调整 BUFFER_SIZE 以确定内存使用和速度之间的最佳平衡。

Answer 6

这是加载任何大小的文本文件而不会导致内存问题的代码。它支持千兆字节大小的文件。它可以在任何类型的机器上顺利运行，您只需要根据您的系统 RAM 配置 CHUNK_SIZE。CHUNK_SIZE越多，一次读取的数据越多

https://gist.github.com/iyvinjose/e6c1cb2821abd5f01fd1b9065cbc759d

下载文件 data_loading_utils.py 并将其导入到您的代码中

用法

import data_loading_utils.py.py
file_name = 'file_name.ext'
CHUNK_SIZE = 1000000


def process_lines(line, eof, file_name):

    # check if end of file reached
    if not eof:
         # process data, data is one single line of the file

    else:
         # end of file reached

data_loading_utils.read_lines_from_file_as_data_chunks(file_name, chunk_size=CHUNK_SIZE, callback=process_lines)

process_lines 方法是回调函数。它将为所有行调用，参数 line 一次代表文件的一行。

您可以根据您的机器硬件配置配置变量CHUNK_SIZE。

Answer 7

您可以尝试先保存分割结果，而不是每次需要字段时都这样做。也许这会加速。

你也可以尝试不要在gui中运行它。在cmd中运行它。