Python compression.zstd 怎样读取大压缩流

读取几十 GB 的 .zst 文件时，最容易踩的坑是直接调用 zstd.decompress()：它会把解压结果作为一个完整的 bytes 返回，文件越大，瞬时内存压力越明显。Python 3.14 新增的 compression.zstd 提供了文件型接口和增量解压器，更适合按块处理大压缩流。

官方文档：https://docs.python.org/3.14/library/compression.zstd.html

本文只解决一个问题：怎样让解压结果以固定大小的块进入业务处理。文件场景优先用 zstd.open()；输入来自管道、socket 或其他类文件对象时，再使用 ZstdDecompressor 的增量接口。

compression.zstd 适合解决什么问题

compression.zstd 在 Python 3.14 加入，模块提供 open()、ZstdFile、一次性 compress/decompress，以及增量的 ZstdCompressor 和 ZstdDecompressor。其中，zstd.open() 返回的对象支持缓冲二进制文件接口，可以像普通文件一样读取；这正好把“解压”和“消费数据”拆成一块一块的动作。

如果目标只是处理一个小字节串，一次性函数很直接；如果目标是大文件，应该让每次读取的大小成为明确的内存预算。例如把块大小设为 1 MiB，业务层就只需要为当前处理块、少量缓冲和自己的临时对象预留空间，不必按完整解压尺寸估算。

用 zstd.open 按块读取大文件

对磁盘上的 .zst 文件，先使用二进制读取模式打开，再循环调用 read(CHUNK_SIZE)。循环体里直接解析、写出或统计当前块，处理完成后让引用离开当前迭代。

图1：文件型 Zstandard 流从压缩输入到处理块的原创静态结构说明图，不是终端或运行截图。

from compression import zstd

CHUNK_SIZE = 1024 * 1024  # 每次只把约 1 MiB 的解压数据交给业务层。


def consume_zstd_file(path: str, output_path: str) -> int:
    total = 0
    with zstd.open(path, "rb") as source, open(output_path, "wb") as target:
        while True:
            chunk = source.read(CHUNK_SIZE)
            if not chunk:
                break  # 空字节表示已经读到解压后的文件末尾。
            target.write(chunk)  # 实际项目可在这里解析，而不是再次完整缓存。
            total += len(chunk)
    return total


written = consume_zstd_file("events.zst", "events.ndjson")
print(f"写出解压数据：{written} 字节")

这里的 target.write(chunk) 只是一个最小的消费例子。如果处理的是 NDJSON，不要为了方便把所有块拼接成一个大字符串；可以在块之间保留一小段未完成的行，找到换行符后立即交给 JSON 解析器。

非文件输入用增量解压器

当压缩数据来自网络读取、管道或自定义的分块来源时，可以给 ZstdDecompressor.decompress() 传入输入块，并用 max_length 限制一次返回的解压数据量。它与文件接口的区别是：你需要自己管理输入块、输出块和当前帧状态。

图2：增量解压器与帧边界状态的原创静态关系说明图，不是运行结果截图。

from compression import zstd

OUTPUT_LIMIT = 1024 * 1024  # 限制单次交给业务层的解压输出大小。


def consume_stream(chunks):
    decoder = zstd.ZstdDecompressor()
    for compressed_chunk in chunks:
        pending = compressed_chunk
        while pending or not decoder.needs_input:
            output = decoder.decompress(pending, max_length=OUTPUT_LIMIT)
            pending = b""  # 解压器可能已经把输入缓冲在内部。
            if output:
                yield output  # 调用方应立即处理，不要把所有块收集到列表。
            if decoder.eof:
                break  # 当前帧结束；后续数据是否属于下一帧要另行决定。
        if decoder.eof:
            break


for data_block in consume_stream(read_from_socket()):
    handle(data_block)  # 业务层按块消费网络解压结果。

max_length 到达上限时，needs_input 可能仍为 False，表示解压器内部还有输出可取；因此示例会在不提供新输入的情况下继续调用一次。相反，needs_input 为 True 时，才需要等待下一块压缩输入。

多帧、EOF 和模块缺失要单独处理

文件接口和模块级 decompress() 可以处理拼接的多个压缩帧，但 ZstdDecompressor 不会自动透明地处理多帧输入。使用增量接口时，读到 eof 后不要把剩余字节悄悄当作同一帧；如果协议允许连续帧，应根据帧边界重新创建解压器，并把剩余输入交给下一帧。

unused_data 用来查看当前帧结束后的剩余数据，eof 用来判断结束标记是否已到达，needs_input 用来判断是否还需要更多压缩输入。发生格式损坏或解压初始化失败时捕获 zstd.ZstdError，记录输入来源并停止消费，不要继续把不完整输出写入下游。

from compression import zstd


def decode_one_frame(data: bytes) -> bytes:
    decoder = zstd.ZstdDecompressor()
    try:
        result = decoder.decompress(data)
    except zstd.ZstdError as exc:
        # 格式错误时停止输出，避免下游误把半截数据当成完整结果。
        raise ValueError("Zstandard 数据不可解压") from exc
    if not decoder.eof:
        raise ValueError("输入尚未包含完整的 Zstandard 帧")
    if decoder.unused_data:
        # 当前函数只处理一帧；多帧协议应把这段数据交给新的解压器。
        raise ValueError("输入包含当前函数未处理的后续帧")
    return result

另外，compression.zstd 是可选模块。如果目标 Python 是 3.14 或更高版本但导入仍失败，要先确认发行版是否构建了这个可选模块；不能仅凭 Python 主版本号就假定它一定存在。生产代码可以在启动阶段给出清楚的依赖提示，而不是等到收到第一份压缩数据时才暴露错误。

把块大小当成接口边界

文件型读取和增量解压都能控制“单次输出”，但它们不会替业务层控制更深一层的内存：解析器可能保留半条记录，解压后的单块也可能触发临时对象分配。因此建议把块大小、未完成记录缓存和下游写入策略放在同一处设计。

常见问题

为什么不用 zstd.decompress()？它适合小输入或确实需要完整结果的场景；大流量处理应改用文件接口或增量解压器。

块大小越小越省内存吗？通常会降低单块内存，但调用次数和系统开销会增加。先用 256 KiB 到 1 MiB 这类明确预算，再结合下游处理速度调整。

只拿到半块压缩数据会失败吗？不一定。增量解压器可以等待更多输入；关键是根据 needs_input 决定是否继续读取，而不是把一次 decompress() 返回空字节误判成结束。