lesson 226 Reliable Data Pipelines · advanced

Ingest every line, quarantine the bad ones

External data arrives broken in small ways: a line cut off mid-transfer, a blank line, a value that isn't a record. Ingestion keeps every good record and sets each bad line aside with its line number and the reason, so nothing is lost and nothing crashes the run.

You'll be able to

Explain why ingestion must neither crash on one bad line nor drop bad lines silently
Parse JSON Lines with line numbers, catching only json.JSONDecodeError, and reject values that aren't objects
Account for every line: loaded + quarantined + blank equals the lines read

The idea

Ingestion is the first step of a pipeline: it moves raw data from outside (an hourly event file, a supplier export) into records your program can trust. Outside data is never perfect. A download is cut off halfway through a line, an editor adds a blank line, a producer sends a list where a record should be.

There are two bad ways to react. Crash on the first bad line, and one broken character in line 9,000 loses the whole hour. Skip bad lines silently, and data disappears with nobody knowing. The reliable pattern does neither: parse each line on its own, keep what parses, and quarantine what doesn't, with its line number and a reason, so someone can look at it, fix the cause and reload it.

Run each example to see what it prints. Running examples doesn't count toward progress.

example-1.py · Sort an hour of events into loaded and rejected

import json

lines = [
    '{"id": "48000000581", "type": "PushEvent", "repo": "tidewater/smartqueue"}',
    '{"id": "48000001631", "type": "CreateEvent", "repo": "lumen-labs/fastshell"',
    '',
    '["48000002275", "IssuesEvent"]',
    '{"id": "48000007504", "type": "WatchEvent", "repo": "lumen-labs/pyauth"}',
]

events = []
rejected = []
for number, line in enumerate(lines, start=1):
    if not line.strip():
        continue
    try:
        record = json.loads(line)
    except json.JSONDecodeError as err:
        rejected.append((number, "not valid JSON: " + err.msg))
        continue
    if not isinstance(record, dict):
        rejected.append((number, "not a JSON object"))
        continue
    events.append(record)

print(len(events), "events loaded")
for number, reason in rejected:
    print("line", number, "rejected:", reason)

Compare that with the shortcut most first versions take: parse everything in one comprehension. It works on clean test data and fails on the first real file.

example-2.py · One bad line stops the whole load

import json

lines = [
    '{"id": "48000000581", "type": "PushEvent"}',
    '{"id": "48000001631", "type": "CreateEv',
    '{"id": "48000007504", "type": "WatchEvent"}',
]
events = [json.loads(line) for line in lines]
print(len(events), "events loaded")

In a real pipeline the lines come from a file and the quarantine is a file too, one JSON object per rejected line. The parsing goes in its own small function that returns either a record or a reason, so the loop only decides where each line goes. Reading the file with for line in f keeps one line in memory at a time (see Files & Directories).

example-3.py · An ingest function with a quarantine file

import json

def parse_line(line):
    """Return (record, None) for a JSON object, or (None, reason) for anything else."""
    try:
        record = json.loads(line)
    except json.JSONDecodeError as err:
        return None, err.msg
    if not isinstance(record, dict):
        return None, "not a JSON object"
    return record, None

def ingest(path, quarantine_path):
    events = []
    with open(path, encoding="utf-8") as source, open(quarantine_path, "w", encoding="utf-8") as bad:
        for number, line in enumerate(source, start=1):
            if not line.strip():
                continue
            record, reason = parse_line(line)
            if reason is None:
                events.append(record)
            else:
                bad.write(json.dumps({"line": number, "reason": reason}) + "\n")
    return events

with open("2025-03-03-00.jsonl", "w", encoding="utf-8") as f:
    f.write('{"id": "48000000581", "type": "PushEvent"}\n')
    f.write('{"id": "48000001631", "type": "CreateEvent"}\n')
    f.write('{"id": "48000007504", "type": "WatchEv"\n')
    f.write('\n')
    f.write('"48000002275"\n')
    f.write('{"id": "48000006895", "type": "IssueCommentEvent"}\n')

events = ingest("2025-03-03-00.jsonl", "quarantine.jsonl")
print([event["type"] for event in events])
with open("quarantine.jsonl", encoding="utf-8") as f:
    print(f.read(), end="")

When to use it: every time data comes from outside your program: files, exports, message queues, API dumps. Mistakes to avoid: except Exception around the whole loop body, which also hides bugs in your own code (a typo becomes "bad data"); numbering lines from 0; reporting blank lines as broken; and quarantining without the line number, so nobody can find the line again. Check that loaded + rejected + blank adds up to the lines you read.

Review

Parse each line on its own; catch only the parser's error, and reject values that aren't records.
Skip blank lines, number lines from 1, and quarantine bad ones with their position and reason.
Loaded + quarantined + blank = lines read. A file that is mostly bad should stop the run instead.
Next: a line can parse and still be wrong. A schema contract says what a good record looks like.

Practise this