60. Data pipelines for training: extraction, cleaning, dedup, JSONL

Most of the work in a real AI project is data, and most data work is Python. A typical offline pipeline turns raw documents (PDFs, EPUBs, HTML, logs, tickets) into clean training or retrieval records:

 raw files ──▶ extract text ──▶ clean/normalize ──▶ filter ──▶ dedup ──▶ chunk ──▶ JSONL / Parquet
 (pdf, epub,    (pypdf,          (unicode NFC,       (length,    (exact hash,  (by tokens,   one record per line,
  html, md)      ebooklib, bs4)   whitespace, PII)    language)   near-dup)     with overlap) with provenance

# labs/corpus.py: build a deduplicated, chunked JSONL corpus from text and markdown files
import hashlib
import json
import re
import sys
import unicodedata
from collections.abc import Iterable, Iterator
from pathlib import Path

EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+")
WS = re.compile(r"[ \t]+")

def extract(path: Path) -> str:
    if path.suffix in {".txt", ".md"}:
        return path.read_text(encoding="utf-8", errors="replace")
    if path.suffix == ".pdf":
        from pypdf import PdfReader                    # uv add pypdf  (imported only when needed)
        return "\n".join(page.extract_text() or "" for page in PdfReader(path).pages)
    raise ValueError(f"unsupported: {path.suffix}")

def clean(text: str) -> str:
    text = unicodedata.normalize("NFC", text)          # one encoding for "é" (Security Ch 7)
    text = EMAIL.sub("<email>", text)                  # scrub obvious PII before it reaches a model
    text = "\n".join(WS.sub(" ", line).strip() for line in text.splitlines())
    return re.sub(r"\n{3,}", "\n\n", text).strip()

def chunks(text: str, max_words: int = 200, overlap: int = 40) -> Iterator[str]:
    words = text.split()
    step = max_words - overlap
    for start in range(0, max(len(words) - overlap, 1), step):
        yield " ".join(words[start : start + max_words])

def build(paths: Iterable[Path], out: Path, min_words: int = 20) -> dict[str, int]:
    seen: set[str] = set()
    stats = {"files": 0, "chunks": 0, "dupes": 0, "short": 0}
    with out.open("w", encoding="utf-8") as f:
        for path in paths:
            stats["files"] += 1
            for i, chunk in enumerate(chunks(clean(extract(path)))):
                if len(chunk.split()) < min_words:
                    stats["short"] += 1
                    continue
                key = hashlib.sha256(chunk.lower().encode()).hexdigest()
                if key in seen:
                    stats["dupes"] += 1
                    continue
                seen.add(key)
                record = {"id": key[:16], "source": str(path), "chunk": i, "text": chunk}
                f.write(json.dumps(record, ensure_ascii=False) + "\n")
                stats["chunks"] += 1
    return stats

if __name__ == "__main__":
    root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(".")
    files = sorted(p for p in root.rglob("*") if p.suffix in {".md", ".txt", ".pdf"})
    print(build(files, Path("corpus.jsonl")))

What to notice:

• JSONL (one JSON object per line) is the lingua franca of LLM data: streamable (Ch 13), appendable, splittable, and readable with head. Use Parquet when the data is large and columnar.
• Provenance in every record (source, chunk). When a model or a RAG answer says something wrong, you need to find where it came from.
• Deduplicate before splitting train/test, or duplicates leak across the split (Ch 56.3). Exact hashing catches copies; near-duplicates need MinHash/LSH (datasketch).
• Scrub PII early and keep raw data out of the training set's directory. Be deliberate about which directories a pipeline reads: point it at an explicit list of roots, never at a home directory, because personal documents (payslips, IDs) end up in corpora this way.
• Chunk by tokens for real models (count with the model's tokenizer, Ch 59); words are an approximation.

Run each stage as a separate, restartable step that writes its output atomically (Ch 35), so a crash at stage 4 does not redo stages 1–3.