Distributed Deduplication and Data Quality

"I read the same cookie banner four hundred million times. I can recite it in my sleep, and now, apparently, so can the model."

A Foundation Model Memorizing Boilerplate

The previous section built a corpus: it pulled documents from many shards of a web crawl, filtered the obvious garbage, and laid the result out for distributed loading. What it did not do is ask whether those documents are distinct. They are not. Real web crawls are saturated with duplication: the same news article syndicated across a hundred outlets, the same documentation page mirrored on a dozen hosts, the same license text and cookie banner and navigation menu stamped onto millions of unrelated pages. Studies of large open corpora routinely find that a substantial fraction of documents are exact or near-exact copies of other documents. If you train on the corpus as-is, the model spends a measurable share of its compute budget re-reading text it has already mastered, and the most-duplicated strings are exactly the ones it learns to reproduce word for word.

Deduplication is therefore not a cosmetic cleanup; it is a data-quality intervention that changes what the model learns. Removing duplicates reduces verbatim memorization (a privacy and a quality problem), it frees compute for genuinely novel text, and it removes a large class of accidental train-on-test leakage. The challenge is scale. We must find duplicates among billions of documents without ever materializing the quadratic set of all document pairs, and we must do it across a cluster because no single machine holds the corpus. The rest of this section shows how, then closes with the related and equally important job of decontaminating against benchmarks.

1. Why Duplicates Hurt a Trained Model Beginner

The first instinct is that a few repeated documents cannot matter much against a trillion tokens, but the arithmetic says otherwise. Under stochastic gradient descent, every appearance of a document contributes its gradient again, so a string that occurs $c$ times is effectively up-weighted by a factor of $c$ relative to a string that occurs once. A passage duplicated ten thousand times across mirrored pages is not background noise; it is a training signal the model sees ten thousand times more often than a unique sentence, and the model obliges by memorizing it exactly. This is the mechanism behind verbatim regurgitation, where a prompt's opening words trigger the model to reproduce a long copyrighted or private passage it saw many times during training.

Duplication also distorts the loss landscape in a subtler way. Held-out evaluation assumes the test distribution is disjoint from training, but if a near-duplicate of a held-out document sits in the training set, the model's apparent generalization is partly memorization, and your validation loss flatters the model. Deduplicating the corpus before splitting off a validation set is the only way to trust the split. Three distinct harms, then, all flow from the same root: wasted compute proportional to the duplication factor, memorization that scales with repetition count, and contaminated evaluation that hides the first two.

2. Exact Deduplication Is a Hash Away Beginner

Exact duplicates are the easy case and you should remove them first, because they are cheap to find and they are the bulk of the volume. Compute a strong content hash of each document (a 64-bit or 128-bit digest of the normalized text), then group documents by their hash. Every group with more than one member is a set of exact copies; keep one representative and drop the rest. This is a single distributed group-by, the same shuffle-by-key pattern that Chapter 6 built MapReduce around: map each document to its hash, shuffle so identical hashes land on the same reducer, and emit one survivor per key. The cost is linear in the number of documents, and the only communication is the hash plus a document pointer, not the document text itself.

Exact dedup alone is never enough, though, because the web does not produce exact copies, it produces near-copies. A mirrored page adds a different header; a syndicated article gains a publisher's footer; a forum post is quoted with one word changed. The content hashes of these documents differ completely even though a human would call them the same text. Catching them requires a notion of similarity, and a way to find similar documents without comparing every pair. That is the job MinHash and LSH were built for.

3. Near-Deduplication with MinHash and LSH Intermediate

We represent each document as a set: the collection of its word $k$-shingles, the overlapping windows of $k$ consecutive words. Two documents are near-duplicates when their shingle sets overlap heavily, measured by the Jaccard similarity

Computing $J$ for all pairs is the quadratic trap: a corpus of $n$ documents has $\binom{n}{2} \approx n^2/2$ pairs, which for a billion documents is half a quintillion comparisons. MinHash and LSH dismantle this in two moves. First, MinHash replaces each large shingle set with a small fixed-length signature whose agreement rate estimates $J$ without storing the sets. For a random hash function $h$, the probability that the minimum hash value over set $A$ equals the minimum over set $B$ is exactly the Jaccard similarity,

so concatenating the per-hash minima from $m$ independent hashes gives an $m$-dimensional signature whose fraction of matching coordinates is an unbiased estimate of $J$. Second, LSH banding turns those signatures into a candidate generator. Split each signature into $b$ bands of $r$ rows ($m = b \cdot r$), and hash each band into a bucket. Two documents become a candidate pair if they collide in any band. The probability that a pair with similarity $s$ becomes a candidate is

an S-shaped curve whose threshold you tune through the choice of $b$ and $r$: more rows per band sharpen the cutoff, more bands lower it. Only candidate pairs are ever compared in full, so the all-pairs cost collapses to roughly the number of true near-duplicate pairs plus a controllable trickle of false candidates. Figure 19.4.1 traces the whole path, from signatures through bands to the duplicate clusters that get collapsed, and on to the benchmark check that follows.

4. Running Dedup as a Distributed Job Intermediate

At corpus scale the three stages map cleanly onto a cluster. Signature computation is embarrassingly parallel: each worker reads its shard of documents and emits a signature per document, with no cross-worker communication. The LSH banding is the shuffle: workers emit (band-bucket key, document id) pairs, and the framework partitions by key so every document sharing a bucket lands on the same reducer. The reducer enumerates candidate pairs within each bucket, confirms them with the signature-agreement estimate, and emits the near-duplicate edges. A final connected-components pass over those edges groups documents into duplicate clusters, and one representative per cluster survives. Only signatures and document identifiers cross the network; the document text stays put on the shard that owns it, which keeps the communication volume small even when the corpus is enormous.

The sharding deserves care for two reasons. First, popular boilerplate produces gigantic buckets: a license string shared by ten million documents would dump ten million ids into one reducer and explode its candidate-pair count back toward quadratic. Production pipelines cap bucket sizes or pre-strip the worst boilerplate so no single key becomes a hot shard, the same skew problem that haunts every distributed group-by. Second, the connected-components step is itself a distributed graph computation when the duplicate graph spans shards, linking it to the distributed graph machinery of later parts. The code below implements the full pipeline in one process so you can watch every stage, then we name the libraries that run it across thousands of machines.

import hashlib, random, re

random.seed(7)
NUM_HASHES = 128            # signature length m
B, R = 32, 4               # LSH bands and rows, b*r = m; tunes the S-curve threshold

def shingles(text, k=3):
    """Word k-shingles: the set of overlapping k-word windows in a document."""
    words = re.findall(r"\w+", text.lower())
    if len(words) < k:
        return {" ".join(words)}
    return {" ".join(words[i:i + k]) for i in range(len(words) - k + 1)}

MAXH = (1 << 61) - 1                                        # a large Mersenne prime
coeffs = [(random.randrange(1, MAXH), random.randrange(0, MAXH)) for _ in range(NUM_HASHES)]
stable_hash = lambda s: int(hashlib.blake2b(s.encode(), digest_size=8).hexdigest(), 16)

def minhash_signature(shingle_set):
    base = [stable_hash(s) for s in shingle_set]            # hash each shingle once
    return tuple(min((a * h + b) % MAXH for h in base)      # per-hash minimum
                 for a, b in coeffs)

# --- LSH banding: emit (band, band-slice) keys; documents that share a key collide.
buckets = {}
for doc_id, sig in signatures.items():                     # signatures: doc_id -> tuple
    for band in range(B):
        key = (band, sig[band * R:(band + 1) * R])
        buckets.setdefault(key, []).append(doc_id)

# Only documents sharing a bucket are ever compared; confirm with signature agreement.
candidate_pairs = {tuple(sorted((ids[i], ids[j])))
                   for ids in buckets.values()
                   for i in range(len(ids)) for j in range(i + 1, len(ids))}

Running the complete pipeline on a small corpus with deliberately injected near-duplicates, exact copies, and one boilerplate-padded mirror produces the output below. The corpus has twelve documents, so the all-pairs check would be sixty-six comparisons; LSH proposes only five candidates, confirms four genuine duplicate pairs, and the token count drops by nearly a third.

=== MinHash + LSH near-dedup ===
documents in corpus      : 12
all-pairs comparisons     : 66 (LSH skipped these)
LSH candidate pairs       : 5
confirmed duplicate pairs : 4
    base-1    ~ mirror-3   est_jaccard=0.67  true_jaccard=0.64
    base-2    ~ exact-1    est_jaccard=1.00  true_jaccard=1.00
    base-3    ~ mirror-2   est_jaccard=0.54  true_jaccard=0.52
    base-5    ~ exact-2    est_jaccard=1.00  true_jaccard=1.00
documents kept            : 8  (base-0, base-4, exact-1, exact-2, mirror-1, mirror-2, mirror-3, unique-1)
documents removed         : 4  (base-1, base-2, base-3, base-5)
tokens before / after     : 185 / 131
token reduction           : 29.2%

=== Benchmark decontamination ===
benchmark items            : 1
training docs scanned      : 9
contaminated docs removed  : 2  (exact-2, leak-1)
    exact-2   overlap with benchmark = 1.00
    leak-1    overlap with benchmark = 0.62
clean training docs kept   : 7

Two details in the output reward a second look. The estimated Jaccard column sits within a few hundredths of the true Jaccard column, which is the MinHash guarantee in action: a 128-dimensional signature estimates similarity well enough to make confident keep-or-drop decisions without ever storing the shingle sets. And the candidate count, five against sixty-six possible pairs, is the whole point of LSH at scale: the comparison work shrank by an order of magnitude even on this toy corpus, and the gap widens enormously as the corpus grows.

# pip install datasketch
from datasketch import MinHash, MinHashLSH

lsh = MinHashLSH(threshold=0.8, num_perm=128)      # S-curve cutoff and signature length
sigs = {}
for doc_id, text in corpus:                        # corpus: iterable of (id, text)
    m = MinHash(num_perm=128)
    for sh in shingles(text):                      # update with each k-shingle
        m.update(sh.encode())
    lsh.insert(doc_id, m)                           # index handles the banding
    sigs[doc_id] = m

near_dups = {d: lsh.query(sigs[d]) for d, _ in corpus}   # candidates per document

5. Decontamination: Keeping the Test Set Out Advanced

Deduplication removes copies within the corpus; decontamination removes a specific, dangerous subset: documents that overlap with the benchmarks you will use to evaluate the model. If a copy of a benchmark question and its answer has been scraped into the training data, the model can memorize it, and your reported score measures recall of training data rather than genuine capability. This is the train-test leakage problem of Chapter 8, raised to corpus scale, and it is exactly the benchmark-contamination pitfall that Chapter 5 warns makes leaderboard numbers untrustworthy. As benchmarks and web crawls both grow, the odds that some test item already lives somewhere in the crawl approach certainty, so decontamination is now a mandatory step, not an optional courtesy.

Mechanically, decontamination reuses the near-dedup machinery with the benchmark set as the reference. Build signatures for every benchmark item, then flag any training document whose similarity to a benchmark item exceeds a threshold and drop it before training. The second half of Output 19.4.1 shows this on the deduplicated corpus: a planted leak-1 document (a benchmark answer padded with site text) is caught at overlap 0.62, and exact-2, a survivor that verbatim-matches the benchmark string, is caught at overlap 1.00. Both leave the corpus, and the evaluation that follows is honest. The choice of threshold and shingle size is a policy decision: too loose and you delete legitimately related training text, too tight and you let paraphrased leaks through. Published recipes lean toward aggressive removal, on the principle that a slightly smaller training set is a cheap price for a trustworthy benchmark.

With duplicates removed and benchmarks held out, the corpus is finally ready to become model input. The next step turns clean text into the integer token streams a transformer consumes, a job that is itself distributed when the vocabulary is trained on a trillion tokens. Section 19.5 takes up tokenization at scale, picking up the clean corpus this section produced and the data-parallel all-reduce that Chapter 15 will feed it into.

6. The Deduplication Pipeline Intermediate

It is worth pausing to ask why deduplication is required at all before going further. The answer has three parts, each tied to a distributed-systems concern. Training on duplicated text is a triple waste: it spends compute on text the model has already absorbed (wasted FLOP budget), it trains the model to memorize exact strings rather than learn general patterns (degraded generalization), and it risks leaking benchmark answers that were scraped alongside ordinary web pages (compromised evaluation). The analogy is concrete: studying for an exam from ten identical copies of the same textbook is worse than studying from ten different textbooks, not just because of the effort wasted re-reading, but because the repetition teaches you to recall pages verbatim instead of understanding the subject. Dedup is the act of collapsing those ten identical copies into one so the model's update steps match the underlying content distribution rather than the web's copying habits.

At corpus scale this motivation translates into a pipeline with distinct stages, each of which maps onto a distributed primitive. The following working example brings together all the machinery described in the earlier sections: shingling, MinHash signatures, LSH banding, candidate confirmation, and cluster-based removal. It runs on a 500-document synthetic corpus with duplicates seeded at a controlled rate, reports how many documents were removed, and prints the final corpus size. Code 19.4.3 shows the full implementation.

"""
Code 19.4.3: Full near-dedup pipeline on a 500-document corpus.
Similarity threshold: 0.8. Reports duplicates removed and final size.
"""
import hashlib, random, re
from collections import defaultdict

random.seed(42)

# ---- Corpus generation: 400 unique docs + 100 near-duplicates --------
BASE_SENTENCES = [
    "distributed training requires careful gradient synchronization across workers",
    "tokenization converts raw unicode text into integer sequences",
    "the attention mechanism scales quadratically with sequence length",
    "data parallelism replicates the model on every accelerator",
    "pipeline parallelism splits layers across multiple devices",
]

def make_doc(idx):
    sent = BASE_SENTENCES[idx % len(BASE_SENTENCES)]
    words = sent.split()
    extra = f"document {idx} discusses " + " ".join(
        random.choices(["scaling", "compute", "bandwidth", "memory", "latency"], k=4)
    )
    return extra + " . " + " ".join(words) + " ."

def perturb(text):
    """Return a near-duplicate by swapping two adjacent words."""
    words = text.split()
    i = random.randrange(len(words) - 1)
    words[i], words[i + 1] = words[i + 1], words[i]
    return " ".join(words)

corpus = {}
for i in range(400):
    corpus[f"doc-{i}"] = make_doc(i)
for i in range(100):
    # near-duplicate of a random base document
    src_id = f"doc-{random.randrange(400)}"
    corpus[f"dup-{i}"] = perturb(corpus[src_id])

# ---- Shingling ----------------------------------------------------------
def shingles(text, k=3):
    words = re.findall(r"\w+", text.lower())
    if len(words) < k:
        return frozenset([" ".join(words)])
    return frozenset(" ".join(words[i:i+k]) for i in range(len(words)-k+1))

# ---- MinHash signatures (m=128) ----------------------------------------
NUM_HASHES = 128
MAXH = (1 << 61) - 1
coeffs = [(random.randrange(1, MAXH), random.randrange(0, MAXH))
          for _ in range(NUM_HASHES)]

def stable_hash(s):
    return int(hashlib.blake2b(s.encode(), digest_size=8).hexdigest(), 16)

def minhash_sig(doc_text):
    hvals = [stable_hash(sh) for sh in shingles(doc_text)]
    if not hvals:
        return tuple(MAXH for _ in range(NUM_HASHES))
    return tuple(min((a * h + b) % MAXH for h in hvals) for a, b in coeffs)

signatures = {doc_id: minhash_sig(text) for doc_id, text in corpus.items()}

# ---- LSH banding (b=32, r=4; threshold ~0.8) ---------------------------
B, R = 32, 4

buckets = defaultdict(list)
for doc_id, sig in signatures.items():
    for band in range(B):
        key = (band, sig[band*R:(band+1)*R])
        buckets[key].append(doc_id)

candidate_pairs = set()
for ids in buckets.values():
    for i in range(len(ids)):
        for j in range(i+1, len(ids)):
            candidate_pairs.add(tuple(sorted([ids[i], ids[j]])))

# ---- Confirm candidates and build duplicate graph ----------------------
THRESHOLD = 0.8

def jaccard_from_sigs(s1, s2):
    return sum(a == b for a, b in zip(s1, s2)) / NUM_HASHES

edges = [(a, b) for a, b in candidate_pairs
         if jaccard_from_sigs(signatures[a], signatures[b]) >= THRESHOLD]

# ---- Union-Find clustering ---------------------------------------------
parent = {d: d for d in corpus}

def find(x):
    while parent[x] != x:
        parent[x] = parent[parent[x]]
        x = parent[x]
    return x

def union(x, y):
    parent[find(x)] = find(y)

for a, b in edges:
    union(a, b)

# Keep one representative per cluster (lexicographically smallest id)
clusters = defaultdict(list)
for d in corpus:
    clusters[find(d)].append(d)

kept = {sorted(members)[0] for members in clusters.values()}
removed = set(corpus) - kept

print(f"original corpus size : {len(corpus)} documents")
print(f"LSH candidate pairs  : {len(candidate_pairs)}")
print(f"confirmed dup pairs  : {len(edges)}")
print(f"duplicates removed   : {len(removed)}")
print(f"final corpus size    : {len(kept)} documents")
print(f"retention rate       : {100*len(kept)/len(corpus):.1f}%")

On the 500-document corpus above (400 unique plus 100 seeded near-duplicates) the pipeline produces the following output:

original corpus size : 500 documents
LSH candidate pairs  : 214
confirmed dup pairs  : 97
duplicates removed   : 83
final corpus size    : 417 documents
retention rate       : 83.4%

7. Quality Filtering Intermediate

Deduplication removes copies; quality filtering removes text that is present only once but is still unsuitable for training. From the distributed-systems angle, quality filtering is another distributed-map job: each worker receives a shard of deduplicated documents, applies a sequence of increasingly expensive tests, and emits a keep-or-reject decision per document without coordinating with other workers. The tests are ordered cheapest first so that the bulk of rejections happen before any heavy computation runs.

Three stages cover the overwhelming majority of practical filtering needs. The first stage applies heuristic rules that can be computed in microseconds from simple text statistics: average line length (very short average lines signal navigation menus or code-snippet dumps), symbol-to-word ratio (a high fraction of punctuation or unicode symbols signals non-prose), and stop-word coverage (a document with almost no common function words is likely a list of proper nouns or a table, not fluent prose). The second stage applies a language-identification (LangID) classifier such as fastText's 176-language model, keeping only documents that exceed a confidence threshold for the target language or language mix. The third stage runs a perplexity filter: score each document against a small n-gram language model (KenLM trained on a clean reference corpus) and reject documents whose perplexity is extremely high (suggesting incoherent text) or extremely low (suggesting templated boilerplate). Code 19.4.4 shows a pipeline that applies all three stages and returns a per-document decision.

"""
Code 19.4.4: Three-stage quality filter pipeline.
Heavy libraries (fasttext, kenlm) are imported via comments;
the structure and decision logic are fully executable with stubs.
"""
import re, math

# Stage 1: heuristic rules (no external libraries needed)
def heuristic_ok(text):
    lines = [l for l in text.splitlines() if l.strip()]
    if not lines:
        return False
    avg_line_len = sum(len(l) for l in lines) / len(lines)
    if avg_line_len < 20:            # likely navigation / menu text
        return False
    words = re.findall(r"\w+", text)
    if not words:
        return False
    symbol_ratio = len(re.findall(r"[^A-Za-z0-9\s]", text)) / max(len(text), 1)
    if symbol_ratio > 0.25:          # too many symbols: code dump or spam
        return False
    STOPWORDS = {"the", "a", "an", "is", "in", "of", "to", "and", "that", "it"}
    sw_coverage = sum(1 for w in words if w.lower() in STOPWORDS) / len(words)
    if sw_coverage < 0.04:           # almost no function words: not prose
        return False
    return True

# Stage 2: language identification
# In production: import fasttext; lid_model = fasttext.load_model("lid.176.bin")
# lid_model.predict(text) returns (label, confidence)
def langid_ok(text, target_lang="en", threshold=0.65):
    # Stub: always returns True for demonstration
    # Replace with: label, conf = lid_model.predict(text.replace("\n", " "))
    #               return label[0] == f"__label__{target_lang}" and conf[0] >= threshold
    return True

# Stage 3: perplexity filter against a small KenLM reference model
# In production: import kenlm; lm = kenlm.Model("reference.arpa")
# lm.perplexity(text) returns bits-per-word perplexity
PERPLEXITY_LOW  = 10     # suspiciously low: templated boilerplate
PERPLEXITY_HIGH = 1500   # suspiciously high: incoherent noise

def perplexity_ok(text):
    # Stub: always returns True for demonstration
    # Replace with: ppl = lm.perplexity(text)
    #               return PERPLEXITY_LOW < ppl < PERPLEXITY_HIGH
    return True

# --- Full pipeline -------------------------------------------------------
def quality_filter(documents):
    """
    documents: list of (doc_id, text) tuples
    Returns: dict mapping doc_id -> {"keep": bool, "stage_rejected": str or None}
    """
    results = {}
    for doc_id, text in documents:
        if not heuristic_ok(text):
            results[doc_id] = {"keep": False, "stage_rejected": "heuristic"}
            continue
        if not langid_ok(text):
            results[doc_id] = {"keep": False, "stage_rejected": "langid"}
            continue
        if not perplexity_ok(text):
            results[doc_id] = {"keep": False, "stage_rejected": "perplexity"}
            continue
        results[doc_id] = {"keep": True, "stage_rejected": None}
    return results

# --- Example run on five representative documents -----------------------
sample_docs = [
    ("doc-1", "Home | About | Contact | Blog | FAQ | Privacy Policy | Terms"),
    ("doc-2", "Distributed training requires careful gradient synchronization. "
              "The all-reduce operation aggregates gradients across all workers "
              "so that each device applies an identical parameter update."),
    ("doc-3", "!!! @#$%^& *** BUY NOW *** click HERE >>> limited OFFER $$$"),
    ("doc-4", "Le modele de langage necessite une tokenisation robuste pour "
              "traiter les textes multilingues correctement."),
    ("doc-5", "cat dog run jump blue red fast slow big small"),  # no stopwords
]

decisions = quality_filter(sample_docs)
print(f"{'doc_id':<10} {'keep':<6} {'rejected_at':<12}")
print("-" * 30)
for doc_id, info in decisions.items():
    stage = info["stage_rejected"] or "kept"
    print(f"{doc_id:<10} {str(info['keep']):<6} {stage:<12}")

Running Code 19.4.4 on the five sample documents produces the following decisions, illustrating how each stage catches a different failure mode:

doc_id     keep   rejected_at
------------------------------
doc-1      False  heuristic
doc-2      True   kept
doc-3      False  heuristic
doc-4      True   kept
doc-5      False  heuristic

In practice, applied to a large dedup'd corpus, these three stages produce retention rates that vary substantially by content type. The table below gives representative figures from published open-data pipelines on CommonCrawl-derived corpora.

The cumulative retention of roughly 35 to 40% is consistent with published reports: large open corpora typically retain around one-third of raw CommonCrawl documents after quality filtering, with the exact fraction depending on the aggressiveness of the thresholds.

8. Cost Model for Deduplication and Quality Filtering Intermediate

Both dedup and quality filtering are distributed-map problems, so their wall-clock cost on a cluster is determined by throughput per node and the number of nodes available. On a 10-trillion-token corpus (roughly 50 TB of text at an average of five bytes per token), MinHash deduplication processes documents at approximately 1 GB/s per worker when signature computation and LSH bucketing are pipelined. A single-machine run at that rate would take $50{,}000 \div 1 \approx 50{,}000$ seconds, around 14 hours per pass. With 5 workers the wall-clock drops to roughly 2.8 hours; with 20 workers it reaches about 40 minutes. Quality filtering, with its three-stage pipeline (heuristic, LangID, KenLM scoring), achieves a lower effective throughput because KenLM perplexity scoring is more expensive than hash computation: around 200 kB/s per worker on CPU. The same 50 TB corpus at 200 kB/s on a single machine would take $50 \times 10^9 \div 200{,}000 \approx 250{,}000$ seconds, roughly 69 hours; with 5 workers this falls to about 14 hours. The numbers make the priority ordering obvious: run exact dedup (fastest) first, then near-dedup with MinHash, then quality filtering only on what remains. Each stage reduces the input to the next, compounding the savings. This is exactly the distributed-map pattern from Chapter 6: the corpus is partitioned across workers with no cross-worker communication until the final cluster-collapse step, so adding nodes gives near-linear speedup right up to the point where network I/O or the cluster manager becomes the bottleneck.