Validate incoming data

Problem

When you need this. A nightly event batch contains a malformed line and two different records with the same event ID. Dropping whichever row the parser dislikes can produce a clean-looking training file while changing which events the dataset represents. Keeping the first duplicate makes the result depend on input order.

You will preserve the source, partition every line into accepted or quarantined records, and write a manifest that controls training eligibility. The synthetic bad batch has 20 lines: seven are staged as accepted, 13 are quarantined, and the batch remains blocked. The existing availability validator supplies the event schema and timestamp checks.

Concepts

Parsing checks whether bytes form a supported record. Schema validation checks whether that record has the required fields, values and timing relationships. Availability policy is a later question: a valid late event may still be unavailable to a historical prediction.

Quarantine retains rejected material and its reason for investigation. A batch manifest is the JSON file listing source/output hashes, line counts, validation versions and training eligibility. Accepted rows in a partly invalid batch are staging output; they are not automatically a repaired training dataset.

Architecture

The validator retains the input bytes, parses and validates individual lines, then finds conflicting event identities across parsed records. It partitions the lines into accepted and quarantined outputs and writes a blocked or eligible manifest. The command’s exit status follows that manifest rather than the existence of an accepted file.

Working code

Preserve the input before interpreting it

The source uses JSON Lines: one UTF-8 JSON value per newline-delimited record. The format permits a JSON value such as null, but this application’s event contract requires an object. A blank line is not a valid JSON Lines record.

The validator reads original bytes and keeps each line with its source position. Quarantine records include the original line encoded as base64, its hash and failure reasons. Even a line containing invalid UTF-8 remains recoverable. Base64 is an encoding, not redaction; quarantine data needs the same access policy as its source.

Download the complete lab and enter mlops-dataset-validation/:

python3.12 -m venv .venv
source .venv/bin/activate
python inspect_dataset.py
python -m unittest -v test_quarantine.py

The fixture is original synthetic user-event data. The program writes and verifies temporary staging directories, then reports their manifests. It uses the standard library; there is no training, model call or data download.

Separate format failures from event-contract failures

The parser rejects invalid UTF-8, malformed JSON, duplicate JSON keys and nonstandard constants such as NaN. It also enforces a 512-byte line limit, including the newline. JSON parsing alone is insufficient: Python’s default decoder accepts repeated object keys and nonstandard numeric constants, so this parser supplies explicit hooks.

"""Preserve raw JSON Lines records, apply the existing event contract and gate a batch."""
import argparse
import base64
from collections import Counter, defaultdict
from dataclasses import dataclass, field
import hashlib
import io
import json
from pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent / 'vendor'))
from availability import validate as validate_event

SCHEMA_VERSION = 'event-v1'
RULES_VERSION = '1'
MAX_LINE_BYTES = 512
SCHEMA_SHA = hashlib.sha256((Path(__file__).parent / 'vendor/availability.py').read_bytes()).hexdigest()


class DuplicateKey(ValueError): pass
class NonstandardConstant(ValueError): pass


def unique_object(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise DuplicateKey('duplicate JSON key: ' + key)
        value[key] = item
    return value


def reject_constant(value):
    raise NonstandardConstant('nonstandard JSON constant: ' + value)


@dataclass
class Entry:
    line: int
    raw: bytes
    record: dict | None = None
    reasons: list = field(default_factory=list)

    def reject(self, code, detail):
        self.reasons.append({'code': code, 'detail': detail})


def validate_batch(raw, max_line_bytes=MAX_LINE_BYTES):
    if not isinstance(raw, bytes):
        raise ValueError('input must be the original bytes')
    if type(max_line_bytes) is not int or not 1 <= max_line_bytes <= 1_048_576:
        raise ValueError('invalid line-byte limit')
    entries, identities = [], defaultdict(list)
    for number, blob in enumerate(io.BytesIO(raw), 1):
        entry = Entry(number, blob); entries.append(entry)
        if len(blob) > max_line_bytes:
            entry.reject('line_too_large', f'{len(blob)} bytes exceeds {max_line_bytes}')
            continue
        if not blob.strip(b' \t\r\n'):
            entry.reject('blank_record', 'JSON Lines needs one value on each line')
            continue
        try:
            value = json.loads(blob.decode('utf-8'), object_pairs_hook=unique_object, parse_constant=reject_constant)
        except UnicodeError:
            entry.reject('invalid_utf8', 'line is not UTF-8'); continue
        except DuplicateKey as error:
            entry.reject('duplicate_json_key', str(error)); continue
        except NonstandardConstant as error:
            entry.reject('nonstandard_json_constant', str(error)); continue
        except json.JSONDecodeError as error:
            entry.reject('malformed_json', error.msg); continue
        if not isinstance(value, dict):
            entry.reject('record_must_be_object', 'valid JSON is not necessarily a valid event'); continue
        entry.record = value
        identity = value.get('id')
        if isinstance(identity, str) and 1 <= len(identity) <= 100:
            identities[identity].append(entry)
        try:
            validate_event([], [value], [])
        except ValueError as error:
            entry.reject('event_contract', str(error))
    # Refuse the entire parseable identity group, even when one row has another error.
    for identity, group in identities.items():
        if len(group) > 1:
            for entry in group:
                entry.reject('duplicate_event_id', identity)
    accepted = [entry for entry in entries if not entry.reasons]
    quarantined = [entry for entry in entries if entry.reasons]
    return {'source': raw, 'entries': entries, 'accepted': accepted, 'quarantined': quarantined,
            'max_line_bytes': max_line_bytes,
            'status': 'ready' if entries and not quarantined else 'blocked'}


def sha(raw):
    return hashlib.sha256(raw).hexdigest()


def write_batch(batch, output_dir):
    target = Path(output_dir)
    target.mkdir()  # Refuse an existing directory rather than mix or replace runs.
    files = {'source.bin': batch['source'], 'accepted.jsonl': b''.join(e.raw for e in batch['accepted'])}
    quarantined = [{'source_line': e.line, 'raw_sha256': sha(e.raw),
                    'raw_base64': base64.b64encode(e.raw).decode(), 'reasons': e.reasons}
                   for e in batch['quarantined']]
    files['quarantine.jsonl'] = ''.join(json.dumps(e, sort_keys=True) + '\n' for e in quarantined).encode()
    for name, raw in files.items():
        (target / name).write_bytes(raw)
    manifest = {'schema_version': SCHEMA_VERSION, 'schema_implementation_sha256': SCHEMA_SHA,
                'rules_version': RULES_VERSION, 'max_line_bytes': batch['max_line_bytes'],
                'source_sha256': sha(batch['source']), 'source_lines': len(batch['entries']),
                'accepted': len(batch['accepted']), 'quarantined': len(batch['quarantined']),
                'accepted_source_lines': [e.line for e in batch['accepted']],
                'status': batch['status'],
                'training_file': 'accepted.jsonl' if batch['status'] == 'ready' else None,
                'batch_reasons': ['empty_batch'] if not batch['entries'] else ['invalid_records'] if batch['quarantined'] else [],
                'reason_counts': dict(sorted(Counter(r['code'] for e in batch['quarantined'] for r in e.reasons).items())),
                'files': {name: sha(raw) for name, raw in files.items()}}
    # Written last: a partially written staging directory must not look ready.
    (target / 'manifest.json').write_text(json.dumps(manifest, indent=2) + '\n')
    return manifest


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('input', type=Path)
    parser.add_argument('--out-dir', type=Path, required=True)
    args = parser.parse_args()
    batch = validate_batch(args.input.read_bytes())
    manifest = write_batch(batch, args.out_dir)
    print(f"{manifest['status']}: accepted={manifest['accepted']}; quarantined={manifest['quarantined']}")
    raise SystemExit(0 if manifest['status'] == 'ready' else 1)


if __name__ == '__main__':
    try:
        main()
    except (OSError, ValueError) as error:
        raise SystemExit(f'input or staging error: {error}')

After parsing, the existing event validator checks the required fields, bounded IDs, UTC times and arrival ordering. An arrival before an event happened is invalid. An arrival after a prediction is different: it can be valid source data even though the historical feature query must exclude it.

test_late_and_future_events_are_not_schema_errors keeps the late-arriving and later events. Quarantining every inconvenient timestamp would hide the availability problem taught by the previous lesson rather than solve it.

The implementation loads this small input file and retains its records in memory. Its line-byte rule is not a total memory bound or a streaming-ingestion design. A production loader would need separate input-size, buffering and storage decisions.

Quarantine the whole conflicting identity group

The second pass groups parseable records by event ID. If an ID occurs more than once, every record in that group receives a duplicate-ID reason. This rule applies even if one group member already failed another check.

Lines 1 and 15 both name u1-known but carry different arrival times. Both are quarantined. Line 2, u1-late, remains an accepted source event. Reversing the batch cannot make the first duplicate win; the identity-group test requires the same refusal of both records.

This is a conservative conflict policy, not an event-correction protocol. A system that supports corrections needs revision and source-precedence rules. It should not infer a valid correction merely because a conflicting row arrived later in a file.

test_an_invalid_duplicate_cannot_leave_a_valid_first_row_accepted adds an unknown field to one duplicate. The other row still cannot pass by itself. Otherwise an error in one record would silently choose the surviving version of the event.

Write a blocked manifest rather than silently dropping rows

Synthetic event batches; original bytes retained, no automatic repair
events.jsonl: blocked; source=20; accepted=7; quarantined=13; training_file=None
valid-events.jsonl: ready; source=8; accepted=8; quarantined=0; training_file=accepted.jsonl
The valid fixture is a separate reference batch, not the dirty batch with bad rows silently removed.

Each staging directory contains four files:

• source.bin: every original input byte.
• accepted.jsonl: the exact accepted lines for inspection.
• quarantine.jsonl: source positions, raw bytes and reasons for rejected lines.
• manifest.json: versions, hashes, counts, batch status and training eligibility.

The manifest names accepted.jsonl as training_file only when the nonempty batch has no quarantined records. The bad batch’s accepted subset exists for inspection, but training_file is null. Downstream jobs must check the manifest; finding a file with accepted rows is not permission to train on it.

The valid fixture is the separate original eight-event reference batch. It is not the bad batch with troublesome rows automatically deleted. The comparison demonstrates the gate, not an unreviewed repair algorithm.

Retain a staging batch and inspect its exit status

To keep the bad batch’s files in a new directory:

python quarantine.py events.jsonl --out-dir blocked-batch

The process reports blocked and exits with status 1. Inspect its quarantine records before deciding how the upstream source should be corrected. To stage the independent valid fixture:

python quarantine.py valid-events.jsonl --out-dir valid-batch

An existing output directory is refused. The validator writes the manifest last, so a partially written staging directory must not be treated as ready. This is a local staging convention, not an atomic release or durable-storage protocol.

The raw-byte test reconstructs the original input from all accepted and quarantined entries, including the invalid UTF-8 line. The manifest test checks actual file hashes. The CLI test executes both commands in temporary directories and verifies that exit status agrees with the batch gate.

Decide which validation rules need to evolve

python inspect_dataset.py --check
python inspect_dataset.py --save my-validation.json

The result record binds the source, validator and rule versions to both manifests. Reason counts can overlap because one record may violate more than one rule; accepted plus quarantined line counts still equal the source line count.

"""Write and verify two staging batches from original synthetic fixtures."""
import argparse
import hashlib
import json
from pathlib import Path
import tempfile
from quarantine import validate_batch, write_batch

ROOT = Path(__file__).resolve().parent


def record():
    reports = {}
    with tempfile.TemporaryDirectory(prefix='mlops-validation-') as temporary:
        for name in ('events.jsonl', 'valid-events.jsonl'):
            raw = (ROOT / name).read_bytes()
            target = Path(temporary) / name
            manifest = write_batch(validate_batch(raw), target)
            assert (target / 'source.bin').read_bytes() == raw
            assert all(hashlib.sha256((target / file).read_bytes()).hexdigest() == digest for file,digest in manifest['files'].items())
            reports[name] = manifest
    return {'mode': 'Local synthetic JSON Lines validation and real staging writes; no training or model',
            'files': {name: hashlib.sha256((ROOT / name).read_bytes()).hexdigest() for name in
                      ('quarantine.py', 'inspect_dataset.py', 'vendor/availability.py', 'events.jsonl', 'valid-events.jsonl')},
            'batches': reports}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--save', type=Path)
    parser.add_argument('--check', action='store_true')
    args = parser.parse_args()
    result = record()
    if args.check and result != json.loads((ROOT / 'result.json').read_text()):
        raise SystemExit('validation results or source bindings differ')
    if args.save:
        args.save.write_text(json.dumps(result, indent=2)+'\n')
    print('Synthetic event batches; original bytes retained, no automatic repair')
    for name, row in result['batches'].items():
        print(f"{name}: {row['status']}; source={row['source_lines']}; accepted={row['accepted']}; "
              f"quarantined={row['quarantined']}; training_file={row['training_file']}")
    print('The valid fixture is a separate reference batch, not the dirty batch with bad rows silently removed.')


if __name__ == '__main__':
    main()

Passing these rules means a batch satisfies this declared contract. It does not establish that events are factually correct, complete or representative. Changes in an upstream schema or correction policy need an explicit rule revision and a new validation result, rather than an exception that silently loosens the old one.

Design exercise: a producer adds an optional device field. Decide whether to reject the batch, introduce a new schema version or accept a documented projection. Explain which original bytes, rule version and source-to-output mapping you would retain for a later investigation.

Failure cases

Keeping the first duplicate makes the dataset depend on input order. The lab instead quarantines the entire conflicting identity group, including a valid-looking record paired with an invalid one. The original bad bytes remain recoverable so a repair can be reviewed rather than guessed.

Deployment

Write each staging attempt into a new directory and have downstream training require an eligible manifest. Review corrections against the retained raw batch and validate the replacement as a new input. Define who may approve a repaired batch; the validator does not make a domain correction or an approval decision.

Scaling

Duplicate detection needs visibility across the complete identity domain of a batch. Chunked or distributed ingestion must preserve that grouping rather than validating partitions independently and missing cross-partition conflicts. Track rejection reasons and byte limits without treating a low rejection count as a measure of dataset usefulness.

Key takeaways

• Valid rows and an eligible batch are different outcomes.
• A conflicting identity affects every member of its group.
• Raw-byte retention allows invalid encodings and parsing failures to be investigated.

Best practices

• Require manifest eligibility before selecting a training file.
• Keep parsing, schema and availability decisions distinct.
• Never overwrite the source batch or an earlier staging attempt.

References

• JSON Lines format: UTF-8, one JSON value per line and newline handling.
• Python, JSON: duplicate object keys, nonstandard constants and parser hooks.

Sources checked October 24, 2026. Counts and file hashes come from local execution of the bundled synthetic batches.