lesson 249 Production Readiness Capstone · advanced

Make the pipeline survive a bad night

In production, everything goes wrong at once: a few rows are malformed, the report storage times out now and then, and the job is killed halfway through. A resilient pipeline combines the safeguards from the Reliable Data Pipelines module and the Observability & Operations module, each for the failure it fits, and counts every outcome.

You'll be able to

Match each failure to its safeguard: bad data → quarantine; transient errors → retry; crashes → checkpoint and resume; a poison batch → set aside
Write a batch loop that retries writes, saves progress after each batch and resumes from it
Report the night with counters: loaded, rejected, retries, batches done, batches set aside

The idea

Each earlier lesson handled one kind of failure. A real night brings several, and each needs a different response:

| Failure | Example | Response |
| --- | --- | --- |
| Bad data | a total of n/a | quarantine the row, keep going (see Reliable Data Pipelines) |
| Transient error | storage times out | retry with backoff (see Observability & Operations) |
| Crash | the job is killed at 02:10 | checkpoint after each batch, resume (see Reliable Data Pipelines) |
| Poison batch | one batch always fails | set it aside with its error, keep going |
| Everything else | a bug, a permission error | stop, loudly |

The last row matters as much as the others: a pipeline that catches *every* error hides bugs and writes half a report. And all of it is counted (see Observability & Operations), so the morning report says what happened: rows loaded and rejected, retries used, batches written, batches set aside.

The examples use a small in-memory sink standing in for the report storage. Its writes are idempotent (keyed by trip ID, the Reliable Data Pipelines module), which makes retries and resumed batches safe; its failures are scripted, so you can watch each safeguard act.

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

example-1.py · Quarantine, retry and checkpoint in one run

class FlakySink:
    """Report storage: rows stored by trip_id; the first writes time out."""
    def __init__(self, timeouts):
        self.rows = {}
        self.timeouts = timeouts

    def write(self, batch):
        if self.timeouts:
            self.timeouts -= 1
            raise TimeoutError("storage did not answer")
        for row in batch:
            self.rows[row["trip_id"]] = row

def write_with_retry(sink, batch, sleep, metrics, attempts=3):
    for n in range(attempts):
        try:
            sink.write(batch)
            return
        except TimeoutError:
            if n == attempts - 1:
                raise
            metrics["retries"] += 1
            sleep(2 ** n)

def run(rows, sink, checkpoint, sleep, batch_size=2):
    metrics = {"loaded": 0, "rejected": 0, "retries": 0}
    for start in range(checkpoint["done"], len(rows), batch_size):
        good = []
        for row in rows[start:start + batch_size]:
            try:
                good.append({"trip_id": row["trip_id"], "total": float(row["total_amount"])})
            except ValueError:
                metrics["rejected"] += 1
        write_with_retry(sink, good, sleep, metrics)
        metrics["loaded"] += len(good)
        checkpoint["done"] = start + batch_size
    return metrics

night = [
    {"trip_id": 16126, "total_amount": "64.19"}, {"trip_id": 16127, "total_amount": "n/a"},
    {"trip_id": 16128, "total_amount": "18.7"}, {"trip_id": 16130, "total_amount": "11.1"},
    {"trip_id": 16131, "total_amount": "26.46"},
]
sink, checkpoint, waits = FlakySink(timeouts=2), {"done": 0}, []
print(run(night, sink, checkpoint, waits.append))
print("stored", sorted(sink.rows), "waits", waits, "checkpoint", checkpoint)

Some errors must not be retried or skipped. If the storage refuses writes because of a permission problem, retrying won't help and skipping would publish half a report. The pipeline stops, and the checkpoint shows exactly how far it got:

example-2.py · A permanent error stops the run

class ReadOnlySink:
    def __init__(self):
        self.rows = {}

    def write(self, batch):
        raise PermissionError("report storage is read-only")

checkpoint = {"done": 2}
rows = [{"trip_id": 16128, "total_amount": "18.7"}]
for start in range(0, len(rows)):
    ReadOnlySink().write(rows)
    checkpoint["done"] += 1
print("finished", checkpoint)

A poison batch is the in-between case: a batch the storage rejects every time (say, a row that breaks one of its constraints) while every other batch is fine. Stopping the whole night for it is too strict, retrying it forever is pointless. Set it aside in a dead-letter list with its error, keep going, and report it, so someone fixes those rows in the morning. And when the job is killed and restarted, the checkpoint means finished batches aren't written again.

example-3.py · Set a poison batch aside, then resume after a crash

class Killed(Exception):
    """Stands in for the scheduler killing the process."""

class Sink:
    def __init__(self, reject_ids=()):
        self.rows = {}
        self.reject_ids = set(reject_ids)
        self.writes = 0

    def write(self, batch):
        self.writes += 1
        bad = [row["trip_id"] for row in batch if row["trip_id"] in self.reject_ids]
        if bad:
            raise ValueError(f"constraint failed for trips {bad}")
        for row in batch:
            self.rows[row["trip_id"]] = row

def run(batches, sink, checkpoint, stop_after=None):
    dead_letters = []
    for number in range(checkpoint["done"], len(batches)):
        if stop_after is not None and number == stop_after:
            raise Killed("job killed by the scheduler")
        try:
            sink.write(batches[number])
        except ValueError as err:
            dead_letters.append((number, str(err)))
        checkpoint["done"] = number + 1
    return dead_letters

batches = [[{"trip_id": 16126}, {"trip_id": 16127}], [{"trip_id": 16128}, {"trip_id": 16130}], [{"trip_id": 16131}, {"trip_id": 16132}]]
sink, checkpoint = Sink(reject_ids=[16130]), {"done": 0}
try:
    run(batches, sink, checkpoint, stop_after=2)
except Killed as err:
    print("crash:", err, "| checkpoint", checkpoint)
print("resumed, set aside:", run(batches, sink, checkpoint))
print("stored", sorted(sink.rows), "writes", sink.writes)

When to use it: any unattended job: nightly reports, imports, backfills. Mistakes to avoid: one except Exception around everything (bugs become "bad data"); retrying data errors; skipping storage errors that aren't the batch's fault; checkpointing before the write succeeds; and a report that says "done" without the counts of what was rejected and set aside.

Review

Match each failure to its safeguard: quarantine bad rows, retry transient errors, checkpoint and resume after crashes, set poison batches aside, stop on everything else.
Move the checkpoint only after a successful write; retry only TimeoutError-like failures.
Count every outcome so the morning report says exactly what happened.
Next: the last review before release, gathering all of this into one go/no-go decision.

Practise this