Dataset Pipelines and Data Quality

Tickets 101 and 102 look like two separate training rows in an incident export. They differ only in leading spaces and letter case, yet both belong to conversation c-51. If 101 trains the router while 102 evaluates it, the held-out score includes a message the model has already seen. That can reward memorization rather than tell us how the router handles a new incident.

The previous chapter trained a ticket router from pre-computed numerical tensors. Now trace those tensors backward: which raw records can become supervision, and which held-out records can an evaluation engineer actually trust? A dataset pipeline turns raw records into model inputs and records how it did so. That record supplies provenance: where each example came from, which transformations changed it, and which gates approved it. Cleaning, splitting, and contamination checks protect different parts of the same training claim.

Ingestion contracts: schema enforcement and type validation

The router consumes features derived from support tickets. Start one step earlier with exported messages and labels from a developer-support conversation dump:

Read these four exported rows as four distinct pipeline decisions:

Ticket 104 presents the first breakdown: a message exists, but no reviewed label does. What should happen before normalization or splitting?

Start with an ingestion contract gate. Three useful kinds of checks ask different questions:

1. Structural gates: Enforce column presence, non-nullity, and strict data types. A float ticket ID or a string where an integer belongs gets rejected immediately.
2. Domain gates: Enforce value bounds, regex patterns, and categorical membership. Here, labels must belong strictly to {"standard", "escalate"}.
3. Distributional gates: Flag batch-level anomalies, such as unexpected null rates or shifts in label ratios. These can reveal a changed source or collection process even when individual rows are valid.

Pydantic's strict mode can reject a string "101" where an integer is required; its exact rules depend on the target type and whether input arrives as Python objects or JSON. Extra-field handling is a separate setting, not a consequence of strict typing.^{[1]Reference 1Strict Modehttps://pydantic.dev/docs/validation/latest/concepts/strict_mode/[2]Reference 2Modelshttps://pydantic.dev/docs/validation/latest/concepts/models/} Pandera offers DataFrame schemas, column checks, and optional hypothesis tests.^{[3]Reference 3DataFrame Schemashttps://pandera.readthedocs.io/en/stable/dataframe_schemas.html} Great Expectations records validation results and supports actions based on them. You must connect a failing result to a stop or quarantine policy; installing a validator doesn't itself prevent training.^{[4]Reference 4Create a Checkpoint with Actionshttps://docs.greatexpectations.io/docs/core/trigger_actions_based_on_results/create_a_checkpoint_with_actions/}

Decide how the contract handles schema changes. An extra field might be harmless to a consumer that ignores it, but fail a closed schema. A rename or type change can break a consumer, and an unchanged schema can still hide a changed meaning, such as a label vocabulary used under a new annotation policy. The small gate below permits extra fields, rejects invalid required values, and logs rows individually. It doesn't implement a batch drift test or verify annotation history.

raw_rows = [
    {"ticket_id": 101, "conversation_id": "c-51", "text": "Build is still failing", "label": "escalate"},
    {"ticket_id": 102, "conversation_id": "c-51", "text": "  BUILD is still failing  ", "label": "escalate"},
    {"ticket_id": 103, "conversation_id": "c-52", "text": "Where is the deploy policy?", "label": "standard"},
    {"ticket_id": 104, "conversation_id": "c-53", "text": "SEV-123 timeline has no reviewed label", "label": None},
]
required = {"ticket_id", "conversation_id", "text", "label"}
labels = {"standard", "escalate"}

def reject_reason(row: object) -> str | None:
    if not isinstance(row, dict):
        return "not_an_object"
    if not required.issubset(row):
        return "missing_field"
    if type(row["ticket_id"]) is not int or row["ticket_id"] <= 0:
        return "invalid_ticket_id"
    conversation = row["conversation_id"]
    if not isinstance(conversation, str) or not conversation.strip() or conversation != conversation.strip():
        return "invalid_conversation_id"
    if not isinstance(row["text"], str) or not row["text"].strip():
        return "blank_text"
    if not isinstance(row["label"], str) or row["label"] not in labels:
        return "invalid_label"
    return None

accepted, rejected = [], []
for row in raw_rows:
    reason = reject_reason(row)
    ticket_id = row.get("ticket_id") if isinstance(row, dict) else None
    if reason is None:
        accepted.append(ticket_id)
    else:
        rejected.append((ticket_id, reason))
print("accepted:", accepted)
print("rejected:", rejected)

accepted: [101, 102, 103]
rejected: [(104, 'invalid_label')]

The gate catches ticket 104 at the intake boundary. It checks that a label is present and in the vocabulary. An unreviewed or incorrect label spelled escalate would still pass. Reviewed status needs its own annotation record, policy version, or review gate. This first example also doesn't check uniqueness across rows; the assembled builder will.

Deterministic normalization and exact fingerprinting

In this fixture, the added spaces and capitalization don't change the intended message. Real tickets can contain code, case-sensitive names, tabs, and meaningful line breaks. Preserve the raw source and choose transformations for the task. Decode input bytes under an explicit encoding policy; don't indiscriminately strip control characters or Unicode characters because they look unusual.

Unicode defines canonical forms NFC/NFD and compatibility forms NFKC/NFKD. NFKC applies compatibility decomposition followed by canonical composition: for example, fi becomes fi, ² becomes 2, and full-width Latin letters become their narrower equivalents.^{[5]Reference 5Unicode Standard Annex 15: Unicode Normalization Formshttps://unicode.org/reports/tr15/} Following it with str.casefold(), which maps German ß to ss, and whitespace collapsing defines an equivalence rule for this lesson. Treat a match as a candidate for review: check whether normalization changed a distinction that matters to the task before merging or filtering records.

Conservative normalization has boundaries. Case-folding merges case-sensitive identifiers like Prod and prod, while compatibility normalization can alter code syntax. Never strip negation or environment names: build passed and build not passed must remain distinct records.

Once text is normalized, a deterministic SHA-256 hash becomes an exact fingerprint. It identifies normalized text for efficient lookup; it does not rank semantic similarity or establish that two records describe the same event. Compare normalized strings directly when exact identity matters, since cryptographic hashes can collide in principle.

import hashlib
import unicodedata

def normalize_text(text: str) -> str:
    normalized = unicodedata.normalize("NFKC", text)
    return " ".join(normalized.casefold().split())

def fingerprint(text: str) -> str:
    return hashlib.sha256(normalize_text(text).encode("utf-8")).hexdigest()

messages = [
    "Build is still failing",
    "  BUILD is still failing  ",
    "Build is not failing",
]
for message in messages:
    print(normalize_text(message), fingerprint(message)[:12])
print("first two duplicate:", fingerprint(messages[0]) == fingerprint(messages[1]))
print("negation preserved:", fingerprint(messages[0]) != fingerprint(messages[2]))

build is still failing c53011fb9834
build is still failing c53011fb9834
build is not failing e038c1b99d05
first two duplicate: True
negation preserved: True

The printed hashes are shortened for display; identity comparisons always use the full 64-character hex digest. Record the Python runtime and Unicode database versions alongside the normalization rule when reproducible bytes matter across architectures.^{[6]Reference 6unicodedata: Unicode Databasehttps://docs.python.org/3/library/unicodedata.html[7]Reference 7hashlib: Secure Hashes and Message Digestshttps://docs.python.org/3/library/hashlib.html}

Applying schema filtering and normalization to tickets 101 through 104 produces a clean separation: ticket 104 is rejected, 102 merges into 101, and 101 plus 103 proceed.

View larger

Duplicate labels and the quarantine boundary

Rows 201, 202, and 203 collapse to the same normalized text, but row 203 carries label standard while rows 201 and 202 carry escalate. Should the pipeline keep the first row, or quarantine the entire group?

A "first-in-wins" rule would silently choose a label. That hides annotation error, guideline drift, or missing context, such as customer service tier or deployment environment, that justified different decisions. Deduplication must account for the inputs available to the model and the intended weighting of examples.

A defensive pipeline isolates the entire fingerprint group into an anomaly quarantine. Label conflict is evidence to investigate, not noise to discard.

from collections import defaultdict
import unicodedata

def key(text: str) -> str:
    clean = " ".join(unicodedata.normalize("NFKC", text).casefold().split())
    return clean

rows = [
    {"ticket_id": 201, "text": "Migration blocked for 12 days", "label": "escalate"},
    {"ticket_id": 202, "text": " migration blocked for 12 days ", "label": "escalate"},
    {"ticket_id": 203, "text": "Migration blocked for 12 days", "label": "standard"},
    {"ticket_id": 204, "text": "Runbook link updated", "label": "standard"},
]
groups = defaultdict(list)
for row in rows:
    groups[key(row["text"])].append(row)

kept = []
quarantined = []
for duplicate_group in groups.values():
    observed_labels = {row["label"] for row in duplicate_group}
    if len(observed_labels) > 1:
        quarantined.extend(row["ticket_id"] for row in duplicate_group)
    else:
        kept.append(duplicate_group[0]["ticket_id"])

print("kept representative rows:", kept)
print("quarantined label conflict:", quarantined)

kept representative rows: [204]
quarantined label conflict: [201, 202, 203]

Grouping runs directly on the normalized string, so hash collisions can't fabricate a false label conflict. This isolated example has no conversation field and picks the first agreeing copy. The assembled builder below adds a more cautious policy: agreeing copies collapse within a conversation, while identical text across conversations is excluded pending review. Common templates can legitimately recur, so this policy changes the data distribution rather than proving that every excluded row was a duplicate event.

Grouped splits and target leakage prevention

Passing schema validation and deduplication doesn't make an example safe for evaluation. Choose the split to match the deployment question. A random row split can be appropriate for independent records, but turns within a conversation aren't independent new incidents.

If turns 1 and 3 of c-51 train the model while turn 2 tests it, shared server names and context can make evaluation easier. That is entity leakage when the intended target is unseen conversations. A model predicting a later turn in an already observed conversation may legitimately use earlier context, but it still can't use future turns.

For our unseen-conversation target, the split unit is conversation_id. Keep all its rows in one partition. If several conversations concern the same incident, customer, or copied document, a conversation key alone may be too narrow. Group those connected records at the level required by the evaluation claim.^{[8]Reference 8GroupKFoldhttps://scikit-learn.org/stable/modules/generated/sklearn.model_selection.GroupKFold.html}

Hashing each conversation key makes assignment stable when new keys arrive. The code uses the first eight hex digits of SHA-256, interprets them as an integer, and takes modulo 100. Thresholds allocate roughly 70%/15%/15% of distinct keys. Row proportions depend on how many rows belong to each key:

import hashlib

def split_for_conversation(conversation_id: str) -> str:
    bucket = int(hashlib.sha256(conversation_id.encode("utf-8")).hexdigest()[:8], 16) % 100
    if bucket < 70:
        return "train"
    if bucket < 85:
        return "validation"
    return "test"

rows = [
    {"ticket_id": 301, "conversation_id": "build-001"},
    {"ticket_id": 315, "conversation_id": "build-001"},
    {"ticket_id": 303, "conversation_id": "build-005"},
    {"ticket_id": 304, "conversation_id": "build-006"},
]
assignments = [(row["ticket_id"], split_for_conversation(row["conversation_id"])) for row in rows]
same_conversation_stays_together = assignments[0][1] == assignments[1][1]
print("assignments:", assignments)
print("same conversation stays together:", same_conversation_stays_together)

assignments: [(301, 'train'), (315, 'train'), (303, 'validation'), (304, 'test')]
same conversation stays together: True

Both turns of build-001 stay in train. build-005 hashes into validation and build-006 into test.

Stable hashing makes assignments repeatable. Because groups can contain different numbers of rows, partition sizes and label distributions still depend on the group mix. Audit both after applying grouped splits.

A future deployment requires a time boundary too: train on information available before a declared cutoff and evaluate on later predictions. A chronological split doesn't repair a feature computed using future incident outcomes. Reconstruct each feature as it was knowable at prediction time, allow for label delays, and fit learned preprocessing on training data alone. Decide explicitly how conversations that cross the cutoff are handled; grouping and chronological assignment can otherwise conflict.

Benchmark contamination and de-duplication audits

Grouped splits protect conversation boundaries. They don't prevent training data from containing locked benchmark questions and answers. A crawler may ingest a public benchmark, its solutions, or a discussion of them. That exposure weakens a claim of performance on unseen tasks, even when its effect on the score is hard to measure.

GPT-3's authors attempted to remove benchmark overlaps using 13-word matches before training. Their later audit used task-dependent overlap lengths, generally 8 to 13 words on non-synthetic tasks. A filtering bug left overlaps, especially in long documents, and retraining was too costly. They omitted four Wikipedia language-modeling benchmarks and the Children's Book Test because almost all examples overlapped and they couldn't extract a reliable clean subset.^{[9]Reference 9Language Models are Few-Shot Learners.https://arxiv.org/abs/2005.14165}

Start with an exact normalized overlap check between candidate training data and locked evaluation sets:

import hashlib
import unicodedata

def clean(text: str) -> str:
    return " ".join(unicodedata.normalize("NFKC", text).casefold().split())

def fp(text: str) -> str:
    return hashlib.sha256(clean(text).encode("utf-8")).hexdigest()

training = ["migration blocked for 12 days", "runbook link updated"]
locked_eval = ["MIGRATION blocked for 12 days", "incident review still pending"]
train_keys = {fp(text) for text in training}
collisions = [text for text in locked_eval if fp(text) in train_keys]

print("exact collisions:", collisions)
print("release evaluation:", len(collisions) == 0)

exact collisions: ['MIGRATION blocked for 12 days']
release evaluation: False

This detects a complete normalized-string match. It misses a benchmark question embedded inside a longer document, a paraphrase, or a question paired with a rewritten answer. An n-gram or substring audit can catch partial copies; broader similarity review can find some edited copies. Shared boilerplate also creates false alarms, so record the matched spans and the review decision.

When training data is unavailable, membership-inference methods such as Min-K% use model token probabilities to look for signs of prior exposure. Those scores also depend on text difficulty and distribution. Samuel et al.'s 2025 evaluation found method limitations and inconsistent detection, including difficulty with instruction-tuning contamination that changes answers.^{[10]Reference 10Towards Data Contamination Detection for Modern Large Language Models: Limitations, Inconsistencies, and Oracle Challengeshttps://aclanthology.org/2025.coling-main.338/} A negative test doesn't certify an uncontaminated model, and a positive score doesn't by itself prove memorization.

Protect locked evaluation material throughout pre-training, fine-tuning, retrieval, prompt examples, and judge construction. Private or newly collected tasks and dated dataset snapshots can strengthen the evidence, but neither n-gram filtering nor post-hoc testing guarantees an exposure-free benchmark.

Near-duplicate detection: MinHash, shingling, and LSH

Consider two support tickets: build has not passed and my build has not passed. Their exact SHA-256 fingerprints are completely different. Yet they share three out of four adjacent word pairs.

Break text into contiguous n-grams of words or characters called shingles. For word pairs ($n=2$):

• Document $A$: {"build has", "has not", "not passed"} (size 3)
• Document $B$: {"my build", "build has", "has not", "not passed"} (size 4)

Measure their set overlap with Jaccard similarity:

For this pair, the intersection contains 3 shared shingles and the union contains 4 distinct shingles:

A similarity of 0.75 crosses a 0.70 inspection threshold:

def shingles(text: str, width: int = 2) -> set[tuple[str, ...]]:
    if type(width) is not int or width < 1:
        raise ValueError("shingle width must be positive")
    words = text.casefold().split()
    if not words:
        return set()
    if len(words) < width:
        return {tuple(words)}
    return {tuple(words[index:index + width]) for index in range(len(words) - width + 1)}

def jaccard(left: str, right: str) -> float:
    a = shingles(left)
    b = shingles(right)
    union = a | b
    return len(a & b) / len(union) if union else 0.0

train_text = "build has not passed"
eval_text = "my build has not passed"
score = jaccard(train_text, eval_text)
print("shared pair score:", round(score, 2))
print("send to human review:", score >= 0.70)

shared pair score: 0.75
send to human review: True

For inputs shorter than two words, this toy shingler uses the whole word tuple as a fallback. For two empty strings, it returns a score of zero by convention rather than evaluating the undefined ratio $0/0$. The ingestion gate rejects blanks first. These choices need to be part of the recorded similarity policy.

All-pairs comparison examines $N(N-1)/2$ pairs, plus the work of comparing their shingle sets. MinHash compresses each set into a signature; LSH then uses those signatures to find promising pairs without examining every pair in typical workloads.^{[11]Reference 11On the Resemblance and Containment of Documents.https://www.cs.princeton.edu/courses/archive/spring13/cos598C/broder97resemblance.pdf}

In the ideal construction, choose $K$ independent uniform random permutations of the shingle universe. Treat each permutation as a ranking $h_k$ and retain the smallest rank in a nonempty set:

The minimum-ranked shingle in $A \cup B$ is uniformly distributed over that union. The two minima agree exactly when that shingle belongs to $A \cap B$, so:

Practical implementations approximate the random rankings with hash functions; dependence and hash collisions can affect the estimate. The fraction of matching signature positions estimates Jaccard similarity.

LSH groups a signature's $K$ positions into $b$ bands of $r$ rows, where $K=b r$. A full matching band makes a candidate pair. Under the ideal independence assumptions, a pair with similarity $s$ becomes a candidate with probability:

This curve shows how candidate probability changes with similarity. FineWeb's documented 2024 configuration used word 5-grams and 112 hashes in 14 bands of 8. Evaluate the curve:

bands, rows_per_band = 14, 8
for similarity in [0.50, 0.70, 0.75, 0.85]:
    probability = 1 - (1 - similarity ** rows_per_band) ** bands
    print(f"J={similarity:.2f}: candidate probability {probability:.1%}")

J=0.50: candidate probability 5.3%
J=0.70: candidate probability 56.5%
J=0.75: candidate probability 77.2%
J=0.85: candidate probability 98.8%

About 22.8% of pairs at $J=0.75$ would be missed under this ideal model. Large buckets can also generate quadratically many candidate pairs, so record the bucket-size distribution and candidate-pair count for each corpus.^{[12]Reference 12The FineWeb Datasets: Decanting the Web for the Finest Text Data at Scalehttps://arxiv.org/abs/2406.17557}

Finding pairs and choosing which documents to retain are separate decisions. FineWeb formed connected duplicate clusters and kept one randomly chosen document per cluster. This can join two documents through an intermediate match even when they don't match each other.^{[12]Reference 12The FineWeb Datasets: Decanting the Web for the Finest Text Data at Scalehttps://arxiv.org/abs/2406.17557} Similarity above a threshold isn't transitive. Predict the weakest pair in this three-message chain:

messages = {
    "A": "build has failed",
    "B": "build has failed again",
    "C": "has failed again",
}

def pairs(text: str) -> set[tuple[str, str]]:
    words = text.split()
    return set(zip(words, words[1:]))

for left, right in [("A", "B"), ("B", "C"), ("A", "C")]:
    a, b = pairs(messages[left]), pairs(messages[right])
    score = len(a & b) / len(a | b)
    print(f"{left}/{right}: J={score:.3f}; matches at 0.60: {score >= 0.60}")

A/B: J=0.667; matches at 0.60: True
B/C: J=0.667; matches at 0.60: True
A/C: J=0.333; matches at 0.60: False

A connected-component rule would put all three in one cluster through B. Keeping one representative could remove a document that wasn't above the threshold with that representative. Under this lesson's policy, candidates go to human review. If you adopt automatic cluster deletion, evaluate this consequence and record the clustering and retention rules alongside the shingler and LSH configuration.

Lee et al.'s ACL 2022 experiments used exact repeated-substring removal and approximate document deduplication. Their deduplicated models emitted memorized training text about ten times less frequently and reached similar or better accuracy in the settings they studied.^{[13]Reference 13Deduplicating Training Data Makes Language Models Better.https://arxiv.org/abs/2107.06499} Results depend on the corpus, deduplication method, and task, so evaluate the selected deletion policy on your own data. FineWeb used MinHash; the original Dolma web pipeline instead used Bloom filters for URL, document, and paragraph deduplication. Bloom filters trade memory for a possibility of false-positive matches.^{[12]Reference 12The FineWeb Datasets: Decanting the Web for the Finest Text Data at Scalehttps://arxiv.org/abs/2406.17557[14]Reference 14Dolma: an Open Corpus of Three Trillion Tokens for Language Model Pretraining Researchhttps://arxiv.org/abs/2402.00159}

View larger

Privacy, PII detection, and provenance

Support tickets can contain personally identifiable information (PII), such as names, personal emails, or phone numbers. Credentials and internal incident details are sensitive too, even when they don't identify a person. An IP address or ticket ID needs context to classify. Models can memorize and later reproduce training text, so data minimization and access controls belong in the pipeline.

PII detection employs a layered defense:

1. Patterns and checks: Recognize candidate email addresses, IP addresses, or credential formats. Format validity doesn't establish whether a value is private, and one regex won't cover every representation.
2. Named entity recognition (NER): Proposes spans such as names or addresses. Entity types, languages, and detection quality depend on the model and configuration.
3. De-identification SDKs: Presidio combines configurable recognizers and NLP models. Originally created at Microsoft, it is now a community-governed project under Data Privacy Stack, according to its current FAQ checked October 10, 2026. Its maintainers explicitly say automated detection cannot find all sensitive information.^{[15]Reference 15Presidio: Data Protection and De-identification SDK.https://github.com/data-privacy-stack/presidio}

Two possible responses to a detected span are:

• Document-level dropping: Remove a record if its risk or access policy rules out retaining it. As one dated recipe, Dolma's original web pipeline dropped documents with more than five detected PII spans and masked spans in documents with five or fewer.^{[14]Reference 14Dolma: an Open Corpus of Three Trillion Tokens for Language Model Pretraining Researchhttps://arxiv.org/abs/2402.00159}
• Span masking: For sparse PII, sensitive spans get replaced with typed surrogate tokens (<EMAIL>, <PHONE_NUMBER>, <PERSON>).

import re

email_pattern = re.compile(r"\b[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}\b")
rows = [
    {"ticket_id": 501, "text": "Incident INC-10234 mentions a private timeline"},
    {"ticket_id": 502, "text": "Send updates to alex@example.com please"},
]

no_email_match = []
needs_redaction_review = []
for row in rows:
    if email_pattern.search(row["text"]):
        needs_redaction_review.append(row["ticket_id"])
    else:
        no_email_match.append(row["ticket_id"])

print("no email match in this narrow gate:", no_email_match)
print("redaction review required:", needs_redaction_review)

no email match in this narrow gate: [501]
redaction review required: [502]

A regex non-match doesn't prove safety: row 501 still contains an internal incident identifier (INC-10234). This gate queues a matched email for review; it doesn't redact text. Apply an explicit access and release policy, evaluate detectors on representative languages and formats, and keep raw sources, reject logs, and manifests under appropriate access controls. Masking one field doesn't establish that the remaining context is anonymous or safe to publish.

The terms provenance and lineage overlap. Here we'll use them with different emphasis:

• Provenance: Captures origin, custody, collection guidelines, and legal licensing.
• Lineage: Maps inputs, transformations, and outputs. OpenLineage run events can record a job, run, and input/output datasets; the completeness of that record depends on the metadata and integrations actually emitted.^{[16]Reference 16The Run Cyclehttps://openlineage.io/docs/spec/run-cycle/}

Dataset versioning connects code with data state. DVC keeps small tracking files in Git and uses a content-addressed cache, with optional remote storage such as S3. S3 itself isn't inherently content-addressed; DVC's layout supplies that behavior.^{[17]Reference 17Internal Directories and Files: Content-Addressed Cachehttps://github.com/treeverse/dvc.org/blob/main/content/docs/user-guide/project-structure/internal-files.md} Delta Lake and Apache Iceberg track table versions for time travel, but retention matters: Delta VACUUM can delete files needed by an old version, and expired Iceberg snapshots cease to be queryable.^{[18]Reference 18Delta Lake Time Travelhttps://delta.io/blog/2023-02-01-delta-lake-time-travel/[19]Reference 19Maintenance: Expire Snapshotshttps://iceberg.apache.org/docs/latest/maintenance/} Preserve the versions and metadata an experiment needs rather than assuming history lasts forever.

Modern foundation model data pipelines: web filtering and synthetic gates

A web-scale builder faces the same decisions with many more records. Text extraction, filtering, deduplication, and sampling change what the model learns; their effects interact with the model, training budget, and target tasks. FineWeb's controlled data ablations show why a recipe needs measurement: aggressive global deduplication underperformed their alternative of deduplicating individual crawl snapshots.^{[12]Reference 12The FineWeb Datasets: Decanting the Web for the Finest Text Data at Scalehttps://arxiv.org/abs/2406.17557}

Those choices show up again as filters, and each filter changes which documents the model is allowed to see:

1. Heuristic text filtering

Rules based on document structure can drop broken extracts and boilerplate without running a model:

• Length bounds: Flag very short or unusually long records. A command reference and a long tutorial have different useful ranges.
• Symbol ratios: Flag extraction failures. Code, equations, and logs legitimately contain many symbols, so a prose rule can remove valuable technical data.
• Language-specific word checks: Some recipes use common-word counts as a text-quality signal. English stopword thresholds aren't a universal rule for other languages.
• Repetition filters: Flag repeated lines, n-grams, or paragraphs. Inspect whether repetition is scraper noise, a template, or useful structure.

Choose thresholds using reviewed samples and downstream ablations. Record removal rates by language, source, and domain; a filter can make one benchmark better while shrinking coverage elsewhere.

2. Model-based perplexity filtering

A reference language model can score how expected a document is under a chosen text distribution. CCNet used language-model perplexity to select text closer to Wikipedia; that is a documented recipe, rather than a required stage in every corpus.^{[20]Reference 20CCNet: Extracting High Quality Monolingual Datasets from Web Crawl Datahttps://arxiv.org/abs/1911.00359}

The reference model scores the perplexity of each candidate document:

Here $x_i$ is a token under the reference model's tokenizer and $N$ is the number scored. Compare perplexity values only when tokenization and scoring procedures match. High values can reflect OCR noise, unfamiliar jargon, another language, or a rare dialect. Low values can reflect predictable boilerplate or useful common prose. Test any selection rule against the data quality you want to improve.

3. Synthetic data quality gating

Synthetic instruction and preference data need additional checks because a generator can produce convincing errors. Keep the teacher model and generation settings in the provenance record, and keep locked evaluation material out of teacher prompts:

• Format validation: Check the format your training interface requires, such as JSON fields or response boundaries. Well-formed text can still be wrong.
• Execution or domain verification: For coding tasks, run meaningful tests in an appropriately isolated environment. Passing a finite test suite covers its tested cases, not every input. Math needs a checker that fits the task, not a script that only has to run.
• Judge-assisted review: Use evaluator scores to prioritize or filter examples, then calibrate against human or verifiable labels. A fluent teacher and judge can share the same error, so a high judge score alone doesn't establish truth.

Building a reproducible candidate artifact

The separate checks still have to agree on one artifact. The pipeline below validates raw records, normalizes text, excludes conflicting groups, assigns grouped splits, and writes JSONL outputs, a rejection log, and an unsigned manifest. A hash records byte identity. A digital signature would require a signing key and a verification procedure, and this example implements neither.

A candidate artifact means these mechanics executed successfully. It doesn't mean privacy or contamination review has signed off on release.

import hashlib
import json
import platform
import unicodedata
from collections import Counter, defaultdict
from pathlib import Path

raw_rows = [
    {"ticket_id": 401, "conversation_id": "c-a", "text": "Build is still failing", "label": "escalate"},
    {"ticket_id": 402, "conversation_id": "c-a", "text": " BUILD is still failing ", "label": "escalate"},
    {"ticket_id": 403, "conversation_id": "c-i", "text": "Runbook page shows stale", "label": "standard"},
    {"ticket_id": 404, "conversation_id": "c-c", "text": "SEV-123 timeline has no reviewed label", "label": None},
    {"ticket_id": 405, "conversation_id": "c-002", "text": "Break-glass request will not open", "label": "standard"},
    {"ticket_id": 406, "conversation_id": "c-e", "text": "Migration ran twice in production", "label": "escalate"},
    {"ticket_id": 407, "conversation_id": "c-l", "text": "Deploy completed this morning", "label": "standard"},
    {"ticket_id": 408, "conversation_id": "build-003", "text": "Build stalled after approval", "label": "escalate"},
    {"ticket_id": 409, "conversation_id": "c-m", "text": "Access request pending review", "label": "standard"},
    {"ticket_id": 410, "conversation_id": "c-m", "text": "Access request pending review", "label": "escalate"},
    {"ticket_id": 411, "conversation_id": "c-x", "text": "Access policy page missing", "label": "standard"},
    {"ticket_id": 412, "conversation_id": "c-y", "text": "Access policy page missing", "label": "standard"},
]
valid_labels = {"standard", "escalate"}
required_fields = {"ticket_id", "conversation_id", "text", "label"}

def normalize_text(text: str) -> str:
    return " ".join(unicodedata.normalize("NFKC", text).casefold().split())

def text_fingerprint(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

def reject_reason(row: object) -> str | None:
    if not isinstance(row, dict):
        return "not_an_object"
    if not required_fields.issubset(row):
        return "missing_field"
    if type(row["ticket_id"]) is not int or row["ticket_id"] <= 0:
        return "invalid_ticket_id"
    conversation = row["conversation_id"]
    if not isinstance(conversation, str) or not conversation.strip() or conversation != conversation.strip():
        return "invalid_conversation_id"
    if not isinstance(row["text"], str) or not row["text"].strip():
        return "blank_text"
    if not isinstance(row["label"], str) or row["label"] not in valid_labels:
        return "invalid_label"
    return None

def stable_split(conversation_id: str) -> str:
    bucket = int(hashlib.sha256(conversation_id.encode("utf-8")).hexdigest()[:8], 16) % 100
    if bucket < 70:
        return "train"
    if bucket < 85:
        return "validation"
    return "test"

def prepare_rows(input_rows):
    id_counts = Counter(
        row["ticket_id"] for row in input_rows
        if isinstance(row, dict) and type(row.get("ticket_id")) is int
    )
    rejected = []
    prepared_by_text = defaultdict(list)
    for row in input_rows:
        reason = reject_reason(row)
        ticket_id = row.get("ticket_id") if isinstance(row, dict) else None
        if reason is None and id_counts[ticket_id] > 1:
            reason = "duplicate_ticket_id"
        if reason:
            rejected.append({"ticket_id": ticket_id, "reason": reason})
            continue
        text = normalize_text(row["text"])
        prepared_by_text[text].append({
            "ticket_id": ticket_id,
            "conversation_id": row["conversation_id"],
            "text": text,
            "label": row["label"],
            "text_sha256": text_fingerprint(text),
        })

    kept = []
    for duplicate_group in prepared_by_text.values():
        labels = {row["label"] for row in duplicate_group}
        conversations = {row["conversation_id"] for row in duplicate_group}
        reason = ("label_conflict" if len(labels) > 1 else
                  "cross_conversation_duplicate" if len(conversations) > 1 else None)
        if reason:
            rejected.extend({"ticket_id": row["ticket_id"], "reason": reason}
                            for row in duplicate_group)
            continue
        duplicate_group.sort(key=lambda row: row["ticket_id"])
        representative = duplicate_group[0]
        kept.append({**representative, "split": stable_split(representative["conversation_id"])})
        rejected.extend({"ticket_id": row["ticket_id"], "reason": "exact_duplicate"}
                        for row in duplicate_group[1:])

    kept.sort(key=lambda row: row["ticket_id"])
    rejected.sort(key=lambda row: (str(row["ticket_id"]), row["reason"]))
    return kept, rejected

kept, rejected = prepare_rows(raw_rows)
assert len(kept) + len(rejected) == len(raw_rows)

output = Path("artifacts")
output.mkdir(exist_ok=True)
rows_path = output / "dev_tickets.v1.jsonl"
rejects_path = output / "dev_tickets.v1.rejects.jsonl"
rows_path.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in kept), encoding="utf-8", newline="\n")
rejects_path.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in rejected), encoding="utf-8", newline="\n")
artifact_sha256 = hashlib.sha256(rows_path.read_bytes()).hexdigest()
manifest = {
    "dataset_id": "developer-ticket-routing",
    "version": "v1",
    "status": "candidate_not_released",
    "pipeline_version": "nfkc-casefold-space__within-conversation-dedup__cross-group-review__v1",
    "source_snapshot": "synthetic-ticket-fixture-v1",
    "source_sha256": hashlib.sha256(json.dumps(raw_rows, sort_keys=True).encode("utf-8")).hexdigest(),
    "python_version": platform.python_version(),
    "unicode_version": unicodedata.unidata_version,
    "split_policy": "sha256(conversation_id)[:8] modulo 100; 70/15/15 thresholds",
    "rows_raw": len(raw_rows),
    "rows_kept": len(kept),
    "reject_reasons": dict(Counter(row["reason"] for row in rejected)),
    "split_counts": dict(Counter(row["split"] for row in kept)),
    "artifact_sha256": artifact_sha256,
    "rejects_sha256": hashlib.sha256(rejects_path.read_bytes()).hexdigest(),
    "pending_reviews": ["split_coverage", "locked_benchmark_overlap", "near_duplicates", "privacy_and_license"],
}
(output / "dev_tickets.v1.manifest.json").write_text(
    json.dumps(manifest, indent=2, sort_keys=True) + "\n",
    encoding="utf-8",
    newline="\n",
)
print("kept rows:", manifest["rows_kept"], "rejected:", manifest["reject_reasons"])
print("split counts:", manifest["split_counts"])
print("artifact digest prefix:", artifact_sha256[:16])
print("status:", manifest["status"])

kept rows: 6 rejected: {'exact_duplicate': 1, 'invalid_label': 1, 'label_conflict': 2, 'cross_conversation_duplicate': 2}
split counts: {'train': 3, 'validation': 2, 'test': 1}
artifact digest prefix: ba384792199e9be7
status: candidate_not_released

The fixed serialization order, UTF-8 encoding, and explicit LF newlines make the fixture's output bytes reproducible under the recorded transformation environment. Running it again overwrites the local v1 paths; it doesn't enforce immutable storage. A production publisher needs a new release identifier or content-addressed location, retained source snapshots, and a trusted record of the release digest.

This manifest captures operational fields. A complete dataset datasheet also documents motivation, composition, collection, uses, limitations, and maintenance so consumers can evaluate fit.^{[21]Reference 21Datasheets for Datasetshttps://arxiv.org/abs/1803.09010} Add annotation policy and review history, collection permissions, known gaps, and retention responsibilities to the accompanying documentation.

Notice the manifest explicitly tracks pending reviews: split coverage, benchmark overlap, near-duplicates, and privacy. Don't mark a dataset ready for release until those checks pass.

Verifying the candidate contract like downstream code

Downstream training code shouldn't blindly ingest files because they exist on disk. A production data loader executes pre-flight assertions before spending GPU compute:

1. Zero group leakage: No conversation ID spans multiple splits.
2. Deterministic split adherence: Every record matches the stable hash assignment.
3. Cryptographic integrity: File bytes match artifact_sha256 and rejects_sha256 in the manifest.
4. Structural validity: 100% of rows pass schema requirements.
5. Fingerprint consistency: Recomputed SHA-256 fingerprints match row metadata.
6. Accounting balance: kept + rejected == total raw rows.

import hashlib
import json
from collections import Counter, defaultdict

rows = [json.loads(line) for line in rows_path.read_text(encoding="utf-8").splitlines()]
saved_manifest = json.loads((output / "dev_tickets.v1.manifest.json").read_text(encoding="utf-8"))
saved_rejects = [json.loads(line) for line in rejects_path.read_text(encoding="utf-8").splitlines()]
groups = defaultdict(set)
for row in rows:
    groups[row["conversation_id"]].add(row["split"])

no_group_leakage = all(len(splits) == 1 for splits in groups.values())
split_rule_matches = all(row["split"] == stable_split(row["conversation_id"]) for row in rows)
digest_matches = hashlib.sha256(rows_path.read_bytes()).hexdigest() == saved_manifest["artifact_sha256"]
reject_digest_matches = hashlib.sha256(rejects_path.read_bytes()).hexdigest() == saved_manifest["rejects_sha256"]
schema_valid = all(reject_reason(row) is None for row in rows)
fingerprints_match = all(
    row["text"] == normalize_text(row["text"])
    and row["text_sha256"] == text_fingerprint(row["text"])
    for row in rows
)
identities_unique = (len({row["ticket_id"] for row in rows}) == len(rows)
                     and len({row["text"] for row in rows}) == len(rows))
counts_match = (
    len(rows) == saved_manifest["rows_kept"]
    and len(rows) + len(saved_rejects) == saved_manifest["rows_raw"]
    and dict(Counter(row["split"] for row in rows)) == saved_manifest["split_counts"]
    and dict(Counter(row["reason"] for row in saved_rejects)) == saved_manifest["reject_reasons"]
)
checks = {
    "conversation and split rule": no_group_leakage and split_rule_matches,
    "artifact and rejection digests": digest_matches and reject_digest_matches,
    "row schema": schema_valid,
    "recomputed fingerprints": fingerprints_match,
    "unique IDs and normalized texts": identities_unique,
    "manifest counts": counts_match,
}
for name, passed in checks.items():
    print(f"{name}: {passed}")
if not all(checks.values()):
    raise ValueError(f"candidate contract failed: {checks}")

conversation and split rule: True
artifact and rejection digests: True
row schema: True
recomputed fingerprints: True
unique IDs and normalized texts: True
manifest counts: True

The checks cover group isolation, byte identity, structure, fingerprints, uniqueness, and accounting. Several may fail for the same mutation. A passing run confirms these listed invariants. An attacker who changes both files and the untrusted manifest can replace the expected hashes too, so obtain the manifest or digest from a trusted release record to establish publisher authenticity.

Auditing distribution drift and split coverage

Even if every row passes structural checks, the dataset might still fail downstream needs. If the validation split contains zero escalate cases, validation accuracy can't evaluate whether your model detects high-priority incidents.

Always audit label coverage and feature distributions across splits:

from collections import defaultdict

required_labels = {"standard", "escalate"}
coverage = defaultdict(set)
for row in rows:
    coverage[row["split"]].add(row["label"])

missing = {
    split: sorted(required_labels - coverage[split])
    for split in ["train", "validation", "test"]
    if required_labels - coverage[split]
}
print("coverage:", {split: sorted(values) for split, values in coverage.items()})
print("missing labels:", missing)
print("label coverage passes:", not missing)

coverage: {'train': ['escalate'], 'validation': ['standard'], 'test': ['standard']}
missing labels: {'train': ['standard'], 'validation': ['escalate'], 'test': ['escalate']}
label coverage passes: False

This fixture fails label coverage: train contains only escalate examples while validation and test contain only standard examples.

Use counts and practical distances before reaching for a significance test:

• Categorical features: Compare label and source proportions. A chi-square independence test needs adequate expected counts and a suitable sampling design.
• Continuous features: Compare lengths or other feature distributions using plots, quantiles, or Wasserstein distance. A two-sample Kolmogorov-Smirnov (KS) test compares continuous distributions under independent sampling assumptions; discrete counts such as token lengths need appropriate handling of ties.^{[22]Reference 22ks_2samp: Two-Sample Kolmogorov-Smirnov Testhttps://docs.scipy.org/doc/scipy/reference/generated/scipy.stats.ks_2samp.html[23]Reference 23wasserstein_distance: Wasserstein-1 Distancehttps://docs.scipy.org/doc/scipy/reference/generated/scipy.stats.wasserstein_distance.html}

Population Stability Index (PSI) and Wasserstein distance are discrepancy measures, not automatically calibrated hypothesis tests. Repeated turns within a conversation also invalidate a naive independent-row interpretation. Huge datasets can make tiny differences statistically significant; the six-row fixture is too sparse for an ordinary chi-square approximation. Future deployment data may legitimately differ from earlier training data, so identical distributions aren't always the goal.

When a partition fails coverage, collect more reviewed groups or design a stratified grouped split before comparing models. Group sizes and label mixtures can make perfect balance infeasible. Preserve group boundaries and the deployment target, and record any changed split as a new artifact. Moving examples after observing test performance makes the test part of model development.

Tracking dataset versions across pipeline evolutions

Suppose pipeline v2 adds an annotation or updates the normalization regex. If model accuracy shifts, did the architecture improve, or did the underlying data change?

A dataset digest and pipeline version identify which data changed. They don't establish that a performance difference was caused by that change: model settings, seeds, and evaluation policy can change too. Hold those factors fixed for a controlled comparison.

This small example identifies two different byte artifacts:

import hashlib
import json

def digest(rows: list[dict]) -> str:
    payload = "\n".join(json.dumps(row, sort_keys=True) for row in rows) + "\n"
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

v1 = [{"text": "build missing", "label": "escalate"}]
v2 = [*v1, {"text": "break-glass request broken", "label": "standard"}]

print("v1:", digest(v1)[:12], "rows:", len(v1))
print("v2:", digest(v2)[:12], "rows:", len(v2))
print("version changed:", digest(v1) != digest(v2))

v1: 6045599184e8 rows: 1
v2: 551369eeb269 rows: 2
version changed: True

Hugging Face Datasets derives cache fingerprints from dataset state and transformations; a non-hashable transformation can receive a random fingerprint with a warning.^{[24]Reference 24The Cache: Dataset and Transformation Fingerprintshttps://huggingface.co/docs/datasets/about_cache} That cache key serves reuse rather than authenticating a release. Store the published artifact's cryptographic digest explicitly alongside its code version and experiment run.

Mapping pipeline gates to enterprise AI systems

The ticket fixture teaches decisions that also appear in larger pipelines. Their implementations and tradeoffs change with scale:

Never quote an evaluation metric without naming the exact dataset artifact, split policy, and decontamination audit that produced it.

Controlled failure labs: break the pipeline on purpose

Run these targeted experiments against the runnable code blocks above to observe how specific assertions catch mutations:

1. Simulate entity leakage: In 08-verify-artifact.py, append {**rows[0], "ticket_id": 999, "split": "test"} to rows before computing groups.
2. Simulate bit rot or file tampering: In 08-verify-artifact.py, append a newline to rows_path after loading saved_manifest and before evaluating digest_matches.
3. Evaluate boundary conditions in shingling: In 06-shingle-overlap.py, print jaccard("build", "build") and jaccard("", "").
4. Violate grouped split integrity: In 04-stable-grouped-split.py, assign partitions by hashing str(row["ticket_id"]) rather than row["conversation_id"].

What each failure reveals

1. no_group_leakage, identities_unique, and manifest counts fail simultaneously. Multiple assertions catch the corruption from different perspectives.
2. digest_matches becomes False. Even an invisible trailing byte breaks cryptographic identity.
3. A single matching word returns 1.0; two empty strings return 0.0. Blank strings must be caught by schema gates, not similarity estimators.
4. Ticket 301 hashes to train while ticket 315 hashes to validation, even though both belong to build-001. The group boundary fails. Whether a trained model's score rises is a separate empirical question.

Carrying clean data into feature engineering

A dataset manifest records the source, transformations, split policy, and resulting byte digests. Retained snapshots let reviewers reproduce and investigate that record. Hashes don't lock files against changes, and passing structural checks leaves semantic and release decisions to resolve.

Reject a bad row, give case and spacing variants one fingerprint, keep a conversation on one side of a split, and tell a matching digest from a trusted release record. Each feature still has to have been knowable when the prediction was made, and it has to mean the same thing in training and in live serving.