How to Shrink Small JSON Records in Python With Zstandard Dictionaries

Many services end up storing a lot of small documents: one JSON record per cache key, one per database row, one per queued message. Compress each of them on its own and the savings are disappointing. In this tutorial’s lab, gzip shrank 1,746 small JSON records by only 1.66 times, even though compressing the very same records as one big file shrank them 8.26 times. Zstandard dictionaries win back a large part of that gap, and since Python 3.14 Zstandard is part of the standard library, so you can use them without installing anything.

By the end you will have trained a dictionary, measured it honestly on records it never saw, built a small tested codec that can rotate dictionaries safely, and stored the result in SQLite. On the way we will step on three traps that are easy to fall into: a scoring mistake that inflates the ratio from 3.34 to 5.21 times, a default that can make compression an order of magnitude slower than it needs to be, and a silent corruption gap that a four-byte checksum closes.

What you will build

1. A script that turns the CISA Known Exploited Vulnerabilities catalog into 1,746 small JSON records.
2. Measurements that show why small records compress badly.
3. A trained dictionary, scored on newer records that it never saw.
4. A feel for how big the dictionary should be, what to train it on, and how to avoid flattering numbers.
5. A look inside a compressed frame, and what the errors look like when a dictionary is wrong or missing.
6. A checksum that turns silent corruption into an error.
7. The fast way to load a dictionary, and the slow default to avoid.
8. A small codec class with dictionary rotation, a decompression size limit, and a pytest suite.
9. An end-to-end test that stores the records in SQLite and compares file sizes and read times.

Prerequisites

• Python 3.14 or newer, which ships the compression.zstd module. I ran the lab on Python 3.15.0 on Windows 11; nothing in the code is Windows specific.
• On Python 3.10 to 3.13, install the backport with python -m pip install backports.zstd. The backports.zstd package describes itself as a “Backport of compression.zstd”. I also ran the whole lab on Python 3.13.14 with backports.zstd 1.8.0 and every Zstandard number came out identical.
• pytest for Step 8 (python -m pip install pytest; I used 9.1.1).
• curl, to download a 1.8 MB file. The lab machine already had curl.exe; on macOS and Linux use curl.
• Skills: you can read basic Python and you know the difference between str and bytes. You do not need to know anything about compression.

Set up the lab folder and the data

The data is the CISA Known Exploited Vulnerabilities (KEV) catalog. It suits this lesson because it is a real list of small, similar JSON documents, and CISA’s GitHub mirror states that “The KEV database is distributed under the Creative Commons 0 1.0 License.” The live catalog grows every few days, so we download a copy pinned to one commit of the mirror. That keeps your numbers identical to the ones printed here.

mkdir zstd-lab
cd zstd-lab
curl.exe -L -o kev.json https://raw.githubusercontent.com/cisagov/kev-data/7923b5f7040256ca9887d268b4c2a02453a2ff30/known_exploited_vulnerabilities.json
python -c "from compression import zstd; print(zstd.zstd_version)"

On macOS or Linux type curl instead of curl.exe. The last command should print 1.5.7, the version of the Zstandard library inside Python. The downloaded file should be 1,789,360 bytes and its SHA-256 should start with 0a4807c4. If the import fails with ModuleNotFoundError, you are on Python 3.13 or older: install the backport and use from backports import zstd in that command.

The idea in plain words

Why small records compress badly

A large part of what a compressor does is spot repetition. When a phrase shows up a second time, the compressor writes a short pointer back to the first copy instead of the phrase. The Zstandard README puts the consequence for small inputs plainly: “The smaller the amount of data to compress, the more difficult it is to compress.” It also gives the reason: “compression algorithms learn from past data how to compress future data.”

A single 800-byte JSON record has almost no past. The key vendorProject appears once, the phrase “which could allow” appears once, and so on, so most of the record has nothing earlier to point back to. Put 1,746 such records in one file and the second record can point back into the first, the third into the first two, and the ratio climbs. The same README says that to solve this, Zstandard offers a training mode, and that “Using this dictionary, the compression ratio achievable on small data improves dramatically.”

What a dictionary changes

A dictionary is a small blob of typical content, 16 KiB in our first example, that the compressor and the decompressor both load before the first byte of a record. It gives every record a ready-made past: the first vendorProject in a record can point back into the dictionary instead of being spelled out. Python’s documentation describes the benefit this way: “Zstandard dictionaries enable more efficient compression of smaller sizes of data, which is traditionally difficult to compress due to less repetition.”

What it costs you

The dictionary becomes part of your data format. A blob written with a dictionary cannot be read without that exact dictionary; the documentation says “This must be the same Zstandard dictionary used during compression.” Lose the dictionary file and you lose the data. A large part of this tutorial is about handling that responsibly.

Words used below

• Compression ratio: original size divided by compressed size. 3x means a third of the bytes; bigger is better.
• Frame: one self-contained compressed unit: a small header, the compressed data, and an optional checksum.
• Training: giving Zstandard sample records so it can pick the byte sequences that will pay off most in the dictionary.
• Digested dictionary: a dictionary that has already been preprocessed for fast reuse. We will measure why that matters in Step 7.

Step 1: Turn the catalog into small records

What the code does

Create kevdata.py. Every later script imports from it. The try and except at the top use the standard library module on Python 3.14 or newer and fall back to the backport on older versions, so the same files run on both. load_records() reads the catalog, sorts the entries from oldest to newest by dateAdded, and turns each entry into compact JSON (no spaces after commas or colons, which is how a service would typically store it) encoded as UTF-8 bytes, because compressors work on bytes. split() returns the oldest 70 percent for training and the newest 30 percent for testing. Step 3 explains why.

# kevdata.py
import json
from pathlib import Path

try:
    from compression import zstd  # Python 3.14 and newer
except ImportError:  # Python 3.9 to 3.13: pip install backports.zstd
    from backports import zstd


def load_records(path="kev.json"):
    """One compact JSON document (as bytes) per catalog entry, oldest first."""
    catalog = json.loads(Path(path).read_text(encoding="utf-8"))
    entries = sorted(catalog["vulnerabilities"], key=lambda v: (v["dateAdded"], v["cveID"]))
    return [
        json.dumps(entry, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
        for entry in entries
    ]


def split(records, train_fraction=0.7):
    """Oldest records first for training, newest records last for testing."""
    cut = int(len(records) * train_fraction)
    return records[:cut], records[cut:]

Now create step01_records.py to look at the data:

# step01_records.py
import statistics

from kevdata import load_records, zstd

records = load_records()
sizes = [len(r) for r in records]

print("zstd library version:", zstd.zstd_version)
print("records:", len(records))
print("total bytes:", f"{sum(sizes):,}")
print("smallest, median, mean, largest:", min(sizes), statistics.median(sizes), round(statistics.mean(sizes)), max(sizes))
print()
print(records[0].decode("utf-8")[:330], "...")

Run it and check the result

python step01_records.py

zstd library version: 1.5.7
records: 1746
total bytes: 1,394,328
smallest, median, mean, largest: 463 725.5 799 2302

{"cveID":"CVE-2010-5326","vendorProject":"SAP","product":"NetWeaver","vulnerabilityName":"SAP NetWeaver Remote Code Execution Vulnerability","dateAdded":"2021-11-03","shortDescription":"SAP NetWeaver Application Server Java Platforms Invoker Servlet does not require authentication, allowing for remote code execution via a HTTP o ...

There are 1,746 records totaling about 1.4 MB, with a median of 725.5 bytes. That is small by compression standards: the typical record is a few hundred bytes of repeated key names and boilerplate sentences, and the largest is only 2,302 bytes. If your count is higher than 1,746, you downloaded today’s catalog instead of the pinned copy. Everything still works, but your numbers will differ from mine.

Step 2: Measure the small-data problem

What the code does

Before adding a dictionary, we need a baseline. step02_baseline.py compresses every record separately with gzip (the same deflate algorithm as zlib, plus a header and trailer), with zlib, and with Zstandard at its default level 3 and at level 19, a much stronger and slower setting. The last row joins all records into one stream with newline separators and compresses that once.

# step02_baseline.py
import gzip
import zlib

from kevdata import load_records, zstd

records = load_records()
raw_total = sum(len(r) for r in records)


def report(label, blobs):
    total = sum(len(b) for b in blobs)
    print(f"{label:<36}{total:>11,} bytes{raw_total / total:>8.2f}x{100 * total / raw_total:>8.1f}%")


print(f"{'':<36}{'total':>17}{'ratio':>8}{'of raw':>9}")
report("raw JSON, one record each", records)
report("gzip level 6, one record each", [gzip.compress(r, compresslevel=6) for r in records])
report("zlib level 6, one record each", [zlib.compress(r, 6) for r in records])
report("zstd level 3, one record each", [zstd.compress(r) for r in records])
report("zstd level 19, one record each", [zstd.compress(r, 19) for r in records])

everything = b"\n".join(records)
report("zstd level 3, all records at once", [zstd.compress(everything)])

Run it and check the result

python step02_baseline.py

                                                total   ratio   of raw
raw JSON, one record each             1,394,328 bytes    1.00x   100.0%
gzip level 6, one record each           839,019 bytes    1.66x    60.2%
zlib level 6, one record each           818,067 bytes    1.70x    58.7%
zstd level 3, one record each           850,259 bytes    1.64x    61.0%
zstd level 19, one record each          838,961 bytes    1.66x    60.2%
zstd level 3, all records at once       168,744 bytes    8.26x    12.1%

Why it matters

The four rows that compress one record at a time land between 1.64 and 1.70 times (the raw row is the 1.00x baseline). The last row compresses the same bytes as a single stream and reaches 8.26 times, about five times better than gzip managed per record. Same data, same algorithm family: the only difference is how much earlier text the compressor saw. Notice what did not help: raising Zstandard from level 3 to level 19 moved the ratio from 1.64 to 1.66. A stronger setting cannot find repetition that is not in the input.

Your gzip and zlib figures can differ slightly from mine because they depend on the zlib build inside your Python (on 3.13.14 I measured 1.68 and 1.72). The Zstandard rows were identical on both interpreters.

Step 3: Train your first dictionary

What the code does

zstd.train_dict(samples, dict_size) takes an iterable of sample bytes and the maximum dictionary size in bytes, and returns a ZstdDict object. We pass 16 KiB. The documentation notes that “The Zstandard documentation suggests an absolute maximum of no more than 100 KB, but the maximum can often be smaller depending on the data.”

The important design choice is how we test. A dictionary is trained once and then used on records that have not been written yet, so the honest test is to train on the past and score on the future. The catalog is ordered by time, which makes that easy: the oldest 1,222 records train the dictionary and the newest 524 records score it. The script prints the date ranges so you can see the split really is chronological.

# step03_dictionary.py
import json
import time

from kevdata import load_records, split, zstd

records = load_records()
train, test = split(records, 0.7)
test_raw = sum(len(r) for r in test)


def added(record):
    return json.loads(record)["dateAdded"]


print(f"training records were added {added(train[0])} to {added(train[-1])}")
print(f"test records were added {added(test[0])} to {added(test[-1])}")
started = time.perf_counter()
zd = zstd.train_dict(train, 16 * 1024)
elapsed = time.perf_counter() - started
print(f"trained on {len(train)} older records in {elapsed:.2f} s")
print("dictionary id:", zd.dict_id, "| size:", len(zd.dict_content), "bytes")
print(f"testing on the {len(test)} newest records ({test_raw:,} raw bytes)")
print()


def report(label, blobs):
    total = sum(len(b) for b in blobs)
    print(f"{label:<44}{total:>9,} bytes{test_raw / total:>7.2f}x{total / len(blobs):>6.0f} per record")


report("zstd level 3, no dictionary", [zstd.compress(r) for r in test])
with_dict = [zstd.compress(r, zstd_dict=zd) for r in test]
report("zstd level 3, 16 KiB trained dictionary", with_dict)

restored = [zstd.decompress(blob, zstd_dict=zd) for blob in with_dict]
print()
print("round trip identical:", restored == test)

Run it and check the result

python step03_dictionary.py

training records were added 2021-11-03 to 2024-11-21
test records were added 2024-11-25 to 2026-10-15
trained on 1222 older records in 0.04 s
dictionary id: 467776482 | size: 16384 bytes
testing on the 524 newest records (563,019 raw bytes)

zstd level 3, no dictionary                   325,477 bytes   1.73x   621 per record
zstd level 3, 16 KiB trained dictionary       180,181 bytes   3.12x   344 per record

round trip identical: True

What is happening

Training took 0.04 seconds, a trivial cost. On the 524 newest records, which the dictionary never saw, Zstandard without a dictionary reaches 1.73 times (621 bytes per record). With the 16 KiB dictionary it reaches 3.12 times (344 bytes per record), about 45 percent fewer bytes per record. The last line confirms that every record decompresses back to exactly the original, using the same dictionary object. A ZstdDict has two attributes you will use constantly: dict_content (the dictionary as bytes, ready to save to a file) and dict_id (a number identifying it, which Step 5 explains).

Step 4: Choose the dictionary size and training data, and avoid flattering yourself

Three decisions shape a dictionary: how big it is, which records train it, and how you score it. In this step we vary each one on purpose. The script is split into four labeled sections; part A comes first.

A. Dictionary size against number of training records

This section trains 21 dictionaries (7 sizes times 3 training-set sizes) and scores each on the newest 30 percent. The training records are always the newest N of the older 70 percent, the closest in time to the test data.

# step04_sweep.py
import random
import zlib

from kevdata import load_records, split, zstd

records = load_records()
older, newer = split(records, 0.7)


def ratio(test, zd=None):
    raw = sum(len(r) for r in test)
    packed = sum(len(zstd.compress(r, zstd_dict=zd)) for r in test)
    return raw / packed


print("=== A. dictionary size (rows) against number of training records (columns)")
print("   trained on the newest N of the older 70%, tested on the newest 30%")
counts = [100, 300, 1000]
print(f"{'':>10}" + "".join(f"{n:>9}" for n in counts))
for kib in [2, 4, 8, 16, 32, 64, 110]:
    cells = []
    for n in counts:
        zd = zstd.train_dict(older[-n:], kib * 1024)
        cells.append(f"{ratio(newer, zd):.2f}x")
    print(f"{kib:>7} KiB" + "".join(f"{c:>9}" for c in cells))
print(f"   no dictionary: {ratio(newer):.2f}x")
print()

=== A. dictionary size (rows) against number of training records (columns)
   trained on the newest N of the older 70%, tested on the newest 30%
                100      300     1000
      2 KiB    3.19x    3.14x    2.75x
      4 KiB    3.33x    3.28x    2.87x
      8 KiB    3.48x    3.40x    3.07x
     16 KiB    3.53x    3.55x    3.20x
     32 KiB    3.57x    3.68x    3.40x
     64 KiB    2.96x    3.81x    3.53x
    110 KiB    2.96x    3.42x    3.57x
   no dictionary: 1.73x

How to read the matrix:

• Every dictionary beats no dictionary. The baseline is 1.73 times; the worst cell is 2.75 times and the best is 3.81 times.
• Bigger is not automatically better. A 16 KiB dictionary from 300 records already reaches 3.55 times; the best cell, 64 KiB from 300 records, gains only 0.26 more. The Python documentation says “Larger dictionaries generally slow down compression, but improve compression ratios.”
• More training records did not help here. At 16 KiB the ratio goes 3.53, 3.55, then 3.20 as the training set grows from 100 to 300 to 1,000 records. The extra records are older; the drift check below shows why that matters.
• Too little material for a big dictionary hurts. With only 100 training records, the 64 KiB and 110 KiB dictionaries both fall to 2.96 times.

Does the data really drift?

A short script checks whether the newest records contain values that the training records never had:

# step04_drift.py
import json

from kevdata import load_records, split

records = load_records()
older, newer = split(records, 0.7)
old_docs = [json.loads(r) for r in older]
new_docs = [json.loads(r) for r in newer]

print(f"newest {len(new_docs)} records whose value never appears in the {len(old_docs)} training records")
for field in ["requiredAction", "vendorProject", "product"]:
    seen = {doc[field] for doc in old_docs}
    unseen = sum(1 for doc in new_docs if doc[field] not in seen)
    print(f"  {field:<15}{unseen:>4} of {len(new_docs)} ({100 * unseen / len(new_docs):.1f}%)")

python step04_drift.py

newest 524 records whose value never appears in the 1222 training records
  requiredAction  154 of 524 (29.4%)
  vendorProject   157 of 524 (30.0%)
  product         286 of 524 (54.6%)

Of the 524 newest records, 154 (29.4 percent) carry a requiredAction sentence that no training record uses, 157 (30.0 percent) name a vendor that no training record names, and 286 (54.6 percent) name a product that no training record names. A dictionary built from older records cannot contain words it never saw, and a bigger pile of older records adds more of the old vocabulary, not the new one.

The finding that more training records did not help deserves a caveat, because Zstandard’s own header file gives advice that looks like the opposite. In zdict.h the maintainers write “We generally recommend passing ~100x the size of the dictionary in samples” and “Having more samples will only improve the dictionaries effectiveness.” That holds when the samples resemble the data you will compress later. Our catalog drifts over time, as the check above shows, so adding older samples diluted the dictionary. The same header says “You should retrain a dictionary when its effectiveness drops. Dictionary effectiveness drops as the data you are compressing changes.” The practical lesson is to measure with your own data instead of trusting a rule of thumb, and to prefer recent samples when your data changes over time.

B. How you split the records changes the answer

Next, the same 16 KiB dictionary and the same 70/30 proportions, but one run splits by time and the other splits at random.

# step04_sweep.py (continued)
print("=== B. how the records were split (16 KiB dictionary, 70% to train)")
shuffled = records[:]
random.Random(7).shuffle(shuffled)
cut = int(len(shuffled) * 0.7)
for label, train, test in [
    ("oldest 70% / newest 30%", older, newer),
    ("random 70% / random 30%", shuffled[:cut], shuffled[cut:]),
]:
    zd = zstd.train_dict(train, 16 * 1024)
    print(f"{label:<26} no dictionary {ratio(test):.2f}x -> with dictionary {ratio(test, zd):.2f}x")
print()

=== B. how the records were split (16 KiB dictionary, 70% to train)
oldest 70% / newest 30%    no dictionary 1.73x -> with dictionary 3.12x
random 70% / random 30%    no dictionary 1.64x -> with dictionary 4.82x

The random split reports 4.82 times; the chronological split reports 3.12 times. A random split mixes the same weeks into both sets, so the dictionary has already seen the kind of wording the test records use. In production you do not get that luxury, because tomorrow’s records are not in today’s training set. The chronological number is the one to plan with.

C. The classic mistake: scoring on the training data

# step04_sweep.py (continued)
print("=== C. the mistake: measuring on the records the dictionary was trained on")
zd = zstd.train_dict(records, 110 * 1024)
print(f"110 KiB dictionary trained on all {len(records)} records, scored on the same records: {ratio(records, zd):.2f}x")
zd = zstd.train_dict(older, 110 * 1024)
print(f"110 KiB dictionary trained on the older records, scored on the newest: {ratio(newer, zd):.2f}x")
print()

=== C. the mistake: measuring on the records the dictionary was trained on
110 KiB dictionary trained on all 1746 records, scored on the same records: 5.21x
110 KiB dictionary trained on the older records, scored on the newest: 3.34x

Scoring the dictionary on the very records it was trained on gives 5.21 times. Scoring a dictionary of the same size on records it had not seen gives 3.34 times. The first number is not a measurement of anything you will experience, because a dictionary built from records contains pieces of those records verbatim. Any time a ratio looks wonderful, check that the test records were kept out of training.

D. A dictionary can be plain pasted bytes

What if you cannot use Zstandard dictionaries, or want to know what the trainer buys you? A dictionary can be nothing more than raw bytes, and zlib also accepts a preset dictionary through the zdict argument of zlib.compressobj(), with no trainer. The zlib manual advises “with the most commonly used strings preferably put towards the end of the dictionary” and warns that “the current implementation of deflate will use at most the window size minus 262 bytes of the provided dictionary”, so only about 32 KiB of a dictionary matters with Python’s default window. Zstandard can also load such untrained bytes as a “raw content” dictionary with ZstdDict(data, is_raw=True). This section pastes together the newest 32 KiB of training records and compares.

# step04_sweep.py (continued)
print("=== D. a dictionary pasted together by hand: the newest 32 KiB of training records")
pasted = b"".join(older)[-32 * 1024:]


def zlib_pack(record):
    packer = zlib.compressobj(6, zdict=pasted)
    return packer.compress(record) + packer.flush()


raw = sum(len(r) for r in newer)
print(f"zlib level 6 with the pasted bytes as zdict:  {raw / sum(len(zlib_pack(r)) for r in newer):.2f}x")
print(f"zstd level 3 with the same bytes as a raw dictionary: {ratio(newer, zstd.ZstdDict(pasted, is_raw=True)):.2f}x")
print(f"zstd level 3 with a 32 KiB trained dictionary (newest 300): {ratio(newer, zstd.train_dict(older[-300:], 32 * 1024)):.2f}x")

=== D. a dictionary pasted together by hand: the newest 32 KiB of training records
zlib level 6 with the pasted bytes as zdict:  3.49x
zstd level 3 with the same bytes as a raw dictionary: 3.07x
zstd level 3 with a 32 KiB trained dictionary (newest 300): 3.68x

Pasting recent records into zlib gives 3.49 times, the same bytes as a Zstandard raw dictionary give 3.07 times, and a trained 32 KiB Zstandard dictionary from the newest 300 records gives 3.68 times. The levels differ (zlib 6, Zstandard 3), so read the first two as being in the same range rather than as a ranking of the libraries. Two takeaways: the trainer did give the best result at the same size, and a dictionary made of recent, representative records is a respectable first version when a trainer is not available.

Step 5: See what a frame remembers and how it fails

What the code does

From now on we train with the newest 300 of the older records, the best-performing choice at 16 KiB. step05_frames.py has three sections. Part A reads the frame header with zstd.get_frame_info(). Part B decompresses with the wrong dictionary and with none. Part C compresses tiny payloads. Run the whole file with python step05_frames.py; the output is shown section by section below.

# step05_frames.py
from pathlib import Path

from kevdata import load_records, split, zstd

records = load_records()
older, newer = split(records, 0.7)
zd = zstd.train_dict(older[-300:], 16 * 1024)
other = zstd.train_dict(older[:300], 16 * 1024)
record = newer[0]

with_dict = zstd.compress(record, zstd_dict=zd)
without = zstd.compress(record)

print("=== A. what the frame header remembers")
for label, frame in [("with dictionary", with_dict), ("without dictionary", without)]:
    info = zstd.get_frame_info(frame)
    print(f"  {label:<19}{len(frame):>4} bytes  dictionary_id={info.dictionary_id}  decompressed_size={info.decompressed_size}")
print("  id of the trained dictionary:", zd.dict_id, "| id of the other one:", other.dict_id)
Path("kev.dict").write_bytes(zd.dict_content)
reloaded = zstd.ZstdDict(Path("kev.dict").read_bytes())
print("  saved to kev.dict and loaded again: same id", reloaded.dict_id == zd.dict_id, "| decodes", zstd.decompress(with_dict, zstd_dict=reloaded) == record)
print()

=== A. what the frame header remembers
  with dictionary     224 bytes  dictionary_id=1118333072  decompressed_size=912
  without dictionary  535 bytes  dictionary_id=0  decompressed_size=912
  id of the trained dictionary: 1118333072 | id of the other one: 254209780
  saved to kev.dict and loaded again: same id True | decodes True

The frame names its dictionary

The record compressed to 224 bytes with the dictionary and 535 bytes without. Notice dictionary_id=1118333072: the frame header records which dictionary the writer used, and it matches the trained dictionary’s dict_id. For a frame written without a dictionary the field reads 0; the Python documentation says “0 means the dictionary ID was not recorded in the frame header”. Careful with the two similar names: a ZstdDict has dict_id, a frame info has dictionary_id, and the documentation warns that “The meaning of 0 for ZstdDict.dict_id is different from the dictionary_id attribute to the get_frame_info() function.” We will only compare non-zero IDs.

This is the feature that makes dictionary rotation practical. RFC 8878, the Zstandard format specification, says the ID field is optional and that “When it’s not present, it’s up to the decoder to know which dictionary to use.” Python writes it by default, so a stored blob can tell you which dictionary it needs without a separate version column. The same RFC adds that “Within private environments, any dictionary ID can be used”; the IDs here fall between the ranges the RFC reserves for dictionaries registered with IANA (32,767 and below, and 2 to the power 31 and above). The script also saves the dictionary to kev.dict and loads it back, which proves dict_content is all you need to ship. The documentation notes that it “can be used with other programs, such as the zstd CLI program.”

B. Wrong dictionary and missing dictionary

# step05_frames.py (continued)
print("=== B. decompressing with the wrong dictionary or none")
for label, kwargs in [("no dictionary", {}), ("a different dictionary", {"zstd_dict": other})]:
    try:
        zstd.decompress(with_dict, **kwargs)
    except zstd.ZstdError as exc:
        print(f"  {label}: ZstdError: {exc}")
print()

=== B. decompressing with the wrong dictionary or none
  no dictionary: ZstdError: Unable to decompress Zstandard data: Dictionary mismatch
  a different dictionary: ZstdError: Unable to decompress Zstandard data: Dictionary mismatch

Both mistakes raise ZstdError with the same message, “Dictionary mismatch”. The message does not say which dictionary was needed, which is why the codec in Step 8 reads dictionary_id first and raises its own error that names the missing dictionary.

C. Tiny payloads grow

# step05_frames.py (continued)
print("=== C. tiny payloads")
for payload in [b"", b"x", b"{}", b'{"a":1}']:
    print(f"  {len(payload)} raw bytes -> {len(zstd.compress(payload))} bytes, with dictionary {len(zstd.compress(payload, zstd_dict=zd))} bytes")
print()

=== C. tiny payloads
  0 raw bytes -> 9 bytes, with dictionary 13 bytes
  1 raw bytes -> 10 bytes, with dictionary 14 bytes
  2 raw bytes -> 11 bytes, with dictionary 15 bytes
  7 raw bytes -> 16 bytes, with dictionary 20 bytes

A frame has a fixed header, so an empty payload becomes 9 bytes, and a 7-byte payload becomes 16 bytes (20 with the 4-byte dictionary ID). Do not compress values of a few dozen bytes without measuring; the header can cost more than the repetition saves.

Step 6: Turn on the checksum

What the code does

step06_checksum.py answers a question that is easy to skip: what happens if a stored blob gets damaged? Storage, copies, and buggy code can damage bytes. The script takes 20 records spread across the newest data, compresses each, then flips every bit of every blob one at a time and tries to decompress. It does this twice: with the default settings, and with the checksum_flag option, which the documentation describes as “A four-byte checksum using XXHash64 of the uncompressed content is written at the end of each frame.” With it, “Zstandard’s decompression code verifies the checksum. If there is a mismatch a ZstdError exception is raised.”

# step06_checksum.py
from kevdata import load_records, split, zstd

records = load_records()
older, newer = split(records, 0.7)
zd = zstd.train_dict(older[-300:], 16 * 1024)

print("flipping every bit of 20 stored records, one bit at a time")
sample = newer[:: len(newer) // 20][:20]
checksum_on = {zstd.CompressionParameter.checksum_flag: 1}
for label, options in [("no checksum (the default)", None), ("with checksum_flag", checksum_on)]:
    outcomes = {"ZstdError": 0, "wrong data, no error": 0, "same data": 0}
    stored_bytes = 0
    for original in sample:
        blob = zstd.compress(original, zstd_dict=zd, options=options)
        stored_bytes += len(blob)
        for position in range(len(blob)):
            for bit in range(8):
                damaged = bytearray(blob)
                damaged[position] ^= 1 << bit
                try:
                    restored = zstd.decompress(bytes(damaged), zstd_dict=zd)
                except zstd.ZstdError:
                    outcomes["ZstdError"] += 1
                else:
                    outcomes["same data" if restored == original else "wrong data, no error"] += 1
    trials = sum(outcomes.values())
    print(f"  {label}: {stored_bytes} stored bytes, {trials} flips")
    for name, count in outcomes.items():
        print(f"    {name:<22}{count:>7}  {100 * count / trials:5.1f}%")

Run it and check the result

python step06_checksum.py

flipping every bit of 20 stored records, one bit at a time
  no checksum (the default): 5405 stored bytes, 43240 flips
    ZstdError               17989   41.6%
    wrong data, no error    25192   58.3%
    same data                  59    0.1%
  with checksum_flag: 5485 stored bytes, 43880 flips
    ZstdError               43821   99.9%
    wrong data, no error        0    0.0%
    same data                  59    0.1%

Why it matters

Without the checksum, 25,192 of the 43,240 single-bit flips (58.3 percent) decompressed to wrong data without any error. Only 41.6 percent raised an exception, and the rest changed nothing. With the checksum, 43,821 of 43,880 flips raised ZstdError, none produced wrong data, and the 59 remaining flips left the decoded text unchanged, the same number as the run without a checksum. The price was 4 bytes per record (5,405 stored bytes grew to 5,485, about 1.5 percent).

This is a simulation of single-bit damage, not a model of every failure, and your storage layer may already verify integrity end to end. But if it does not, a silent wrong answer is far worse than an exception, and four bytes is cheap insurance. Our codec turns the checksum on.

Step 7: Avoid the speed trap

What the code does

Ratios are only half the story. step07_speed.py measures microseconds per record: it runs each approach over the 524 test records after a warm-up pass, seven times (three for the level table), and reports the median. The key comparison is how the dictionary is handed to compress(). The documentation explains: “Digesting a dictionary is a costly operation that occurs when loading a Zstandard dictionary.” It also says what happens by default: “If passing a ZstdDict without any attribute, an undigested dictionary is passed by default when compressing and a digested dictionary is generated if necessary and passed by default when decompressing.” In other words, plain zstd_dict=zd may repeat the costly step on every compress call, while zd.as_digested_dict asks for the preprocessed form.

# step07_speed.py
import statistics
import time

from kevdata import load_records, split, zstd

records = load_records()
older, newer = split(records, 0.7)
zd = zstd.train_dict(older[-300:], 16 * 1024)
test = newer


def microseconds_per_record(work, rounds=7):
    """Median over several passes of the whole test set, in microseconds per record."""
    work()  # warm-up pass
    passes = []
    for _ in range(rounds):
        started = time.perf_counter_ns()
        work()
        passes.append((time.perf_counter_ns() - started) / 1000 / len(test))
    return statistics.median(passes)


def compress_with(option):
    return lambda: [zstd.compress(r, zstd_dict=option) for r in test]


reused = zstd.ZstdCompressor(zstd_dict=zd)
FRAME = zstd.ZstdCompressor.FLUSH_FRAME

print("=== compress, microseconds per record (level 3)")
rows = [
    ("no dictionary", compress_with(None)),
    ("zstd_dict=zd (undigested when compressing)", compress_with(zd)),
    ("zstd_dict=zd.as_digested_dict", compress_with(zd.as_digested_dict)),
    ("one ZstdCompressor reused, FLUSH_FRAME", lambda: [reused.compress(r, FRAME) for r in test]),
]
for label, work in rows:
    print(f"  {label:<44}{microseconds_per_record(work):>8.1f}")

blobs = [zstd.compress(r, zstd_dict=zd) for r in test]

Run it and check the result

python step07_speed.py

=== compress, microseconds per record (level 3)
  no dictionary                                    8.4
  zstd_dict=zd (undigested when compressing)      59.0
  zstd_dict=zd.as_digested_dict                    4.9
  one ZstdCompressor reused, FLUSH_FRAME           2.9

With no dictionary, compressing a record took 8.4 microseconds. Passing zstd_dict=zd directly took 59.0 microseconds in the run printed above, and between 59.0 and 96.9 across the four runs I made while writing (77.3, 83.8, 96.9, and 59.0), against 4.9 microseconds for zd.as_digested_dict. That is 12 to 20 times slower than the digested dictionary, and 7 to 12 times slower than not using a dictionary at all, a bad surprise if you only watched the ratio. Reusing a single ZstdCompressor and ending each record’s frame with FLUSH_FRAME was fastest at 2.9 microseconds. I did not use it in the codec below, because I did not test sharing one compressor object between threads, and the one-shot call with a digested dictionary is nearly as fast.

Decompression and compression level

# step07_speed.py (continued)
print("=== decompress, microseconds per record")
plain_blobs = [zstd.compress(r) for r in test]
rows = [
    ("no dictionary", lambda: [zstd.decompress(b) for b in plain_blobs]),
    ("zstd_dict=zd (digested after the first use)", lambda: [zstd.decompress(b, zstd_dict=zd) for b in blobs]),
    ("a new ZstdDecompressor(zd) for every record", lambda: [zstd.ZstdDecompressor(zd).decompress(b) for b in blobs]),
]
for label, work in rows:
    print(f"  {label:<44}{microseconds_per_record(work):>8.1f}")

print()

=== decompress, microseconds per record
  no dictionary                                    2.6
  zstd_dict=zd (digested after the first use)      1.8
  a new ZstdDecompressor(zd) for every record      1.6

Decompression is the easy side. The default already digests the dictionary once, and decompressing with a dictionary (1.8 microseconds) was slightly faster than without (2.6), presumably because the compressed input is smaller; I did not dig into the reason.

# step07_speed.py (continued)
print("=== level against ratio and compress time, with the digested dictionary")
raw = sum(len(r) for r in test)
digested = zd.as_digested_dict
for level in [1, 3, 9, 19]:
    packed = sum(len(zstd.compress(r, level, zstd_dict=digested)) for r in test)
    us = microseconds_per_record(lambda: [zstd.compress(r, level, zstd_dict=digested) for r in test], rounds=3)
    print(f"  level {level:<3}{raw / packed:>6.2f}x{us:>9.1f} us per record")

=== level against ratio and compress time, with the digested dictionary
  level 1    3.36x      4.4 us per record
  level 3    3.55x      5.0 us per record
  level 9    3.87x     18.7 us per record
  level 19   3.94x    157.4 us per record

Level 3 is the sweet spot for records this size: level 19 gave 3.94 times instead of 3.55 times, about 11 percent better, but cost 157.4 microseconds instead of 5.0, roughly 31 times as long. Keep the default unless you measure a reason to change it.

Step 8: Build a small codec with rotation, a checksum, and a safe size limit

What the code does

Now we put the lessons into one class, kevcodec.py:

• It takes dictionaries oldest first and always writes with the newest, but keeps all of them in a registry keyed by dict_id so older blobs stay readable after you retrain.
• It digests the writing dictionary once (as_digested_dict) and turns on the checksum. The documentation says that when you need advanced options, the level argument must be omitted and the level set in options, so both go into one options dictionary.
• On read it asks the frame which dictionary it needs. A frame with ID 0 is read without one; an ID the registry does not know raises UnknownDictionary, an error you can alert on separately from corruption.
• It enforces max_output, a limit on how large the decompressed data may be. We will see why in a moment.

# kevcodec.py
from kevdata import zstd

CP = zstd.CompressionParameter


class UnknownDictionary(Exception):
    """The frame names a dictionary this codec was not given."""


class PayloadTooLarge(Exception):
    """The decompressed data would be bigger than max_output."""


class DictionaryCodec:
    """Compress small records with the newest dictionary and decompress them with whichever
    dictionary the frame header names. Pass the dictionaries oldest first, newest last."""

    def __init__(self, dictionaries, level=3, max_output=1_000_000):
        if not dictionaries:
            raise ValueError("pass at least one dictionary, newest last")
        self.max_output = max_output
        self._by_id = {d.dict_id: d for d in dictionaries}
        self._newest = dictionaries[-1].as_digested_dict  # digest once, not on every call
        self._options = {CP.compression_level: level, CP.checksum_flag: 1}

    def compress(self, data):
        return zstd.compress(data, options=self._options, zstd_dict=self._newest)

    def decompress(self, blob):
        info = zstd.get_frame_info(blob)
        dictionary = None
        if info.dictionary_id != 0:
            dictionary = self._by_id.get(info.dictionary_id)
            if dictionary is None:
                raise UnknownDictionary(f"frame needs dictionary {info.dictionary_id}")
        if info.decompressed_size is not None and info.decompressed_size > self.max_output:
            raise PayloadTooLarge(f"frame announces {info.decompressed_size} bytes")
        decoder = zstd.ZstdDecompressor(zstd_dict=dictionary)
        data = decoder.decompress(blob, self.max_output + 1)  # ask for one byte more than allowed
        if len(data) > self.max_output:
            raise PayloadTooLarge(f"more than {self.max_output} bytes")
        if not decoder.eof:
            raise zstd.ZstdError("incomplete frame")
        return data

Why a limit on the output?

Compressed data can be tiny and expand enormously. Any code that decompresses bytes it did not produce itself (uploads, network messages, cache values another service can write) needs an output limit. Python’s decompressor supports this directly: “If max_length is non-negative, the method returns at most max_length bytes of decompressed data.” The codec asks for max_output + 1 bytes, so receiving more than max_output proves the data is too large without ever producing all of it. It also checks decoder.eof, which is false when a frame was cut short.

The next script builds a stand-in for a hostile input, 64 MiB of zero bytes, and shows both ways a frame can arrive: with its size announced in the header, and from a streaming compressor, which does not know the size in advance.

# step08_guard.py
import time

from kevcodec import DictionaryCodec, PayloadTooLarge
from kevdata import load_records, split, zstd

records = load_records()
older, newer = split(records, 0.7)
codec = DictionaryCodec([zstd.train_dict(older[-300:], 16 * 1024)], max_output=1_000_000)

bomb = zstd.compress(bytes(64 * 1024 * 1024))  # 64 MiB of zero bytes
info = zstd.get_frame_info(bomb)
print(f"64 MiB of zero bytes compresses to {len(bomb):,} bytes ({64 * 1024 * 1024 // len(bomb):,} to 1)")
print("the frame header announces", f"{info.decompressed_size:,}", "bytes")

started = time.perf_counter()
try:
    codec.decompress(bomb)
except PayloadTooLarge as exc:
    print(f"codec refused it in {1000 * (time.perf_counter() - started):.2f} ms: {exc}")

# A frame written by a streaming compressor does not know its own size.
streamer = zstd.ZstdCompressor()
chunks = [streamer.compress(bytes(1024 * 1024)) for _ in range(64)] + [streamer.flush()]
streamed = b"".join(chunks)
print("streamed frame announces:", zstd.get_frame_info(streamed).decompressed_size)

started = time.perf_counter()
try:
    codec.decompress(streamed)
except PayloadTooLarge as exc:
    print(f"codec refused it in {1000 * (time.perf_counter() - started):.2f} ms: {exc}")

started = time.perf_counter()
data = zstd.decompress(streamed)  # what the one-shot function does without a limit
print(f"zstd.decompress with no limit produced {len(data):,} bytes in {1000 * (time.perf_counter() - started):.1f} ms")

python step08_guard.py

64 MiB of zero bytes compresses to 2,067 bytes (32,466 to 1)
the frame header announces 67,108,864 bytes
codec refused it in 0.01 ms: frame announces 67108864 bytes
streamed frame announces: None
codec refused it in 0.56 ms: more than 1000000 bytes
zstd.decompress with no limit produced 67,108,864 bytes in 58.4 ms

The 64 MiB input compresses to 2,067 bytes (32,466 to 1). The frame announces its size, so the codec refuses it in 0.01 milliseconds without decompressing anything. The streamed frame announces nothing (None), so the header check cannot help, but the output cap refuses it after about half a millisecond. For contrast, the plain zstd.decompress() with no limit produced all 67,108,864 bytes in 58.4 milliseconds, and because it returns the whole result as one bytes object, the memory it needs grows with the output: the same 2 KB trick can be scaled far beyond 64 MiB. Treat the announced size as a cheap early check and the cap on actual output as the real defense. If you also accept archives, How to Prevent Path Traversal in Python File Downloads and Archive Extraction covers the other half of safe archive handling.

Test it with pytest

A codec that guards data is worth a test suite that attacks it. Create test_kevcodec.py:

# test_kevcodec.py
import pytest

from kevcodec import DictionaryCodec, PayloadTooLarge, UnknownDictionary
from kevdata import load_records, split, zstd

RECORDS = load_records()
OLDER, NEWER = split(RECORDS, 0.7)
DICT_A = zstd.train_dict(OLDER[-300:], 16 * 1024)
DICT_B = zstd.train_dict(RECORDS[-300:], 16 * 1024)


def test_every_newer_record_round_trips():
    codec = DictionaryCodec([DICT_A])
    assert all(codec.decompress(codec.compress(r)) == r for r in NEWER)


def test_new_frames_name_the_newest_dictionary():
    codec = DictionaryCodec([DICT_A, DICT_B])
    assert zstd.get_frame_info(codec.compress(NEWER[0])).dictionary_id == DICT_B.dict_id


def test_old_frames_still_decode_after_rotation():
    old_blob = DictionaryCodec([DICT_A]).compress(NEWER[0])
    assert DictionaryCodec([DICT_A, DICT_B]).decompress(old_blob) == NEWER[0]


def test_unknown_dictionary_is_reported_by_name():
    blob = DictionaryCodec([DICT_A]).compress(NEWER[0])
    with pytest.raises(UnknownDictionary):
        DictionaryCodec([DICT_B]).decompress(blob)


def test_frames_written_without_a_dictionary_still_decode():
    assert DictionaryCodec([DICT_A]).decompress(zstd.compress(NEWER[0])) == NEWER[0]


def test_every_single_bit_flip_is_caught_or_harmless():
    codec = DictionaryCodec([DICT_A])
    blob = codec.compress(NEWER[0])
    for position in range(len(blob)):
        for bit in range(8):
            damaged = bytearray(blob)
            damaged[position] ^= 1 << bit
            try:
                restored = codec.decompress(bytes(damaged))
            except (zstd.ZstdError, UnknownDictionary, PayloadTooLarge):
                continue
            assert restored == NEWER[0]  # a flip that changed nothing


def test_truncated_frame_is_an_error():
    codec = DictionaryCodec([DICT_A])
    with pytest.raises(zstd.ZstdError):
        codec.decompress(codec.compress(NEWER[0])[:-10])


def test_garbage_is_an_error():
    with pytest.raises(zstd.ZstdError):
        DictionaryCodec([DICT_A]).decompress(b"this is not a zstd frame")


def test_announced_size_over_the_limit_is_refused():
    bomb = zstd.compress(bytes(8 * 1024 * 1024))
    with pytest.raises(PayloadTooLarge):
        DictionaryCodec([DICT_A], max_output=1_000_000).decompress(bomb)


def test_frame_without_an_announced_size_is_still_limited():
    streamer = zstd.ZstdCompressor()
    blob = b"".join([streamer.compress(bytes(1024 * 1024)) for _ in range(8)] + [streamer.flush()])
    assert zstd.get_frame_info(blob).decompressed_size is None
    with pytest.raises(PayloadTooLarge):
        DictionaryCodec([DICT_A], max_output=1_000_000).decompress(blob)


def test_output_exactly_at_the_limit_is_allowed():
    codec = DictionaryCodec([DICT_A], max_output=len(NEWER[0]))
    assert codec.decompress(codec.compress(NEWER[0])) == NEWER[0]


def test_codec_needs_a_dictionary():
    with pytest.raises(ValueError):
        DictionaryCodec([])

What each group of tests protects:

• test_every_newer_record_round_trips proves the basic contract on all 524 test records.
• The rotation tests show that new frames name the newest dictionary, that an old blob still decodes after you add a newer dictionary, and that a missing dictionary raises UnknownDictionary. Plain frames written without any dictionary still decode.
• test_every_single_bit_flip_is_caught_or_harmless repeats the Step 6 experiment against the codec: any flip must either raise an error or leave the data unchanged.
• The truncated, garbage, over-limit, streamed, and exactly-at-the-limit tests check the size guard from both sides: it refuses too much, and it does not refuse data that is exactly allowed.

python -m pytest -q test_kevcodec.py

............                                                             [100%]
12 passed in 0.09s

Step 9: Store the records in SQLite and measure the whole thing

What the code does

Finally, an end-to-end test. step09_sqlite.py writes the same 1,746 records into three SQLite files, one with the JSON as text, one with each record compressed alone by Zstandard, and one using the codec with a 16 KiB dictionary. It runs VACUUM so the file size is honest, then does 1,000 random lookups by CVE ID and reports the median time per lookup, including decompression. Each table is a plain CREATE TABLE kev (id TEXT PRIMARY KEY, body). The dictionary here was trained on the newest 300 of the stored records, so treat the sizes as plausible but slightly flattering.

# step09_sqlite.py
import json
import random
import sqlite3
import statistics
import time
from pathlib import Path

from kevcodec import DictionaryCodec
from kevdata import load_records, zstd

records = load_records()
codec = DictionaryCodec([zstd.train_dict(records[-300:], 16 * 1024)])

layouts = {
    "kev_json.db": lambda r: r.decode("utf-8"),
    "kev_zstd.db": lambda r: zstd.compress(r),
    "kev_dict.db": lambda r: codec.compress(r),
}
readers = {
    "kev_json.db": lambda v: v.encode("utf-8"),
    "kev_zstd.db": lambda v: zstd.decompress(v),
    "kev_dict.db": lambda v: codec.decompress(v),
}

ids = [json.loads(r)["cveID"] for r in records]
print(f"{'database':<14}{'file size':>12}{'bytes per row':>15}{'median read':>14}")
for name, pack in layouts.items():
    path = Path(name)
    path.unlink(missing_ok=True)
    db = sqlite3.connect(path)
    db.execute("CREATE TABLE kev (id TEXT PRIMARY KEY, body)")
    db.executemany("INSERT INTO kev VALUES (?, ?)", [(i, pack(r)) for i, r in zip(ids, records)])
    db.commit()
    db.execute("VACUUM")
    wanted = random.Random(11).sample(ids, 1000)
    timings = []
    for key in wanted:
        started = time.perf_counter_ns()
        (value,) = db.execute("SELECT body FROM kev WHERE id = ?", (key,)).fetchone()
        readers[name](value)
        timings.append((time.perf_counter_ns() - started) / 1000)
    db.close()
    size = path.stat().st_size
    print(f"{name:<14}{size:>12,}{size / len(records):>15.0f}{statistics.median(timings):>11.1f} us")

check = sqlite3.connect("kev_dict.db")
(value,) = check.execute("SELECT body FROM kev WHERE id = ?", (ids[-1],)).fetchone()
print()
print("newest row restored from the dictionary database:")
print(codec.decompress(value).decode("utf-8")[:150], "...")
check.close()

Run it and check the result

python step09_sqlite.py

database         file size  bytes per row   median read
kev_json.db      1,699,840            974       29.4 us
kev_zstd.db      1,024,000            586       32.0 us
kev_dict.db        438,272            251       31.5 us

newest row restored from the dictionary database:
{"cveID":"CVE-2026-21589","vendorProject":"Atlassian","product":"Multiple Products","vulnerabilityName":"Atlassian Multiple Products Files or Director ...

What this shows

The dictionary database is 3.9 times smaller than the plain JSON file (438,272 against 1,699,840 bytes) and 2.3 times smaller than the per-record Zstandard database (1,024,000 bytes). Per row that is 974, 586, and 251 bytes. The file shrinks less than the payload alone would suggest, because it also holds the primary-key index and page overhead. And reads cost about the same: the median lookup was 29.4 microseconds for JSON text, 32.0 for per-record Zstandard, and 31.5 with the dictionary, so decompression adds at most about 3 microseconds to the database lookup itself. If you want to go further with SQLite itself, How to Fix SQLite “database is locked” Errors in Python With WAL Mode, BEGIN IMMEDIATE, and Checkpoints covers concurrency. I tested the codec with SQLite only; it needs nothing from the store except the ability to hold bytes.

Common mistakes and how to catch them

Check that the whole thing works end to end

1. Run python step01_records.py and confirm 1,746 records.
2. Run step02_baseline.py through step09_sqlite.py in order (the file numbers match the step numbers, and Step 4 has two files, step04_sweep.py and step04_drift.py) and compare each printout with the blocks above. Zstandard ratios, byte counts, and file sizes must match exactly when you use the pinned data and the same Zstandard version (1.5.7); the gzip and zlib figures in Step 2 and Step 4 part D depend on the zlib build in your Python, and timings will differ on your machine, but the order of magnitude should hold.
3. Run python -m pytest -q test_kevcodec.py and confirm 12 passed.
4. Last, try your own data. Replace kev.json with a few thousand of your own records, keep the time-based split, and look at the ratio in Step 3 before you trust any claim, including the numbers in this article.

Where to go next

• Read the compression.zstd reference for finalize_dict() (which turns a hand-built raw dictionary into a regular one), ZstdFile for compressed files on disk, and the advanced compression parameters.
• The Python 3.14 release notes say that “support for reading and writing Zstandard compressed archives has been added to the tarfile, zipfile, and shutil modules.” See What’s New in Python 3.14 and PEP 784 for the design.
• Treat a dictionary like a schema: it is a versioned part of your data format. How to Evolve a Protobuf Schema Without Breaking Old Clients covers the same discipline for another format, and How to Stop a Cache Stampede in Python is a good next read if the records live in a cache.
• Plan retraining. Record the compression ratio you actually achieve, and when it drops, train a new dictionary on recent records, add it to the codec’s list, and let old blobs age out.