逐行流式传输 .zst 压缩文件

Question

我正在尝试筛选压缩在 .zst 中的大型数据库。我知道我可以简单地解压缩它，然后处理生成的文件，但这会占用我的 SSD 上的大量空间，并且需要 2 个多小时，因此我希望尽可能避免这种情况。

通常，当我处理大文件时，我会使用如下代码逐行流式传输它

with open(filename) as f:
    for line in f.readlines():
        do_something(line)

我知道 gzip 有这个

with gzip.open(filename,'rt') as f:
    for line in f:
        do_something(line)

但它似乎不适用于 .zsf，所以我想知道是否有任何库可以以类似的方式解压缩和流式传输解压缩的数据。例如：

with zstlib.open(filename) as f:
    for line in f.zstreadlines():
        do_something(line)

Answer 1

知道要使用哪个包以及相应的文档可能会有点令人困惑，因为似乎有几个 Python 绑定到实际的 Zstandard 库。

下面，我指的是 Gregory Szorc 的库，我从conda默认通道安装了该库：

conda install zstd

# check:

conda list zstd
# # Name                    Version                   Build  Channel
# zstd                      1.5.5                hc292b87_0  

（尽管文档说要安装pip，但除非没有其他方法，否则我不会这样做，因为我希望我的 conda 环境保持可用）。

我只是根据文件中的注释推断这个版本是 G. Szorc 的版本__init__.py：

# Copyright (c) 2017-present, Gregory Szorc
# All rights reserved.
#
# This software may be modified and distributed under the terms
# of the BSD license. See the LICENSE file for details.

"""Python interface to the Zstandard (zstd) compression library."""

from __future__ import absolute_import, unicode_literals

# This module serves 2 roles:
#
# 1) Export the C or CFFI "backend" through a central module.
# 2) Implement additional functionality built on top of C or CFFI backend.

因此，我认为相应的文档就在这里。

无论如何，安装后快速测试：

import zstandard as zstd

with zstd.open('test.zstd', 'w') as f:
    for i in range(10_000):
        f.write(f'foo {i} bar\n')

with zstd.open('test.zstd', 'r') as f:
    for i, line in enumerate(f):
        if i % 1000 == 0:
            print(f'line {i:4d}: {line}', end='')

生产：

line    0: foo 0 bar
line 1000: foo 1000 bar
line 2000: foo 2000 bar
line 3000: foo 3000 bar
line 4000: foo 4000 bar
line 5000: foo 5000 bar
line 6000: foo 6000 bar
line 7000: foo 7000 bar
line 8000: foo 8000 bar
line 9000: foo 9000 bar

笔记：

1. 如果文件是用二进制（不是文本）编写的，则使用mode='rb', 与常规文件相同。底层文件总是以二进制模式写入，但是如果我们使用文本模式open，那么根据open的文档，“（...） ifio.TextIOWrapper打开以文本模式读取或写入”。
2. 请注意，我使用的是 的迭代器f，而不是readlines(). 从内联文档字符串来看，它们听起来像是readlines()从文件中返回行列表，即整个内容都在内存中。使用迭代器，很可能任何时候只有文件的一部分位于内存中（在zstd的缓冲区中）。
3. 然而，阅读这部分文档后，我对上述内容不太确定。请继续关注...（编辑：根据经验进行测试，它成立，见下文）。

附录

关于上面的注释2和3：我通过经验测试，通过将行数更改为1亿行并比较两个版本的内存使用情况（使用htop）：

流媒体版本

with zstd.open('test.zstd', 'r') as f:
    for i, line in enumerate(f):
        if i % 10_000_000 == 0:
            print(f'line {i:8d}: {line}', end='')

--内存使用量没有增加。

阅读行版本

with zstd.open('test.zstd', 'r') as f:
    for i, line in enumerate(f.readlines()):
        if i % 10_000_000 == 0:
            print(f'line {i:8d}: {line}', end='')

--内存使用量增加了几 GB。

这可能特定于安装的版本 (1.5.5)。