如何在Python中用自定义字典实现短字符串独立压缩？

问题背景 #

数百万条长度小于20字符的短字符串，单独用zlib或lz4压缩时，由于算法头部开销、缺乏上下文参考，压缩结果反而比原字符串更大。需要一种自适应字典压缩方案：先通过批量字符串训练生成自定义大小的字典，再用该字典对每个短字符串独立压缩，降低开销并提升压缩率。

可行方案：基于zlib或lz4的字典压缩 #

zlib和lz4均支持自定义字典压缩，其中lz4的字典压缩开销更低，更适配短字符串场景。以下是两种库的实现方式：

1. 基于lz4的自适应字典压缩实现 #

lz4提供了compress_with_dictionary和decompress_with_dictionary接口，可直接复用预训练字典。我们可以封装成你期望的DictionaryCompressor类：

import lz4.block
from typing import List

class DictionaryCompressor:
    def __init__(self, dictionary_size: int = 1_048_576):
        self.dictionary_size = dictionary_size
        self._dictionary = b""

    def update(self, s: bytes):
        # 将新字符串追加到字典，超过设定大小则截断尾部
        self._dictionary += s
        if len(self._dictionary) > self.dictionary_size:
            self._dictionary = self._dictionary[-self.dictionary_size:]

    def compress(self, s: bytes) -> bytes:
        # 使用训练好的字典压缩字符串
        return lz4.block.compress(s, dictionary=self._dictionary)

    def decompress(self, compressed: bytes) -> bytes:
        # 用相同字典解压
        return lz4.block.decompress(compressed, dictionary=self._dictionary)

# 示例用法
inputs = [b"hello world", b"foo bar", b"HELLO foo bar world", b"bar foo 1234", b"12345 barfoo"]
D = DictionaryCompressor(dictionary_size=1_000_000)
for s in inputs:
    D.update(s)

# 压缩测试
for s in inputs:
    compressed = D.compress(s)
    print(f"原字符串: {s}, 压缩后长度: {len(compressed)}, 原长度: {len(s)}")
    # 验证解压正确性
    assert D.decompress(compressed) == s

2. 基于zlib的自适应字典压缩实现 #

zlib通过compressobj()指定wbits参数启用字典压缩，需先训练字典并传入压缩器：

import zlib
from typing import List

class DictionaryCompressor:
    def __init__(self, dictionary_size: int = 1_048_576):
        self.dictionary_size = dictionary_size
        self._dictionary = b""

    def update(self, s: bytes):
        self._dictionary += s
        if len(self._dictionary) > self.dictionary_size:
            self._dictionary = self._dictionary[-self.dictionary_size:]

    def compress(self, s: bytes) -> bytes:
        # 初始化压缩器，传入训练好的字典
        compressor = zlib.compressobj(wbits=zlib.MAX_WBITS)
        compressor.set_dictionary(self._dictionary)
        return compressor.compress(s) + compressor.flush()

    def decompress(self, compressed: bytes) -> bytes:
        decompressor = zlib.decompressobj(wbits=zlib.MAX_WBITS)
        decompressor.set_dictionary(self._dictionary)
        return decompressor.decompress(compressed) + decompressor.flush()

# 示例用法同lz4版本

关键说明 #

• 字典训练逻辑：示例中直接将所有输入字符串追加到字典，超过设定大小则保留最新部分。如果需要更高效的字典，可统计所有字符串的高频子串生成代表性字典（不过对于短字符串，直接追加的方式已足够）。
• lz4 vs zlib：lz4的字典压缩速度更快，压缩后额外开销更小，更适合短字符串；zlib压缩率略高，但速度和开销不如lz4。
• Smaz的局限性：Smaz采用硬编码固定字典，无法根据数据集自适应调整，不适用于需要灵活适配不同短字符串集合的场景。

内容的提问来源于stack exchange，提问作者Basj