Distributed Dataset Construction

"They spent nine months building the cluster and one afternoon picking the data. The model remembers which decision they got wrong."

A Quality Classifier That Was Outvoted

A foundation model is, in a precise sense, a compression of its training corpus, so the corpus is the most important input the project controls. Yet the raw material is the open web: trillions of pages of HTML, most of it boilerplate, spam, navigation chrome, machine translation, and duplicate text, in hundreds of languages and dozens of encodings. Turning that into a corpus a model should learn from is not a preprocessing afterthought; it is a distributed computation on the same scale as the training run itself, and frequently on the same cluster. A modern crawl such as Common Crawl is on the order of hundreds of terabytes of compressed WARC files per monthly snapshot, and a competitive pretraining set draws on many snapshots. No single machine reads that, so the construction job is partitioned across hundreds or thousands of workers from its first byte.

This section walks the construction pipeline end to end: why each stage is a distributed job, how language and quality filtering work and why quality filtering dominates the others, and what the output format must look like so that Section 8.6's streaming data loaders can feed thousands of training workers without stalling. Deduplication, the single most impactful filter of all, is large enough to get its own treatment in Section 19.4; here we build the funnel around it.

1. Why Building the Corpus Is a Petabyte MapReduce Job Beginner

Every stage of corpus construction has the shape that Chapter 6 formalized: an embarrassingly parallel map over independent records, occasionally followed by a shuffle that groups records by a key. Extraction maps over individual web pages, each parsed without reference to any other, which is a pure map with no reduce. Language identification and quality scoring are likewise per-document maps. Deduplication is the one stage that genuinely needs a shuffle, because deciding whether two documents are near-duplicates requires bringing candidate pairs together by a shared key (a hash or a MinHash band), which is exactly the group-by-key that the MapReduce shuffle and Chapter 7's Spark joins provide.

The scale is what forces distribution. A single monthly Common Crawl snapshot is on the order of $90\,\text{TB}$ compressed and several times that uncompressed; a pretraining corpus draws on many snapshots plus curated sources, so the input routinely runs to a petabyte or more. Reading a petabyte at a generous $1\,\text{GB/s}$ per worker takes one worker about $10^6$ seconds, roughly eleven and a half days, before any actual processing. Spread across $W$ workers the wall-clock divides by $W$, so a thousand-worker job reads it in about a quarter of an hour of raw I/O. The map work is perfectly parallel, the storage is shared object storage (Section 8.6), and the only synchronization is the dedup shuffle. This is the canonical case for Spark or a MapReduce-style framework, and it is the moment in the book where the data-processing machinery of Part II is applied at its full intended scale.

2. Crawl and Extract: Text Out of Markup Beginner

The pipeline begins with ingestion. Most teams do not crawl the web themselves; they start from Common Crawl's WARC (Web ARChive) files, which package raw HTTP responses, and read them in parallel from object storage. Each worker streams a slice of the WARC shards, and for each captured page it must turn HTML into the visible text a human would read. That extraction is harder than deleting tags. Boilerplate (navigation bars, cookie banners, footers, ad templates) repeats across millions of pages and carries no learnable signal, so good extractors estimate the main content region and discard the chrome. Production pipelines use trafilatura or resiliparse for this, tuned to keep article text while dropping menus.

Extraction is also where encoding and structure are normalized: character sets are unified to UTF-8, HTML entities are resolved, and the document is reduced to plain text plus a little metadata (source URL, language guess, timestamp). Because every page is processed independently, extraction is the purest map in the pipeline and scales linearly with worker count. It is also where the first large volume drop happens: pages that are pure markup, that fail to parse, or that yield only a few words of body text are dropped immediately, so the hundreds of terabytes of extracted text are already much smaller than the petabytes of raw HTML they came from.

3. Filtering: Language, Heuristics, and a Model That Judges Data Intermediate

Filtering decides what the model learns from, and it works in layers of increasing sophistication. The first layer is language identification: a fast classifier (commonly a fastText model) labels each document, and the pipeline keeps only the target languages. The second layer is heuristic quality filtering, a set of cheap rules drawn from the Gopher and C4 recipes: drop documents that are too short, that have an unusual ratio of symbols to words, that consist mostly of lists or bullet fragments, that repeat lines or n-grams excessively, or that lack the stopwords a fluent passage would contain. These rules are crude individually but remove a large mass of obvious junk at almost no cost.

The third and most consequential layer is classifier-based quality scoring: a model judges the data. A lightweight classifier, often a linear model or small neural net over n-gram or embedding features, is trained to distinguish a reference set of high-quality text (for example, pages linked from curated sources) from random web text, then scores every candidate document. Documents below a threshold are dropped. Formally, if a learned scorer $q(d) \in [0,1]$ estimates the probability that document $d$ resembles the high-quality reference distribution, the pipeline keeps the set $$\mathcal{D}_{\text{keep}} = \{\, d : \text{lang}(d) \in \mathcal{L} \;\wedge\; h(d) \;\wedge\; q(d) \ge \tau \,\},$$ where $\mathcal{L}$ is the target language set, $h(d)$ is the conjunction of heuristic rules, and $\tau$ is a tunable threshold trading corpus size against average quality. Choosing $\tau$ is a real decision: too high and you discard usable tokens the scaling laws of Section 19.2 say you need, too low and low-grade text dilutes the signal.

The 2024 FineWeb and DataComp-for-Language-Models (DCLM) results made the stakes of this layer unambiguous. Both showed that a model-based quality filter, applied at scale, produces a corpus on which a fixed-compute model trains to materially higher benchmark accuracy than the same model trained on a larger but less-filtered corpus. The filtering stage, in other words, moved more benchmark points than most architectural tweaks. That is the empirical heart of the data-centric view, and we state it as the section's thesis.

4. A Distributed-Style Pipeline End to End Intermediate

The code below runs the construction pipeline in miniature on a synthetic crawl, using only the Python standard library so it executes anywhere. It builds twenty thousand HTML-ish "pages" containing a realistic mix (genuine English prose, non-English text, spam boilerplate, and near-empty stubs), then runs the three real stages: a parallel extract that strips markup, a filter combining language identification with a heuristic-plus-classifier quality gate, and a shard step that routes survivors to balanced output shards by a stable content hash. The map phase runs across eight worker processes with ProcessPoolExecutor, the same partition-and-map structure a Spark job uses at petabyte scale. The point is to watch how much each stage removes and what token yield remains, the numbers a real pipeline reports to its operators.

import re
import random
import hashlib
from concurrent.futures import ProcessPoolExecutor
from collections import Counter

# ---- 1. Build a synthetic web crawl: a list of raw HTML-ish records. ----
ENGLISH = ("the quick brown fox jumps over lazy dogs while data flows through "
           "the cluster and workers reduce gradients across the network").split()
SPANISH = "el rapido zorro marron salta sobre perros perezosos mientras datos".split()
BOILER = "click here subscribe now buy cheap pills free download winner".split()

def make_doc(i):
    r = random.random()
    if r < 0.18:                     # boilerplate / spam page
        body = " ".join(random.choices(BOILER, k=random.randint(4, 30)))
    elif r < 0.34:                   # non-English page
        body = " ".join(random.choices(SPANISH, k=random.randint(20, 120)))
    elif r < 0.42:                   # near-empty page
        body = " ".join(random.choices(ENGLISH, k=random.randint(1, 6)))
    else:                            # genuine English prose
        body = " ".join(random.choices(ENGLISH, k=random.randint(60, 400)))
    nav = "<nav>home about contact</nav>"
    return f"<html><head><title>doc {i}</title></head><body>{nav}<p>{body}</p></body></html>"

# ---- 2. EXTRACT: strip tags, keep visible text. Runs in parallel workers. ----
TAG = re.compile(r"<[^>]+>")
def extract(doc):
    text = TAG.sub(" ", doc)               # drop all HTML tags
    text = re.sub(r"\s+", " ", text).strip()
    return text

# ---- 3. FILTER: language id + heuristic quality + a tiny quality classifier. ----
EN_VOCAB = set(ENGLISH) | {"home", "about", "contact", "doc"}
def english_fraction(tokens):
    if not tokens:
        return 0.0
    return sum(t in EN_VOCAB for t in tokens) / len(tokens)

def quality_score(tokens):
    """Cheap stand-in for a learned quality classifier: rewards length and
    lexical diversity, penalizes spam-like repetition."""
    if not tokens:
        return 0.0
    diversity = len(set(tokens)) / len(tokens)
    spam = sum(t in set(BOILER) for t in tokens) / len(tokens)
    length_ok = min(len(tokens), 50) / 50.0
    return 0.5 * diversity + 0.5 * length_ok - spam

def keep(text):
    toks = text.split()
    if len(toks) < 10:                     # heuristic: too short
        return False, toks
    if english_fraction(toks) < 0.5:       # language filter
        return False, toks
    if quality_score(toks) < 0.35:         # classifier-based quality gate
        return False, toks
    return True, toks

def process_chunk(chunk):
    """One 'worker': extract then filter its shard, return surviving token lists."""
    kept, n_extracted = [], 0
    for doc in chunk:
        text = extract(doc)
        n_extracted += 1
        ok, toks = keep(text)
        if ok:
            kept.append(toks)
    return n_extracted, kept

def pct(x, base):
    return f"{100.0 * x / base:5.1f}%"

if __name__ == "__main__":
    random.seed(0)
    N_DOCS = 20_000
    crawl = [make_doc(i) for i in range(N_DOCS)]
    raw_bytes = sum(len(d) for d in crawl)

    # Split the crawl across K workers, MapReduce-style.
    K = 8
    chunks = [crawl[i::K] for i in range(K)]
    extracted_total, survivors = 0, []
    with ProcessPoolExecutor(max_workers=K) as ex:
        for n_extracted, kept in ex.map(process_chunk, chunks):
            extracted_total += n_extracted
            survivors.extend(kept)

    # ---- 4. SHARD: route survivors to balanced shards by a stable hash. ----
    N_SHARDS = 4
    shard_counts, shard_tokens = Counter(), Counter()
    for toks in survivors:
        h = hashlib.blake2b(" ".join(toks).encode(), digest_size=8).digest()
        sid = int.from_bytes(h, "big") % N_SHARDS   # stable content-hash routing
        shard_counts[sid] += 1
        shard_tokens[sid] += len(toks)

    total_tokens, kept_docs = sum(shard_tokens.values()), len(survivors)
    print("STAGE                         docs       retained   tokens")
    print(f"crawl (raw HTML)          {N_DOCS:8d}    100.0%   ({raw_bytes/1e6:.1f} MB raw)")
    print(f"extract (text from HTML)  {extracted_total:8d}  {pct(extracted_total, N_DOCS)}")
    print(f"filter (lang+qual)        {kept_docs:8d}  {pct(kept_docs, N_DOCS)}   {total_tokens:8d}")
    print(f"\nfinal corpus: {kept_docs} docs, {total_tokens} tokens across {N_SHARDS} shards")
    for sid in range(N_SHARDS):
        print(f"  shard {sid}: {shard_counts[sid]:5d} docs  {shard_tokens[sid]:8d} tokens")
    print(f"\nyield: {pct(kept_docs, N_DOCS)} of crawled docs survived to training")

STAGE                         docs       retained   tokens
crawl (raw HTML)             20000    100.0%   (19.7 MB raw)
extract (text from HTML)     20000  100.0%
filter (lang+qual)           12079   60.4%    2705564

final corpus: 12079 docs, 2705564 tokens across 4 shards
  shard 0:  2892 docs    650410 tokens
  shard 1:  3110 docs    698546 tokens
  shard 2:  3047 docs    678772 tokens
  shard 3:  3030 docs    677836 tokens

yield:  60.4% of crawled docs survived to training

Two properties of Output 19.3.1 carry over directly to real pipelines. First, the per-shard balance: routing by a content hash gives shards of nearly equal size without any coordination between workers, which matters because the training loaders of Section 8.6 assign shards to workers and any imbalance becomes a straggler. Second, content-hash routing is stable and idempotent: rerun the pipeline and a document lands in the same shard, so incremental crawls append cleanly and reproducibility is free. The synthetic yield of sixty percent is generous by design; the structural lesson, not the percentage, is the takeaway.

5. The Output Format: Sharded, Shuffled, Streaming-Friendly Intermediate

The funnel's output is not one giant file; it is many shards, each a sequence of documents in a binary or compressed-line format, sized so that one shard streams comfortably from object storage while a worker trains on the previous one. Three properties matter. The data must be sharded so that thousands of training workers each read a disjoint slice in parallel, exactly the partitioning arc that runs from Section 8.6 through the training loop. It must be shuffled at the document level so that consecutive training examples are not all from one website or one crawl date, which would bias the gradient; global shuffling at petabyte scale is itself a distributed shuffle, done once at construction time rather than per-epoch. And it must be deduplicated, the subject of Section 19.4, because duplicate documents waste compute and, worse, encourage memorization.

The layout is deliberately streaming-friendly: shards are written so a loader can fetch the next one over the network while the current one feeds the GPUs, hiding I/O latency behind computation in the same way the collective overlap of Chapter 16 hides communication behind the backward pass. Whether the corpus is finally tokenized at construction time or streamed as text and tokenized on the fly (Section 19.5), the shard boundaries and the shuffle are fixed here, because re-shuffling a multi-trillion-token corpus mid-training is not something anyone wants to do twice.

# pip install datatrove[all]
from datatrove.pipeline.readers import WarcReader
from datatrove.pipeline.extractors import Trafilatura
from datatrove.pipeline.filters import (
    LanguageFilter, GopherQualityFilter, GopherRepetitionFilter)
from datatrove.pipeline.writers import JsonlWriter
from datatrove.executor import LocalPipelineExecutor

pipeline = [
    WarcReader("s3://commoncrawl/.../segment/"),  # ingest raw crawl
    Trafilatura(),                                 # HTML -> main text
    LanguageFilter(languages=["en"]),              # keep target languages
    GopherRepetitionFilter(),                      # drop repetitive junk
    GopherQualityFilter(),                         # heuristic quality gate
    JsonlWriter("output/shards/"),                 # sharded, streaming-ready
]
LocalPipelineExecutor(pipeline=pipeline, tasks=64).run()  # 64 parallel workers

6. The Data-Centric View: Better Data Beats More Data Advanced

The scaling laws of Section 19.2 are stated for a fixed data distribution: given that distribution, more compute and more tokens predict lower loss. What they do not say is that all tokens are equal, and they are not. Two corpora of identical token count can produce models that differ by many benchmark points, because the loss the model minimizes is the loss on its data, and a model trained on cleaner data minimizes a more useful objective. The data-centric reframing of the last few years says: at a fixed compute budget, the highest-return move is frequently to improve the data distribution, not to add parameters or steps. Filtering, deduplication, and careful source mixing change the distribution the scaling law operates over, effectively shifting the whole loss curve down.

This is why the construction pipeline of this section deserves the same engineering rigor as the training loop. Every decision in the funnel, the extractor's boilerplate cutoff, the language threshold, the quality classifier's reference set, the dedup aggressiveness of Section 19.4, the mixing weights across web, code, and books, reshapes the distribution the model compresses. The team that treats these as a distributed-data engineering problem, measured and ablated like any other system parameter, builds better models at the same cost than the team that treats data as a fixed input and pours all its effort into the GPUs.

We now have the corpus as a funnel: ingested, extracted, filtered by language and a model that judges quality, and emitted as balanced, shuffled shards ready for the loaders. One stage was deferred because it is large and subtle enough to stand alone, deduplication, which removes the near-duplicate documents that waste compute and drive memorization, and which is the single most impactful filter of all. That is the subject of Section 19.4.