加速读取压缩的 bz2 文件'rb' 模式

问题描述

我有一个超过 10GB 的 BZ2 文件。我想在不解压成临时文件的情况下阅读它（它会超过 50GB）。

使用这种方法：

import bz2,time
t0 = time.time()
time.sleep(0.001) # to avoid / by 0
with bz2.open("F:\test.bz2",'rb') as f:
    for i,l in enumerate(f):
        if i % 100000 == 0:
            print('%i lines/sec' % (i/(time.time() - t0)))

我每秒只能读取约 25 万行。在类似的文件中，首先解压，我每秒得到大约 3M 行，即 x10 因子：

with open("F:\test.txt",'rb') as f:

我认为这不仅是由于固有的解压 CPU 时间（因为解压到临时文件的总时间 + 作为未压缩文件的读取比此处描述的方法小得多），而且可能是缺少缓冲，或者其他原因。 bz2.open 还有其他更快的 Python 实现吗？

如何在二进制模式下加速读取 BZ2 文件，并循环“行”？（以 \n 分隔）

注意：目前 time to decompress test.bz2 into test.tmp + time to iterate over lines of test.tmp 远小于 time to iterate over lines of bz2.open('test.bz2')，这可能不应该是这种情况。

链接主题：https://discuss.python.org/t/non-optimal-bz2-reading-speed/6869

解决方法

您可以使用 BZ2Decompressor 来处理大文件。它以增量方式解压缩数据块，开箱即用：

t0 = time.time()
time.sleep(0.000001)
with open('temp.bz2','rb') as fi:
    decomp = bz2.BZ2Decompressor()
    residue = b''
    total_lines = 0
    for data in iter(lambda: fi.read(100 * 1024),b''):
        raw = residue + decomp.decompress(data) # process the raw data and  concatenate residual of the previous block to the beginning of the current raw data block
        residue = b''
        # process_data(current_block) => do the processing of the current data block
        current_block = raw.split(b'\n')
        if raw[-1] != b'\n':
            residue = current_block.pop() # last line could be incomplete
        total_lines += len(current_block)
        print('%i lines/sec' % (total_lines / (time.time() - t0)))
    # process_data(residue) => now finish processing the last line
    total_lines += 1
    print('Final: %i lines/sec' % (total_lines / (time.time() - t0)))

在这里，我读取了一大块二进制文件，将其送入解压缩器并接收一大块解压缩数据。请注意，解压后的数据块必须连接起来才能恢复原始数据。这就是为什么最后一个条目需要特殊处理。

在我的实验中，它的运行速度比使用 io.BytesIO() 的解决方案快一点。众所周知，bz2 速度很慢，因此如果您感到困扰，可以考虑迁移到 snappy 或 zstandard。

关于在 Python 中处理 bz2 所需的时间。使用 Linux 实用程序将文件解压缩为临时文件，然后处理普通文本文件可能是最快的。否则，您将依赖 Python 的 bz2 实现。

,

这种方法已经比原生 bz2.open 提高了 2 倍。

import bz2,time,io

def chunked_readlines(f):
    s = io.BytesIO()
    while True:
        buf = f.read(1024*1024)
        if not buf:
            return s.getvalue()
        s.write(buf)
        s.seek(0)
        L = s.readlines()
        yield from L[:-1]
        s = io.BytesIO()
        s.write(L[-1])  # very important: the last line read in the 1 MB chunk might be
                        # incomplete,so we keep it to be processed in the next iteration
                        # TODO: check if this is ok if f.read() stopped in the middle of a \r\n?

t0 = time.time()
i = 0
with bz2.open("D:\test.bz2",'rb') as f:
    for l in chunked_readlines(f):       # 500k lines per second
    # for l in f:                        # 250k lines per second
        i += 1
        if i % 100000 == 0:
            print('%i lines/sec' % (i/(time.time() - t0)))

也许可以做得更好。

如果我们可以将 s 用作简单的 bytes 对象而不是 io.BytesIO，我们可以获得 x4 的改进。但不幸的是，在这种情况下，splitlines() 的行为不符合预期：splitlines() and iterating over an opened file give different results。