m04 — The Pretraining Data Pipeline

A fully worked design. Two petabytes of raw crawl in, fifteen trillion clean, deduplicated, deterministically-ordered tokens out — and a dataloader that 1,024 ranks can resume from mid-epoch without re-reading or skipping a single document.

The hard parts are not the ones people expect. Not throughput — the CPU cost of this whole pipeline is a few hundred cores. The hard parts are global deduplication at 15 billion documents and reproducible ordering under a changing world: a resumed run, a resized cluster, a fixed bug. Both are correctness problems disguised as engineering ones.

Table of Contents

• The Prompt
• 1. Requirements and Scope
• 2. Scale Numbers
• 3. API Surface
• 4. Data Model
• 5. High-Level Architecture
• 6. Deep Dive A: Global Deduplication at 15 Billion Documents
• 7. Deep Dive B: Deterministic Order and Exact Resumption
• 8. Failure and Recovery
• 9. Bottlenecks and Evolution
• 10. Tradeoffs Explicitly Rejected
• The Hostile Critique
• The Revision
• References

The Prompt

"Design the data pipeline for a pretraining run. We're targeting 15 trillion tokens from web crawl plus curated sources. It needs to be reproducible — if a run diverges we have to be able to tell whether it was the data."

"Tell whether it was the data" is the requirement. Everything else follows from it.

Reproducibility in a training pipeline is not a nice-to-have or a compliance box. When a multi-million-dollar run produces a worse model than the last one, the first question is what changed, and if the data pipeline cannot answer with certainty, the team burns weeks on ablations that a content hash would have resolved in an hour.

So the design target is: given a run ID, reconstruct the exact token sequence every rank saw, at every step. That is a much stronger requirement than "process the data correctly", and it constrains the shuffle, the resumption, and the versioning of every filter.

The second thing to say early: the compute for this pipeline is trivial compared to the training run it feeds. Tokenizing 100 TB is about 200 cores for three days (§2). The pipeline's cost is measured in engineer-months and in mistakes, not in CPU — which is why the design should spend its complexity on correctness and traceability rather than on throughput.

1. Requirements and Scope

Clarifying questions asked

"Is this a one-shot corpus build, or a continuously updated one?" Assumed: versioned snapshots, rebuilt every few months, with runs pinned to a snapshot. A continuously-updating corpus makes reproducibility impossible by construction — two runs started a day apart would see different data with no record of the difference. Immutability of a released snapshot is the foundational decision and everything else is easier once it is made.

"What's the token budget and the mixture?" Assumed 15T tokens: web ~70%, code ~15%, curated (books, papers, reference) ~10%, multilingual ~5%. The mixture weights are a research parameter that changes often, so the design must make re-mixing cheap — which means mixing happens at sampling time, not at corpus-build time. That is a load-bearing decision, made here, in the first five minutes.

"How exact does deduplication need to be?" Assumed: exact dedup mandatory (byte-identical documents), fuzzy dedup at ~0.7 Jaccard for near-duplicates. Not because 0.7 is magic, but because it is the operating point where LSH is cheap and it is the published choice in several corpora, so it is defensible and comparable.

"Do we need to remove eval sets?" Yes, and this is not optional — contamination invalidates every benchmark number the run produces, and it is discovered after the run, by an external party, in public. Treated as a required stage, not a filter option.

"Who consumes the output — one framework, or several?" Assumed one training framework, many runs. That lets the output format be optimized for sequential reads at a fixed sequence length rather than being generic.

Functional

1. Ingest raw crawl + curated sources; normalize to documents with provenance.
2. Filter: language ID, quality, safety, PII redaction.
3. Deduplicate: exact and fuzzy, globally across the whole corpus.
4. Decontaminate against a registry of eval sets.
5. Tokenize into fixed-length training sequences.
6. Emit an immutable, content-addressed, versioned snapshot.
7. Serve it to N data-parallel ranks with deterministic order and exact resumption.

Non-functional

Explicitly out of scope

• The tokenizer's training (BPE vocab construction). We consume a pinned tokenizer artifact.
• Data selection research (what mixture is best) — we make mixtures cheap to change, we do not choose them.
• RLHF / SFT data. Different scale, different provenance requirements, different pipeline.
• Legal/licensing determination. We record provenance so that decision is possible; we do not make it.

2. Scale Numbers

Output. 15T tokens. Vocab 128,256 → does not fit in uint16, so tokens are uint32:

15e12 tokens x 4 bytes = 60 TB tokenized

Say the uint16/uint32 thing out loud. A vocab of 65,535 or less halves the corpus to 30 TB and halves every read during training. A tokenizer decision made by the modelling team doubles the storage and I/O cost of the data platform, and nobody notices until someone does this arithmetic. It is exactly the kind of cross-team coupling this round is looking for.

Input. Web text yield after filtering is brutally low — 1–5% of raw crawl survives quality filtering and dedup. At 3%:

need ~60 TB of clean text  ->  ~2 PB of raw crawl

Documents. At ~1,000 tokens average: 15 billion documents.

Dedup, and why it is deep dive A:

all-pairs comparisons = (15e9)^2 / 2 = 1.1e20        <- impossible, by a factor of ~1e12
MinHash signatures, 128 perms x 4 B  = 512 B/doc
                                     = 7.7 TB of signatures alone
LSH band table, 16 bands             = 240 billion entries = 3.8 TB

The signature index is bigger than most systems' entire dataset, and it exists only to answer "have I seen something like this". That framing — the index for a side question is 4 TB — is what makes the scale concrete.

Compute, which is the surprise:

Compare to the training run it feeds: thousands of GPUs for months. The data pipeline is ~0.1% of the cost of the run and 100% of its correctness risk. Say that ratio — it justifies spending the design's effort on correctness and provenance rather than on throughput optimization, and it preempts the "how do you make it fast" line of questioning by showing you already know that is not the problem.

Shuffling.

60 TB / 100 GB shards = 600 shards, each fits in one host's RAM
a 1 GB streaming shuffle buffer holds 250M tokens = ~250k documents
                                     = 0.0017% of the corpus

A streaming shuffle buffer is not a shuffle. It reorders within a 0.0017% window. If the corpus is written source-by-source, the model sees hours of Wikipedia followed by hours of GitHub, and the loss curve will show it. Global shuffle must happen at build time, offline — deep dive B.

Resumption. 1,024 data-parallel ranks, each with its own read position, each streaming ~59 GB per epoch-shard. 1,024 positions to checkpoint atomically with the model state.

3. API Surface

# BUILD (offline, batch)
POST /snapshots            {sources[], filters{}, tokenizer_ref, seed}  -> {snapshot_id}
GET  /snapshots/{id}       -> {manifest_digest, token_count, stage_versions{}, stats{}}

# CONSUME (training time)
GET  /snapshots/{id}/manifest                  -> the immutable shard list + digests
loader = DataLoader(snapshot_id, mixture, seed, dp_rank, dp_world, seq_len)
state  = loader.state_dict()                   # goes INTO the model checkpoint
loader.load_state_dict(state)                  # exact resume, any dp_world

# TRACE
GET  /trace/token?snapshot=..&shard=..&offset=..
  -> {doc_id, source_url, crawl_date, filters_passed[], dedup_cluster, license}

Three decisions worth defending:

snapshot_id is a content digest of the manifest, not a name. v3-final-FIXED is how you get two incompatible corpora with the same label. A digest makes "is this the same data?" a string comparison, and it is the mechanism by which the original requirement is met.

state_dict() / load_state_dict() mirror the model's checkpoint API deliberately. Dataloader state must be saved in the same checkpoint, atomically with the model. If they are separate files, a crash between the two writes produces a run that resumes with the right weights and the wrong data position — and that produces silent data repetition that nobody will ever detect, because there is no error and the loss curve looks fine. Coupling the two APIs makes the atomic save the natural thing to do.

Mixture weights are a consumer parameter, not a snapshot parameter. One corpus build serves many mixture experiments. This is the decision from §1 and it saves a week of rebuild per experiment. The cost is that the mixture must be applied by sampling at read time, which constrains the shuffle design (deep dive B).

4. Data Model

document   (doc_id, source, url, crawl_date, raw_digest, text_digest,
            lang, quality_scores{}, filters_applied[], dedup_cluster_id, license_hint)
shard      (shard_id, snapshot_id, mixture_bucket, byte_offset_index,
            token_count, doc_ids[], digest)
snapshot   (snapshot_id, manifest_digest, created_at, stage_versions{}, token_count)
eval_ngrams(ngram_hash, eval_set)                       -- decontamination index
minhash    (doc_id, signature[128])                     -- 7.7 TB, transient
band_index (band_id, band_hash, doc_id)                 -- 3.8 TB, transient

doc_id is derived from content, not assigned. doc_id = H(text_digest, source). Two runs of the pipeline over the same input produce the same IDs, which is what makes the whole thing reproducible. An auto-increment ID would silently destroy reproducibility while looking perfectly reasonable in a schema review.

stage_versions is the field that answers the prompt's question. Every stage — filter, dedup, tokenizer — records its code version and config hash in the snapshot. When run N+1 is worse than run N, diffing two stage_versions maps answers "was it the data?" in seconds. Without it, the answer takes weeks and is usually "we think so".

minhash and band_index are marked transient and that is a real decision. 11.5 TB of intermediate state exists only during the build. Keeping it would let you incrementally dedup a new crawl against the old corpus — attractive, and it costs 11.5 TB of permanent storage plus the obligation to keep it consistent with a corpus that is supposed to be immutable. Rebuild instead; it is a few hundred core-hours. Recomputing is cheaper than remembering, which is worth saying because it is the opposite of the usual instinct.

mixture_bucket on the shard, not on the document. Shards are homogeneous by source category, so a mixture is "read shards from bucket A 70% of the time" — a sampling decision at read time with no data movement. That is what makes §3's consumer-side mixture cheap.

5. High-Level Architecture

  raw crawl (2 PB)          curated sources
        │                          │
  ┌─────▼──────────────────────────▼──────────┐
  │ 1. EXTRACT: WARC -> text, boilerplate strip │  content-addressed, idempotent
  └─────┬───────────────────────────────────────┘
  ┌─────▼───────────────────────────────────────┐
  │ 2. FILTER: lang ID · quality · safety · PII  │  each records its verdict
  └─────┬───────────────────────────────────────┘   (never deletes silently)
  ┌─────▼───────────────────────────────────────┐
  │ 3. EXACT DEDUP: text_digest, keep-first      │  cheap; ~30% of web
  └─────┬───────────────────────────────────────┘
  ┌─────▼───────────────────────────────────────┐
  │ 4. FUZZY DEDUP: MinHash -> LSH -> clusters   │  deep dive A
  └─────┬───────────────────────────────────────┘
  ┌─────▼───────────────────────────────────────┐
  │ 5. DECONTAMINATE: n-gram overlap vs evals    │  before tokenization
  └─────┬───────────────────────────────────────┘
  ┌─────▼───────────────────────────────────────┐
  │ 6. TOKENIZE + PACK into seq_len sequences    │  pinned tokenizer artifact
  └─────┬───────────────────────────────────────┘
  ┌─────▼───────────────────────────────────────┐
  │ 7. GLOBAL SHUFFLE (two-pass, offline)        │  deep dive B
  │    + write shards + manifest + digests        │
  └─────┬───────────────────────────────────────┘
        │  IMMUTABLE SNAPSHOT (60 TB, content-addressed)
  ┌─────▼───────────────────────────────────────┐
  │ DATALOADER: per-rank deterministic stream    │  deep dive B
  │  mixture sampling · resumable state_dict      │
  └──────────────────────────────────────────────┘

Five decisions:

1. Every stage is idempotent and content-addressed. Re-running a stage on the same input produces the same output, so a failure resumes at the stage boundary rather than the beginning. At 2 PB, a pipeline that must restart from scratch on failure is a pipeline that never completes. This is the property that makes a week-long build actually finish in a week.

2. Filters annotate; they do not delete. A document that fails quality gets filters_applied: ["quality:0.31<0.5"] and is excluded from the snapshot — but the record survives. Then "how much did we drop and why" is a query, not an archaeology project, and changing a threshold does not require re-extracting 2 PB.

3. Exact dedup before fuzzy. Exact is a hash join and removes ~30% of web text; fuzzy is 100× more expensive per document. Running the cheap filter first is obvious and easy to get backwards in a diagram.

4. Decontamination before tokenization. n-gram matching operates on text, and doing it after tokenization would tie the contamination index to a tokenizer version — so changing tokenizers would silently invalidate decontamination. Order in this pipeline encodes dependencies, and this is the one that bites.

5. The shuffle is offline and materialized, not done at read time. §2: a streaming buffer shuffles 0.0017% of the corpus. Deep dive B.

6. Deep Dive A: Global Deduplication at 15 Billion Documents

Why it matters more than it sounds

Duplicated training data is not merely wasted compute:

• Memorization. Documents seen many times are memorized verbatim, which is a privacy problem and a legal one.
• Measured quality loss. Deduplicated corpora train better models at equal token budget — this is one of the better-replicated results in the field.
• Benchmark contamination. Duplicates of an eval set that decontamination missed on one copy will be caught on another only if you dedup first.

And it must be global. Deduplicating within each crawl snapshot is easy and nearly useless — the same page appears in every monthly crawl. The whole difficulty is that the comparison is all-to-all across the entire corpus.

Exact dedup: easy, do it first

key = H(normalized_text)       # normalize: whitespace, unicode NFC, lowercase-for-hash-only
keep the earliest crawl_date per key; record the rest as cluster members

A distributed hash join over 15B rows. Hours on a modest cluster. Removes ~30% of web text.

One trap worth naming: normalization must be versioned and pinned, because changing it changes which documents are considered identical, which changes the corpus, which is exactly the "was it the data?" question. stage_versions["exact_dedup"] covers it — and noticing that a normalization tweak is a corpus change is the kind of thing that separates people who have operated one of these from people who have read about one.

Fuzzy dedup: MinHash + LSH

Why not all-pairs: (15e9)² / 2 = 1.1e20 comparisons. At a billion comparisons per second per core it is 3.5 million core-years. Not a tuning problem — a wrong-algorithm problem, and saying the number is how you demonstrate that.

MinHash estimates Jaccard similarity in constant space:

shingles(doc)  = set of 5-word n-grams
signature[i]   = min over shingles of h_i(shingle)     for i in 0..127
P(sig_A[i] == sig_B[i]) = Jaccard(A, B)                <- the whole theorem

128 permutations × 4 bytes = 512 B/doc, so any two documents' similarity is estimable from 1 KB regardless of their length. Standard error ≈ 1/sqrt(128) ≈ 8.8%.

LSH turns "compare everything" into "compare things that collide":

split the 128-value signature into b=16 bands of r=8 values
band_key = H(band_index, signature[8i : 8i+8])
two docs are CANDIDATES if they share any band_key

The probability two documents with Jaccard s become candidates is 1 - (1 - s^r)^b, which is an S-curve with its knee near:

\[ s^* \approx (1/b)^{1/r} = (1/16)^{1/8} \approx \mathbf{0.707} \]

So b and r are not tuning knobs — they are the similarity threshold. Being able to state that relationship, and to say which direction to move each to shift the threshold, is the difference between having used MinHash and having understood it.

Cost:

signatures:  15e9 x 512 B                  = 7.7 TB
band table:  15e9 x 16 bands x 16 B/entry  = 3.8 TB, 240e9 entries

Both are sequential-scan-and-sort workloads, not random-access ones — which is what makes them affordable: sort the band table by band_key, and colliding documents become adjacent runs. A distributed sort of 3.8 TB is routine.

Clustering, and the decision people skip

Candidate pairs form a graph. Connected components are duplicate clusters. Which member do you keep?

This is a real decision with real consequences, and "keep one arbitrarily" is a defect:

Choose: highest quality score, with the score's version pinned in stage_versions. Determinism comes from pinning the classifier, not from avoiding it — and that is the general resolution to "deterministic vs good": pin the input to the decision rather than degrading the decision.

And a false-positive check worth mentioning unprompted: at threshold 0.707, documents that merely share a long boilerplate header can collide. Sample the clusters, look at them, and measure the false-positive rate before trusting the pipeline. "I would look at a hundred of them" is a better answer than any threshold justification — this is a stage whose output nobody inspects and whose errors are invisible in aggregate statistics.

Transitivity, the failure that hides

Connected components are transitive; similarity is not. A—B similar, B—C similar, A—C entirely different — yet all three land in one cluster and two get dropped. Chains of these can collapse large, diverse sets into a single cluster.

Detect it: alarm on cluster size distribution. A cluster with 10 million members is a bug, not a duplicate set, and it is the signature of a boilerplate shingle turning into a hub node.

Bound it: cap cluster size, and for oversized clusters fall back to pairwise verification within the cluster. This is the failure that quietly deletes 5% of your corpus, and no aggregate metric shows it — the token count drops slightly and nobody investigates. Naming it is worth as much as the algorithm.

7. Deep Dive B: Deterministic Order and Exact Resumption

The requirement, stated precisely

Given (snapshot_id, seed, dp_world, dp_rank, step), produce exactly the documents that rank saw at that step — on any machine, at any time, after any failure, including after dp_world changes.

That last clause is the one that breaks naive designs, and it is not hypothetical: a cluster loses nodes, a run resumes at 896 ranks instead of 1,024, and the data order must still be correct.

Why streaming shuffle is not enough

From §2: a 1 GB shuffle buffer holds ~250k documents = 0.0017% of the corpus. If shards are written source-by-source, that buffer never spans two sources. The model sees the corpus sorted by source, which is close to the worst possible curriculum and shows up as oscillating loss.

The fix is a real shuffle at build time, in two passes:

PASS 1 (scatter):  for each document: write it to output shard H(doc_id, seed) % 600
                   -> each shard is a uniform random sample of the whole corpus
PASS 2 (in-shard): load a 100 GB shard into RAM, shuffle it, write it back
                   -> full randomness within a shard, and shards are already random samples

Two passes over 60 TB, both sequential. The result is a globally shuffled corpus in which reading any shard sequentially is statistically equivalent to sampling randomly. The read path is then trivially fast because the randomness is baked in — which is the point: move the randomness offline, so the hot path is sequential.

The order function

Order must be a pure function, not a stateful iterator:

def order(snapshot, seed, epoch):
    """A deterministic permutation of shard IDs. No state, no RNG object."""
    shards = snapshot.shard_ids                       # sorted; from the manifest
    return deterministic_shuffle(shards, key=(seed, epoch))

def rank_stream(snapshot, seed, epoch, dp_rank, dp_world):
    """Which shards this rank reads, in order."""
    perm = order(snapshot, seed, epoch)
    return perm[dp_rank::dp_world]                    # strided, not blocked

Strided ([rank::world]), not blocked ([rank*n : (rank+1)*n]), and this is deliberate. Under striding, changing dp_world from 1,024 to 896 redistributes which shards go to which rank but keeps every shard assigned to exactly one rank, with no re-derivation of shard boundaries. Under blocking, changing the world size shifts every boundary and the mapping is unrecoverable.

And seed and epoch are the only entropy. No random.shuffle() on a global RNG whose state depends on how many times it has been called — that is the standard way this becomes irreproducible and it is invisible until someone tries to reproduce a run.

Resumption state

{
  "snapshot_id":   "sha256:...",
  "seed":          1337,
  "epoch":         0,
  "global_step":   48213,
  "per_rank": [ {"shard_idx": 12, "doc_offset": 8842}, ... ],   # one per rank
  "dp_world_at_save": 1024,
}

Saved inside the model checkpoint, in the same atomic write (§3). Separate files mean a crash between them yields correct weights and a wrong data position — silent repetition, no error, no alarm. This is the single most common data-pipeline bug in real training runs and it is prevented by an API decision, not by a runtime check.

Resuming with a different world size

The hard case. 1,024 ranks saved; 896 available.

The wrong answer — "redistribute the remaining shards evenly" — silently re-reads data assigned to a rank that is gone and skips data another rank had already consumed. The mixture is now wrong in an unrecorded way.

The right answer: make consumption a property of the shard, not of the rank.

# Consumed shards are recorded in a global set, not implied by rank position.
consumed = set of (shard_id, fully_consumed | doc_offset)

def resume(consumed, seed, epoch, dp_world):
    remaining = [s for s in order(snapshot, seed, epoch) if s not in consumed_fully]
    my_shards = remaining[dp_rank::dp_world]
    # Partially-consumed shards carry their offset; a rank picking one up
    # starts where the previous owner stopped.

Because shard ordering is a pure function of (seed, epoch) and consumption is tracked per-shard, any world size can resume correctly. The work is redistributed; the data is not re-read or skipped.

Cost: the state is now O(shards) = 600 entries rather than O(ranks) = 1,024 — actually smaller, and it does not grow with the cluster. A rare case where the more correct design is also the cheaper one, which happens when the original design was tracking the wrong entity.

Mixture sampling, deterministically

Mixture weights are a consumer parameter (§3), so sampling happens at read time — and must still be reproducible:

def next_bucket(step, dp_rank, weights):
    # Hash-based, not RNG-state-based: any (step, rank) is computable directly,
    # so resumption needs no replay and no RNG state in the checkpoint.
    h = H(seed, epoch, dp_rank, step) / 2**64
    return weighted_choice(weights, h)

Direct computation from (step, rank), never a stateful RNG. Resuming at step 48,213 must not require replaying 48,213 draws — and more importantly, must not silently work by replaying them in a slightly different order.

The property to state: anything that must be reproducible after a resume should be a pure function of the position, not the accumulated state of a generator. That single rule prevents most reproducibility bugs in this class of system.

8. Failure and Recovery

"Refuse to resume" is the right behaviour on missing loader state, and it is worth defending because it will be argued with. The alternative — resume from step 0 of the data — silently repeats data the model has already seen, and every downstream metric is quietly wrong. A loud failure that costs an hour beats a silent one that costs a run.

The GPU-idle row is the one that actually costs money. A 1,024-GPU run at $2.50/GPU-hour is $2,560/hour; 2% idle is $1,200/day burned waiting for data. The dataloader must sustain ≥2× the consumption rate:

consumption = 1024 ranks x 4096 tokens x 4 B / step_time(~2 s) = 8.4 GB/s
target      = 2x = ~17 GB/s sustained from storage

17 GB/s is a real storage requirement and it belongs in this design, not in someone else's. It is met by sequential reads from many shards in parallel — which is the other reason the shuffle is materialized offline (§7): random reads at this rate would need a very different, much more expensive storage tier.

9. Bottlenecks and Evolution

Now: the bottleneck is the fuzzy-dedup shuffle-and-sort (3.8 TB band table) at build time, and storage read bandwidth at training time. Neither is CPU.

Interventions in order:

1. Incremental dedup against a published corpus. Keep the band index for the released snapshot (3.8 TB) so a new crawl deduplicates against it without a full rebuild. Reverses §4's "transient" decision — correctly, once rebuild frequency rises above roughly monthly. The cost is that the index must be versioned in lockstep with the corpus, which is exactly the kind of coupling that makes reproducibility harder. Worth it later, not now, and the trigger is measurable.
2. Quality classifier upgrades. The highest-leverage change to the final model, and the most dangerous to reproducibility: a new classifier is a new corpus. Requires a new snapshot ID and an A/B at small scale before adoption. Never patched into an existing snapshot.
3. Better decontamination. n-gram overlap misses paraphrases. Embedding-based detection catches more and has false positives that delete legitimate data. Measure both directions before switching, and keep the n-gram check as a floor.
4. Multi-epoch and repetition policy. At 15T tokens the interesting question becomes how many times to repeat high-quality data. Requires the sampler to support per-bucket repeat counts — cheap to add now, expensive to retrofit, so add the hook now even if the policy is "1".
5. Streaming ingestion for continuously updated corpora. Directly conflicts with §1's immutability decision. The resolution is frequent immutable snapshots, not mutable data — worth stating, because "make it streaming" is a natural-sounding suggestion that would destroy the design's foundational property.

10. Tradeoffs Explicitly Rejected

Rejected: all-pairs deduplication. 1.1e20 comparisons = 3.5M core-years. Wrong algorithm, not slow implementation.

Rejected: per-snapshot (local) deduplication only. Cheap and nearly useless — the same pages recur in every crawl, so local dedup removes almost none of the actual duplication.

Rejected: streaming shuffle buffers as the only shuffle. §2: 0.0017% of the corpus. Produces a source-ordered curriculum.

Rejected: mutable "latest" corpus. Destroys reproducibility, which is the stated requirement. Immutable snapshots with digests.

Rejected: storing dataloader state separately from the model checkpoint. A crash between the two writes yields silent data repetition. Atomic, single checkpoint.

Rejected: blocked shard assignment ([rank*n:(rank+1)*n]). Breaks on any world-size change. Strided.

Rejected: assigning doc_id by auto-increment. Non-reproducible across builds while appearing entirely normal. Content-derived IDs.

Rejected: deleting filtered documents. Keeping the annotation makes threshold changes a query instead of a 2 PB re-extraction.

Rejected: tokenizing before decontamination. Ties the contamination index to a tokenizer version, so a tokenizer change silently invalidates decontamination.

Rejected: uint16 tokens with a 128k vocab. Does not fit. Mentioned only because the reverse — a ≤65k vocab — halves the corpus, and that is a conversation worth having with the modelling team rather than absorbing silently.

The Hostile Critique

C1. "Two-pass shuffle: pass 1 scatters every document to a random shard. That's 15 billion random writes across 600 destinations. You describe it as sequential. Walk me through what actually happens at the storage layer, and tell me how long pass 1 takes."

C2. "You keep the highest-quality document in each dedup cluster, with the classifier pinned. The classifier scores documents, but the cluster is defined by MinHash. So a cluster contains a high-quality Wikipedia article and a scraped SEO copy of it with a spam footer. Which scores higher on a fastText quality classifier trained to prefer Wikipedia-like text, and are you sure?"

C3. "consumed is 'a global set' of shards. Global to what? You have 1,024 ranks writing to it. If it's in the checkpoint, only rank 0 writes it and it's stale for everyone else. If it's a service, it's on the training hot path. Which is it?"

C4. "Decontamination runs before tokenization against 'a registry of eval sets'. New benchmarks are published after your snapshot is built. Your model gets evaluated on a benchmark that didn't exist when you built the corpus. What do you do — and what do you tell people about your reported numbers?"

C5. "17 GB/s sustained read, and the shuffle is materialized so reads are sequential. But your mixture sampler picks a bucket per step by hash — so consecutive steps read from different buckets, in different shards, at random offsets. Where did the sequential access go?"

C6. "Yield is 3%, so 2 PB in gives 60 TB out. You process 2 PB through extract, filter, dedup and only then discard 97% of it. What does that cost, and would you order the stages differently if you did the arithmetic?"

The Revision

R1 — Pass 1 must be a sort, not a scatter (answers C1)

The critique is correct and the original description was wrong about the physics. "Write each document to a random shard" is 15 billion small appends to 600 destinations. Even with per-shard write buffers, this is a shuffle in the MapReduce sense — the expensive part of any distributed sort, and describing it as "sequential" was hand-waving.

Change: state it as what it is, and size it.

PASS 1 = distributed sort by shuffle_key = H(doc_id, seed)
  map:     read 60 TB sequentially, compute key, write to N local spill files
  shuffle: exchange spills over the network      <- 60 TB across the fabric
  reduce:  each reducer receives ~100 GB, sorts in RAM, writes one shard

The cost is one full network shuffle of 60 TB. On a 100 Gb/s-per-node fabric with 100 nodes, aggregate ~1.25 TB/s, so ~48 seconds of pure transfer — in practice tens of minutes with spill I/O and skew. That is entirely affordable, but it is a different cost than described and it needs a cluster that can do a 60 TB sort, which is a real infrastructure requirement.

And the optimization the correction reveals: with buffered writes at 64 MB per destination, memory is 600 destinations × 64 MB = 38 GB per writer — feasible, and it converts 15 billion small writes into ~1 million large ones. Which is the standard answer, and it is only visible once you stop calling it "a scatter" and start calling it "a sort".

Cost: a real dependency on a distributed sort framework. Accepted — this is Spark/Ray's core competency and building it by hand would be the mistake.

R2 — Cluster representative selection must be source-aware, not score-aware (answers C2)

The critique identifies a genuine and embarrassing failure mode, and the answer to "are you sure?" is no. A quality classifier trained to score Wikipedia-like text highly will happily score an SEO scrape of a Wikipedia article highly too — it is, after all, Wikipedia text. The spam footer is a small fraction of the document and may not move the score below the winner.

So the pipeline can systematically prefer scraped copies over originals, which is worse than arbitrary selection because it is biased rather than random.

Change: representative selection becomes lexicographic over multiple signals, with source authority first.

def pick_representative(cluster):
    return min(cluster, key=lambda d: (
        SOURCE_RANK[d.source],        # curated < known-good domain < general web
        -d.quality_score,             # then quality
        d.crawl_date,                 # then earliest seen
        d.doc_id,                     # then deterministic tie-break
    ))

Source authority dominates because it is the signal the classifier cannot see. A curated source's copy wins over any web copy regardless of score, which is exactly the ordering the critique's example requires.

And a detection mechanism, because the fix should be verifiable: measure how often the chosen representative differs in length from the cluster's median by more than 20%. A representative systematically longer than its cluster is a footer/boilerplate signal. Sample and read them — this is the stage from §6 whose errors are invisible in aggregate, and the critique is a concrete instance of exactly that.

Cost: SOURCE_RANK is a hand-maintained ordering, which is a curation burden and a place for bias to enter deliberately rather than accidentally. That is an improvement — an explicit, reviewable table beats an implicit preference learned by a classifier nobody inspects.

R3 — Consumption state is per-rank in the checkpoint, reconciled at load (answers C3)

The critique is right that "a global set" was undefined, and both readings it offers are bad: a service on the hot path adds a network dependency to every step, and rank-0-only state is stale.

Change: each rank keeps its own consumption record; the union is formed only at checkpoint save, which is already a synchronization barrier.

# During training: each rank tracks only its own shards. No coordination at all.
local = {"shards_done": [...], "current": ("shard_412", offset=8842)}

# At checkpoint (an existing all-reduce barrier -- no new synchronization):
all_local = all_gather(local)                # 1024 x ~600 B = 600 KB. Trivial.
checkpoint["dataloader"] = merge(all_local)  # written atomically with the model

# At load, at ANY world size:
consumed = checkpoint["dataloader"]["shards_done"]      # the global set, materialized once
partial  = checkpoint["dataloader"]["partials"]         # shard -> offset

The global set exists exactly at checkpoint time and nowhere else. No service, no hot-path coordination, no staleness — because the only moment the union is needed is the only moment all ranks are already synchronized.

And it rides an existing barrier, so the added cost is 600 KB in an all-gather that already happens. When a design needs global state, look for a barrier that already exists before inventing a service — that is the transferable form of this fix.

Cost: if a rank dies between checkpoints, its in-flight partial progress since the last checkpoint is lost and its shard is re-read from the last recorded offset. Bounded by the checkpoint interval, and re-reading a few thousand documents is statistically irrelevant at 15T tokens — as long as it is recorded, which it is.

R4 — Decontamination is a post-hoc measurement as well as a pre-hoc filter (answers C4)

The critique names a problem that cannot be solved at build time, and the honest response is to say so rather than to pretend the filter is sufficient.

Change: two mechanisms, not one.

(a) Pre-hoc filter — decontaminate against every eval set known at build time. Unchanged.

(b) Post-hoc contamination report — retain the corpus's n-gram index (not the corpus, just the index) so that any future benchmark can be checked against the corpus after the fact.

13-gram index over the final corpus: ~15e12 tokens -> sampled at 1/10 -> ~1.5e12 entries
Bloom-filtered to ~2 TB, retained with the snapshot forever.

Later: new benchmark published
  -> query the index
  -> publish contamination rate ALONGSIDE the benchmark score

What to tell people, which is the real question the critique asks: publish the contamination rate with the score. "We score 82.4 on BenchmarkX; 0.3% of its items have a 13-gram overlap with our training corpus; excluding those, 82.1." That is the answer that survives scrutiny, and it is only possible because the index was retained. Retaining a 2 TB index is cheap insurance against a public credibility problem.

And the limit, stated plainly: n-gram overlap catches copied text and misses paraphrase and translation. There is no complete solution, and a design claiming decontamination is "handled" is overclaiming. The right posture is measurement and disclosure, not a filter that is asserted to be sufficient.

Cost: 2 TB permanent per snapshot, and the discipline to run the check whenever a benchmark is reported. The second is organizational and is the part that actually fails.

R5 — Sampling must be shard-aligned in runs, not per-step (answers C5)

The critique catches a direct contradiction between §7's sampler and §8's storage requirement, and it is right: per-step bucket sampling means consecutive steps land in different shards, so the "sequential reads" claim is false and the 17 GB/s requirement would need random-access storage.

Change: sample a bucket per run of steps, not per step, and prefetch whole shards.

RUN = 512                     # steps per bucket switch
def bucket_for(step, dp_rank):
    h = H(seed, epoch, dp_rank, step // RUN) / 2**64
    return weighted_choice(weights, h)

Each rank reads a bucket's shard sequentially for 512 steps (~17 minutes at 2 s/step), then switches. Reads are sequential within a run; the mixture is correct in expectation over many runs.

Is the mixture still right? Yes, and it is worth showing rather than asserting: bucket choice is i.i.d. across runs, so over an epoch of ~24,000 runs per rank the empirical mixture converges to the weights with standard error sqrt(p(1-p)/24000) — under 0.3% for a 70% bucket. Negligible, and now demonstrated rather than hoped for.

And the correlation caveat: ranks must not switch buckets in lockstep, or the whole cluster hammers one bucket's shards simultaneously. Including dp_rank in the hash (as above) decorrelates them — without it this fix would create a thundering herd on storage, trading one problem for a worse one.

Cost: within a 512-step run the batch is less mixture-diverse. At a global batch of 1,024 sequences with each rank independently choosing, every batch still contains many buckets — the diversity lives across ranks rather than across steps, which is sufficient.

R6 — Filter order must be cost-ordered, and the arithmetic changes it (answers C6)

The critique is right that the arithmetic was never done, and doing it reorders the pipeline.

The cost of the current order:

extract 2 PB    -> 1 PB text        (cheap: I/O bound)
filter 1 PB     -> 200 TB           (fastText: ~2 MB/s/core = 139,000 core-hours)
exact dedup     -> 140 TB           (hash join)
FUZZY DEDUP 140 TB                  <- MinHash on 140 TB of which 60 TB survives
decontaminate, tokenize 60 TB

Fuzzy dedup — the most expensive stage — runs on 2.3× more data than it needs to, because quality filtering has already removed most of the junk but dedup runs on everything that passed.

Change: cheapest-and-most-selective first, always.

1. extract                        2 PB -> 1 PB
2. EXACT DEDUP (hash only)        1 PB -> 700 TB   <- moved UP: pure hashing, ~free,
                                                       removes 30% before any classifier runs
3. cheap filters: length, lang ID, charset          <- ~0.1x the cost of quality scoring
                                  700 TB -> 300 TB
4. quality + safety classifiers   300 TB -> 100 TB  <- now runs on 3.3x less data
5. FUZZY DEDUP                    100 TB -> 65 TB   <- 1.4x less than before
6. decontaminate, tokenize        65 TB  -> 60 TB

Savings: the quality classifier runs on 300 TB instead of 1 PB (~97,000 core-hours saved), and fuzzy dedup on 100 TB instead of 140 TB. The pipeline gets roughly 3× cheaper from reordering alone, with no algorithmic change.

The rule, which generalizes past this design: order filters by selectivity / cost descending. Exact dedup is nearly free and removes 30% — it belongs first, and it was fourth. Language ID is cheap and very selective — before quality scoring, not after.

The one ordering constraint that overrides cost: decontamination must precede tokenization (§5) because it operates on text. Everything else is free to reorder by cost, and checking whether a pipeline's order is a dependency order or merely a habitual one is worth doing every time.

References

• ../README.md#d6-surrounding-systems — where data pipelines sit in the Track D concept inventory
• m05-eval-harness.md — the contamination registry this consumes, and R4's post-hoc check
• m08-training-fault-tolerance.md — checkpointing the dataloader state atomically with the model
• ../../systems-design/designs/d07-log-analytics.md — immutable segments, sort-based pipelines, index-vs-query cost
• ../../coding/WARMUP.md#chapter-9-deduplication-and-probabilistic-structures — MinHash, Bloom filters and dedup as timed coding problems
• Broder, A. On the Resemblance and Containment of Documents. 1997 — MinHash
• Leskovec, Rajaraman, Ullman. Mining of Massive Datasets, ch. 3 — LSH banding and the (1/b)^(1/r) threshold
• Lee, K. et al. Deduplicating Training Data Makes Language Models Better. ACL 2022 — the measured quality result
• Penedo, G. et al. The RefinedWeb Dataset / FineWeb. — filtering yields and the ordering of stages in practice
• Soldaini, L. et al. Dolma: an Open Corpus of Three Trillion Tokens. ACL 2024 — a fully documented pipeline of exactly this shape
• Dodge, J. et al. Documenting Large Webtext Corpora. EMNLP 2021 — contamination measurement and disclosure