Python 3.14 compression.zstd 怎么从内存压缩落到文件：level、open 与可选模块边界

Python 3.14 的 compression.zstd 适合把压缩边界留在标准库里处理：小对象可以直接用 compress()，落盘文件则用 open()，不必先把所有逻辑改成第三方客户端。真正需要核对的是输入是否已经足够大、压缩级别是否值得，以及部署环境有没有提供这个可选模块。

先把压缩基线测出来

假设服务每天把一批结构化日志暂存为字节串，第一步不是马上调高压缩级别，而是记录原始长度、压缩长度和解压后的 SHA-256。这样后面换级别或换文件接口时，至少有一条可以复现的基线。

from compression import zstd
import hashlib

payload = (b"event=checkout status=ok region=cn-east-1\n" * 4000)
packed = zstd.compress(payload)
restored = zstd.decompress(packed)

print("raw:", len(payload))
print("packed:", len(packed))
print("same:", hashlib.sha256(payload).digest() == hashlib.sha256(restored).digest())

这里的 same 只能说明这一轮数据可逆，不代表压缩率在所有日志上都一样。短文本、已经压缩过的图片和随机字节，结果很可能不划算。

压缩级别怎么选：先拿默认值做对照

zstd.compress(data, level=None) 允许显式传入压缩级别。建议先把默认值当成基线，再用少量代表性样本比较 CPU 时间、压缩后字节数和解压时间；不要只盯着最终文件大小。

from compression import zstd
import time

def measure(data: bytes, level: int | None):
    start = time.perf_counter()
    packed = zstd.compress(data, level=level)
    cost_ms = (time.perf_counter() - start) * 1000
    return len(packed), round(cost_ms, 2)

for level in (None, 3, 9):
    print(level, measure(payload, level))

样本量较小时，计时结果会被进程调度和模块初始化操作放大。更稳妥的做法是先做预热再重复多轮测试，分别记录 P50 和 P95 耗时，再决定批处理任务要不要调高压缩级别。线上请求链路一般对尾延迟敏感度很高，只有归档类任务更能接受压缩带来的额外CPU开销。

从内存切到文件：用 open 保持核对动作

需要写入 .zst 文件时，可以把压缩流直接交给 zstd.open()。写完后不要只看文件存在，至少重新打开读取，并把恢复后的内容与原始摘要对上。

from compression import zstd
import hashlib

with zstd.open("events.zst", "wb") as output:
    output.write(payload)

with zstd.open("events.zst", "rb") as source:
    restored = source.read()

assert hashlib.sha256(restored).digest() == hashlib.sha256(payload).digest()

使用模块自带文件接口的好处是调用方不需要手动拼接压缩帧；对应的代价是文件名、访问权限、磁盘剩余空间以及写入不完整的临时文件，都需要应用层自行管理。建议先写入同目录下的临时文件，确认写入完成、内容校验无误后再原子替换目标文件，避免后续消费者读到未完成的损坏内容。

部署前要查两个兼容边界

当前解释器是否真的带有 compression.zstd

官方文档把 compression.zstd 标为 Python 3.14 新增的可选模块。镜像、发行版和本地解释器的构建方式不同，不能只根据 python --version 就断言它一定可导入。

try:
    from compression import zstd
except ImportError as exc:
    raise RuntimeError("当前 Python 缺少 compression.zstd") from exc

print(zstd.zstd_version_info)

把这项版本和模块可用性检查放在服务启动阶段，出现问题时会比第一次处理归档请求才报错更容易定位。如果业务需要兼容低版本环境，回退逻辑要做成明确的部署层能力，不要藏在异常捕获里静默切换算法。

消费端是否认识 Zstandard

生产端能生成 .zst 不等于所有脚本、日志采集器和旧服务都能读取。发布前列出文件消费者，逐一用一份小样本做读取验收；如果只能由 Python 3.14 读取，就把版本要求写进部署文档和运行检查。

常见问题与边界

为什么压缩后反而变大

待压缩数据长度太短、内容本身接近随机或者已经经过其他压缩处理时，zstd的帧头和附加字典信息占用的体积，可能反而超过压缩节省的空间。提前记录好原始长度和压缩后长度，压缩率达不到业务预设阈值的话，就直接保留原始内容或者换其他存储策略。

能不能把 level 固定成最高

不建议直接默认开最高压缩级别。最高级别一般是消耗更多CPU资源来换取更小的压缩结果，收益要结合实际场景判断：你的场景是在线请求、批量归档还是跨网络传输，差别很大。用和生产一致的代表性样本跑完压测再固定配置，同时保留默认级别的对照数据方便后续回溯。

旧版本 Python 能直接导入吗

不能把 Python 3.14 新加入的 compression.zstd 模块当成全版本通用的旧接口处理。需要兼容旧运行环境时，要在安装环节和启动阶段显式检测版本状态，提前准备好经过完整测试的兼容替代路径。

最后做一次可复现核对

一条实用的验收记录至少包含 Python 版本、zstd.zstd_version_info、输入字节数、压缩后字节数、耗时、解压摘要和消费端读取结果。这样压缩级别或基础镜像发生变化后，差异不会只停留在“感觉变慢了”。