当传递大小提示而不是压缩流时，zstandard 压缩如何表现？

Question

zstd compressor 可以在streaming模式下运行，或者可以预先给定要压缩的总大小（例如，这个Python binding中的size参数）

预先给定大小时，库的行为如何？它是更快，还是使用更少的内存或更有效地压缩？当您压缩比给定大小更多或更少的数据时会发生什么？

Answer 1

我针对西里西亚语料库的狄更斯文本测试了 python-zstandard 库。

无论大小已知或未知，压缩所需的时间大致相同。压缩器为这个 10MB 的文件生成相同数量的字节，外加一个 3 字节的标头。

如果您告诉压缩器错误的字节数，当输入多于或少于预期时它就会失败。

如果在压缩时不知道大小，则必须使用流式解压缩 API 而不是一次性 .decompress(bytes) A​​PI，但我可能缺少刷新帧/关闭帧命令。

我们选择 22 级，这样内存差异会更加明显。在更合理的级别 <= 19 中，压缩时内存使用量 < 100MB，解压缩时内存使用量 < 20MB - 说明为什么命令行工具使用标志来保护极端压缩级别。

根据 scalene profiler，在 22 级，

peak memory	function
267MB	oneshot
777MB	onestream
266MB	rightsize
774MB	multistream

decompression peak memory	function
9.9MB	one-shot decompression
128.5MB	streaming decompression, size unknown
19.3MB	streaming decompression, size known
(fails)	one-shot decompression, size unknown

"""
Test zstd with different options and data sizes.
"""

import pathlib
import zstandard
import time
import io
import contextlib


@contextlib.contextmanager
def timeme():
    start = time.monotonic()
    yield
    end = time.monotonic()
    print(f"{end-start}s")


# The Collected works of Charles Dickens from the Silesia corpus
uncompressed = pathlib.Path("dickens").read_bytes()

ZSTD_COMPRESS_LEVEL = 22


def oneshot():
    compressor = zstandard.ZstdCompressor(level=ZSTD_COMPRESS_LEVEL)
    with timeme():
        result = compressor.compress(uncompressed)
        print("One-shot", len(result))
        return result


def onestream():
    compressor = zstandard.ZstdCompressor(level=ZSTD_COMPRESS_LEVEL)
    with timeme():
        bio = io.BytesIO()
        with compressor.stream_writer(bio, closefd=False) as writer:
            writer.write(uncompressed)
            writer.close()
        print("One-stream", len(bio.getvalue()))
        return bio.getvalue()


def rightsize():
    compressor = zstandard.ZstdCompressor(level=ZSTD_COMPRESS_LEVEL)
    with timeme():
        bio = io.BytesIO()
        with compressor.stream_writer(
            bio, closefd=False, size=len(uncompressed)
        ) as writer:
            writer.write(uncompressed)
            writer.close()
        print("Right-size", len(bio.getvalue()))
        return bio.getvalue()


def multistream():
    compressor = zstandard.ZstdCompressor(level=ZSTD_COMPRESS_LEVEL)
    with timeme():
        bio = io.BytesIO()
        with compressor.stream_writer(bio, closefd=False) as writer:
            CHUNK = len(uncompressed) // 10
            for i in range(0, len(uncompressed), CHUNK):
                writer.write(uncompressed[i : i + CHUNK])
            writer.close()
        print("Chunked stream", len(bio.getvalue()))
        return bio.getvalue()


def wrongsize():
    # This one's easy - you get an exception
    compressor = zstandard.ZstdCompressor(level=ZSTD_COMPRESS_LEVEL)
    with timeme():
        bio = io.BytesIO()
        with compressor.stream_writer(
            bio, size=len(uncompressed) + 100, closefd=False
        ) as writer:
            writer.write(uncompressed)
            writer.close()

        print("Wrong-size", len(bio.getvalue()))


has_size = oneshot()

no_size = onestream()

rightsize()

multistream()

oneshot()


def d1():
    decompress = zstandard.ZstdDecompressor()
    assert uncompressed == decompress.decompress(has_size)


d1()


def d2():
    # the decompress.decompress() API errors with zstd.ZstdError: could not
    # determine content size in frame header
    decompress = zstandard.ZstdDecompressor().stream_reader(no_size)
    assert uncompressed == decompress.read()


d2()


def d3():
    # streaming decompression with sized input
    decompress = zstandard.ZstdDecompressor().stream_reader(has_size)
    assert uncompressed == decompress.read()


d3()