1

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

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

我修改了以下代码作为旧问题的答案，以解压缩每个流并将其写入文件：

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 原始数据）表明我做错了什么。

有没有另一种方法可以更有效地做到这一点，或者速度慢是我遇到的读取-解压缩-写入重复瓶颈的结果吗？

感谢您的任何指示或建议！

4

3 回答 3

4

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

你说供应商说他们是

使用标准的 zlib 压缩库。这些是构建 gzip 实用程序的相同压缩例程。

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

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

Python 的zlib 文档提供了一个Decompress实际上非常适合翻阅文件的类。它包括一个属性unused_data，其文档明确指出：

确定一串压缩数据在哪里结束的唯一方法是实际解压缩它。这意味着当压缩数据包含在较大文件的一部分时，您只能通过读取数据并将其后跟一些非空字符串提供给解压缩对象的 decompress() 方法来找到它的结尾，直到不再使用未使用的数据属性空字符串。

因此，您可以这样做：编写一个循环，一次读取data一个块（甚至不需要将整个 800MB 文件读入内存）。将每个块推送到Decompress对象，并检查unused_data属性。当它变成非空时，你就有了一个完整的对象。将其写入磁盘，创建一个新的解压缩对象并使用unused_data最后一个对象初始化 iw。这可能有效（未经测试，因此请检查正确性）。

编辑：由于您的数据流中确实有其他数据，因此我添加了一个与下一个 ZLIB 开始对齐的例程。您需要找到并填写在数据中标识 ZLIB 流的两字节序列。（请随意使用您的旧代码来发现它。）虽然通常没有固定的 ZLIB 标头，但对于每个流来说它应该是相同的，因为它由协议选项和标志组成，对于整个运行来说它们可能是相同的。

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 上测试您所做的任何事情，直到获得足够的性能。

于 2013-05-12T12:46:28.460 回答
1

在现代处理器（例如 2 GHz i7）上解压缩 833 MB 大约需要 30 秒。所以，是的，你做错了什么。尝试在每个字节偏移处解压缩以查看是否出现错误是问题的一部分，尽管不是全部。有更好的方法来查找压缩数据。理想情况下，您应该找到或弄清楚格式。或者，您可以使用RFC 1950 规范搜索有效的 zlib 标头，尽管您可能会得到误报。

更重要的可能是您一次将整个 833 MB 读取到内存中，然后将 3 GB 解压缩到内存中，可能每次都是大块。你的机器有多少内存？您可能正在颠簸到虚拟内存。

如果您显示的代码有效，则数据不会被压缩。zip 是一种特定的文件格式，通常带有 .zip 扩展名，它将原始 deflate 数据封装在本地和中央目录信息结构中，旨在重建文件系统中的目录。您必须有一些不同的东西，因为您的代码正在寻找并且显然正在寻找 zlib 流。你有什么格式？你在哪里得到它？它是如何记录的？你能提供前 100 个字节的转储吗？

应该这样做的方式不是将整个内容读入内存并立即解压缩整个流，也将其解压缩到内存中。相反，请使用zlib.decompressobj允许您一次提供一个片段的界面，并获取生成的可用解压缩数据。您可以读取小得多的输入文件，使用文档格式或查找 zlib（RFC 1950 标头）查找解压缩数据流，然后通过解压缩对象一次运行这些数据块，写出解压缩数据你想要的地方。 decomp.unused_data可用于检测压缩流的结束（如您找到的示例中所示）。

于 2013-05-12T16:47:21.833 回答
0

根据您在评论中的描述，听起来他们正在将他们分别发送给您的各个文件连接在一起。这意味着每个都有一个 32 字节的标题，您需要跳过。

如果您不跳过这些标头，它可能会具有您所描述的行为：如果幸运的话，您将收到 32 个无效标头错误，然后成功解析下一个流。如果运气不好，这 32 个字节的垃圾看起来就像是真正的流的开始，并且您将浪费大量时间来解析任意数量的字节，直到最终出现解码错误。（如果你真的不走运，它实际上会成功解码，给你一大块垃圾并吃掉一个或多个后续流。）

因此，请尝试在每个流完成后跳过 32 个字节。

或者，如果您有更可靠的方法来检测下一个流的开始（这就是为什么我告诉您打印出偏移量并在十六进制编辑器中查看数据，而 Alexis 告诉您查看 zlib 规范)，而是这样做。

于 2013-05-12T13:18:28.950 回答