Python 和 zlib : Terribly slow decompressing concatenated streams

Question

我收到了一个压缩文件，其中包含多个单独的压缩 XML 流。压缩后的文件大小为 833 MB。

如果我尝试将它作为单个对象解压缩，我只会得到第一个流(大约 19 kb)。

我修改了以下代码作为对 older question 的回答解压缩每个流并将其写入文件:

import zlib

outfile = open('output.xml', 'w')

def zipstreams(filename):
    """Return all zip streams and their positions in file."""
    with open(filename, 'rb') as fh:
        data = fh.read()
    i = 0
    print "got it"
    while i < len(data):
        try:
            zo = zlib.decompressobj()
            dat =zo.decompress(data[i:])
            outfile.write(dat)
            zo.flush()
            i += len(data[i:]) - len(zo.unused_data)
        except zlib.error:
            i += 1
    outfile.close()

zipstreams('payload')
infile.close()

此代码运行并产生所需的结果(所有 XML 数据都解压缩到一个文件中)。问题是需要好几天才能生效!

尽管压缩文件中有数万个流，但看起来这应该是一个更快的过程。大约 8 天解压 833mb(估计原始 3gb)表明我做错了一些事情。

是否有另一种方法可以更有效地执行此操作，或者速度慢是读取-解压缩-写入的结果——我一直遇到的重复瓶颈？

感谢您的任何指点或建议!

Answer 1

如果不了解您实际处理的文件格式的更具体知识，很难说太多，但很明显，您的算法对子字符串的处理是二次的——当您有数万个子字符串时，这不是一件好事他们。那么让我们看看我们知道什么:

你说供应商说他们是

using the standard zlib compression library.These are the same compression routines on which the gzip utilities are built.

由此我们可以得出结论，组件流采用原始 zlib 格式，并且未封装在 gzip 包装器(或 PKZIP 存档，或其他任何格式)中。 ZLIB 格式的权威文档在这里:https://www.rfc-editor.org/rfc/rfc1950

因此，让我们假设您的文件与您描述的完全一样:一个 32 字节的 header ，后跟连接在一起的原始 ZLIB 流，中间没有任何其他内容。(编辑: 毕竟不是这样的)。

Python 的 zlib documentation提供了一个 Decompress 类，它实际上非常适合翻动您的文件。它包含一个属性 unused_data，其 documentation明确指出:

The only way to determine where a string of compressed data ends is by actually decompressing it. This means that when compressed data is contained part of a larger file, you can only find the end of it by reading data and feeding it followed by some non-empty string into a decompression object’s decompress() method until the unused_data attribute is no longer the empty string.

因此，这就是您可以做的:编写一个循环来读取 data，比如说，一次一个 block (甚至不需要将整个 800MB 的文件读入内存)。将每个 block 推送到 Decompress 对象，并检查 unused_data 属性。当它变成非空时，你就有了一个完整的对象。将其写入磁盘，创建一个新的解压缩对象并使用上一个的 unused_data 初始化 iw。这可能会起作用(未经测试，因此请检查其正确性)。

编辑:由于您的数据流中确实有其他数据，因此我添加了一个与下一个 ZLIB 开始对齐的例程。您需要在您的 数据中找到并填写标识 ZLIB 流的双字节序列。 (随意使用您的旧代码来发现它。)虽然通常没有固定的 ZLIB header ，但每个流的 header 应该相同，因为它由 protocol options and flags, 组成。这在整个运行过程中可能是相同的。

import zlib

# FILL IN: ZHEAD is two bytes with the actual ZLIB settings in the input
ZHEAD = CMF+FLG  
    
def findstart(header, buf, source):
    """Find `header` in str `buf`, reading more from `source` if necessary"""

    while buf.find(header) == -1:
        more = source.read(2**12)
        if len(more) == 0:  # EOF without finding the header
            return ''
        buf += more
        
    offset = buf.find(header)
    return buf[offset:]

然后您可以前进到下一个流的开始。我添加了 try/except 对，因为相同的字节序列可能出现在流外:

source = open(datafile, 'rb')
skip_ = source.read(32) # Skip non-zlib header

buf = ''
while True:
    decomp = zlib.decompressobj()
    # Find the start of the next stream
    buf = findstart(ZHEAD, buf, source)
    try:    
        stream = decomp.decompress(buf)
    except zlib.error:
        print "Spurious match(?) at output offset %d." % outfile.tell(),
        print "Skipping 2 bytes"
        buf = buf[2:]
        continue
    
    # Read until zlib decides it's seen a complete file
    while decomp.unused_data == '':
        block = source.read(2**12)
        if len(block) > 0:       
            stream += decomp.decompress(block)
        else:
            break # We've reached EOF
        
    outfile.write(stream)
    buf = decomp.unused_data # Save for the next stream
    if len(block) == 0:
        break  # EOF

outfile.close()

PS 1. 如果我是你，我会将每个 XML 流写入一个单独的文件。

PS 2. 您可以测试在文件的前 MB 上执行的任何操作，直到获得足够的性能。