Python 3.14 标准库新增压缩能力怎样使用

Python 3.14 把 Zstandard 能力带进标准库，入口是 compression.zstd。如果任务是压缩一段 bytes，直接用 compress()；如果要处理文件，优先用 zstd.open()；数据持续从网络或队列到达，则使用 ZstdCompressor 做增量压缩。先确认当前解释器包含这个可选模块，再按数据形态选 API，能避免把一次性接口硬套到大文件上。

先确认解释器真的提供 compression.zstd

这个模块虽然属于 Python 3.14 标准库，但官方文档明确它是可选模块，某些发行版构建可能没有带上底层 zstd。检查时不要只看 sys.version_info，还要实际导入：

import sys

try:
    from compression import zstd
except ImportError as exc:
    # 版本不足或发行版缺少可选模块时，给出可行动的诊断
    raise RuntimeError("当前 Python 没有 compression.zstd，请检查 3.14+ 和发行版构建") from exc

print(sys.version)
    print(zstd.zstd_version_info)  # 查看运行时链接的 zstd 版本

如果项目还要支持 Python 3.13 及更早版本，导入失败应进入兼容分支，而不是在模块顶层无条件导入。此时可以继续使用第三方 zstandard 包或旧的标准库压缩模块，具体选择由部署环境和文件格式决定。

小块数据直接用 compress 和 decompress

配置快照、短消息和缓存字段适合一次性接口。它接收类 bytes 对象并返回 bytes；解压时，发送端和接收端必须对字典、格式和数据完整性保持一致：

from compression import zstd

payload = b"user_id=17\nfeature=search\n" * 20

try:
    packed = zstd.compress(payload, level=5)  # 级别越高通常越省空间，但不等于一定更快
    restored = zstd.decompress(packed)
except zstd.ZstdError as exc:
    # 压缩或解压失败时不要把损坏数据当成空结果继续处理
    raise ValueError("Zstandard 数据不可用") from exc

assert restored == payload
    print(len(payload), len(packed))

这里的 level=5 只是示例参数。生产环境应使用自己的样本测量压缩比、CPU 时间和内存，再决定级别；不能把压缩后的字节数直接当成所有数据都能达到的比例。

文件场景优先使用 zstd.open

zstd.open 可以打开路径，也可以包装一个已经打开的文件对象。二进制文件使用 wb 或 rb，文本文件才使用 wt 或 rt：

from compression import zstd

source = "events-2026.jsonl"
target = "events-2026.jsonl.zst"

with open(source, "rb") as src, zstd.open(target, "wb", level=3) as dst:
    # 分块复制，避免把整个输入文件一次性读入内存
    while chunk := src.read(1024 * 1024):
        dst.write(chunk)

with zstd.open(target, "rb") as src:
    restored = src.read()  # 小于内存预算时再一次性读取解压结果

    print(len(restored))

关闭 zstd.open 返回的对象会完成压缩流的收尾；把写入对象交给下游时，仍要保留清晰的所有权边界。若包装的是已有文件对象，关闭 Zstandard 层不会自动关闭底层对象，调用方应按自己的生命周期管理它。

分片输入必须用 flush 完成最后一个 frame

网络响应、日志流或消息队列通常不会一次拿到全部数据，此时可以复用一个压缩器。compress() 可能暂时只缓存输入，最后的 flush() 才负责输出尾部数据并结束当前 frame：

from compression import zstd

compressor = zstd.ZstdCompressor(level=3)
parts = []
for block in (b"header\n", b"body-1\n", b"body-2\n"):
    # CONTINUE 允许压缩器继续引用前面的数据
    parts.append(compressor.compress(block))

parts.append(compressor.flush())  # 必须保留尾部输出，才能得到完整可解压数据
packed = b"".join(parts)
restored = zstd.decompress(packed)
print(restored.decode("utf-8"))

如果要把输入拆成多个独立 frame，可使用 FLUSH_FRAME；但这会改变后续数据能否引用前一段内容的边界。连续压缩不要在每个小块后都结束 frame，否则可能牺牲压缩效率。

迁移时看格式和上下游，不要只看新 API

Python 3.14 新增的不只是 compression.zstd：tarfile、zipfile 和 shutil 也增加了读取或写入 Zstandard 压缩归档的支持。同时，compression.gzip 等新路径是原模块的规范入口，但旧名称没有立即废弃。需要和旧脚本、其他语言或命令行工具交换文件时，先确认对方支持的格式，再决定是否切换。

常见疑问

Python 3.14 一定能导入 compression.zstd 吗？不一定。它是可选模块，解释器版本满足只是必要条件，还要看具体发行版是否包含底层 zstd 支持。

压缩级别越高就越适合生产吗？不一定。级别会影响 CPU、内存、延迟和压缩比，应使用接近真实的数据样本做基准，再结合上下游解压能力选择。