Python compression.zstd 怎样用 ZstdDict 压缩重复数据

如果接口每次只压缩几百字节到几 KB，却反复出现相同字段名、固定前缀或相似 JSON 结构，单独调用 Zstandard 往往没有足够上下文。Python 3.14 的 compression.zstd 可以先用一批相似样本训练 ZstdDict，再把它传给 compress() 和 decompress()。关键不是“给压缩算法加一个参数”这么简单，而是压缩端和解压端必须持有同一份字典。

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

最稳妥的做法是：用真实业务样本训练一次字典，把字典作为版本化资源与服务一起分发；压缩和解压都显式传入同一份 ZstdDict，并把字典缺失、版本不一致和模块不可用当成部署边界处理。

为什么小块重复数据适合 ZstdDict

Zstandard 字典的价值在于补充“小数据缺少上下文”的问题。官方文档说明，字典适合许多相似的小数据块；样本越接近真实请求，字典越可能捕捉到稳定的字段和片段。它不是所有压缩任务的通用加速器：大文件本身已经有足够重复内容，额外加载字典反而可能增加管理成本。

先把边界分成三层：样本集合只负责训练，ZstdDict 负责保存训练结果，压缩数据块只负责传输或存储。图中连线表达对象之间的静态关系，不表示已经在本机执行。

用相似样本训练 ZstdDict

train_dict(samples, dict_size) 接收可迭代的 bytes 样本，并返回 ZstdDict。样本不要随便拼一段演示文本，应该来自同一种数据契约，例如同一类事件、短 JSON 响应或固定格式的日志片段。dict_size 是字典最大字节数，官方文档建议绝对上限不超过 100 KB；更大的字典通常会增加压缩成本，过小又可能学不到足够的重复片段。

from compression import zstd

samples = [
    b'{"kind":"order","region":"east","items":[1,2]}',
    b'{"kind":"order","region":"east","items":[3,4]}',
    b'{"kind":"order","region":"west","items":[5,6]}',
]

try:
    # 样本应来自同一类小消息，避免把互不相关的数据混进字典。
    dictionary = zstd.train_dict(samples, dict_size=8 * 1024)
except zstd.ZstdError as exc:
    # 训练失败时保留原始异常，避免静默切换成未知的压缩格式。
    raise RuntimeError("无法训练 Zstandard 字典") from exc

训练阶段不需要把原始样本写进压缩帧。生产环境更应该保存训练输入的来源、字典版本和适用的数据类型，后续服务才能判断一份字典是否仍然匹配当前协议。

同一字典如何贯穿压缩与解压

得到字典后，可以把它传给模块级 compress() 与 decompress()。解压侧使用的字典必须与压缩侧相同；只要字典内容或版本不一致，就不能把“能解压”当作默认保证。

from compression import zstd

payload = b'{"kind":"order","region":"east","items":[7,8]}'

try:
    # 压缩端和解压端显式使用同一个字典对象，避免两端资源错配。
    compressed = zstd.compress(payload, level=5, zstd_dict=dictionary)
    restored = zstd.decompress(compressed, zstd_dict=dictionary)
except zstd.ZstdError as exc:
    # 线上应记录字典版本和消息版本，便于区分数据损坏与字典不匹配。
    raise RuntimeError("Zstandard 字典压缩或解压失败") from exc

if restored != payload:
    # 这是业务层的内容一致性判断，不要把压缩成功误当成数据正确。
    raise ValueError("解压后的内容与原始 payload 不一致")

模块级函数适合完整的内存块。要处理连续数据，可以创建 ZstdCompressor 或 ZstdDecompressor；前者在所有输入提交后需要调用 flush() 完成压缩帧。多帧输入时，官方文档提醒 ZstdDecompressor 不会像模块级 decompress() 那样透明处理全部帧，选错接口会造成边界行为不同。

高频调用时怎样减少字典加载开销

ZstdDict 提供 as_digested_dict 和 as_undigested_dict。官方文档说明，字典摘要化是有成本的，但在多次压缩或解压时复用已摘要化的字典可以降低重复加载开销。不要在每条消息到来时重新训练字典，也不要把摘要化对象和字典版本分开管理。

from compression import zstd

# 服务启动时准备一次；请求处理函数只复用这个对象。
digested_dictionary = dictionary.as_digested_dict

def compress_message(payload: bytes) -> bytes:
    # 重复调用复用已摘要化字典，避免每条消息都重新加载字典。
    return zstd.compress(payload, level=5, zstd_dict=digested_dictionary)

如果压缩级别会变化，需要重新考虑摘要化字典的复用策略，因为字典信息可能与压缩级别相关。把压缩级别和字典版本写入配置，而不是散落在调用点，排查会更直接。

安全边界：字典、输入和部署版本要一起管理

这里的“安全”主要是数据边界和故障可追踪性。字典本身不是可信输入的替代品，也不会自动阻止异常大的解压结果。解压来自外部的数据时，应结合消息大小上限、超时和资源预算；需要增量处理时，用解压器的 max_length 和 needs_input 等状态控制读取，不要无界地把输出累积到内存。

部署时至少记录三项：字典版本、训练样本对应的协议版本、压缩级别。字典随服务升级时，压缩端和解压端要有兼容窗口；如果旧消息可能滞留，就保留旧字典的读取能力，直到消息生命周期结束。

什么时候不值得使用字典压缩

常见问题

只在压缩端传入字典，解压端能自动识别吗？

不能把它当成自动识别机制。官方接口要求解压时提供用于压缩的同一份字典，应用需要自行完成字典分发和版本管理。

字典越大，压缩比就一定越好吗？

不一定。更大的字典可能改善压缩比，但也可能增加压缩速度和加载成本；应使用真实样本衡量，而不是只扩大 dict_size。

可以把 ZstdDict 放在每条消息里吗？

通常不划算。字典应作为可复用的版本化资源管理，随每条小消息重复携带会抵消压缩收益。

小结

compression.zstd 的字典压缩路径可以概括为：用相似 bytes 样本训练 ZstdDict，压缩和解压显式复用同一字典，高频场景复用摘要化字典，并把字典版本、消息协议和资源限制一起纳入部署设计。真正值得投入的场景，是“小块、相似、数量多”，而不是所有 zstd 数据。