🐍 Python for Data Engineers · Chapter 10

Python: validating events at the gate

Every pipeline has a front door. Validation is the lock — and where the rejected records go matters more than the check itself.

Study this lesson free in the app →

The basics

Validation is the check that runs on every incoming record before your pipeline trusts it: are the required fields there, are the types right, does the value make sense?

The problem it solves: producers break. A mobile release sends event_time in a new format, a retry delivers the same ride twice, a bug sends fare: null. If nothing stops these at the door, they flow into tables and dashboards, and someone prices surge in Bengaluru off a number that was never true.

The mental model: a bouncer with a notebook. Every record either walks in, or is turned away and written down with the reason. The notebook — the dead letter queue — is what lets you find the broken producer and replay the records once it is fixed.

Explained

Bad data does not announce itself. It arrives as fare: null, event_time: "31/07/2026", a ride_completed with no ride_id, and a duplicate from a retry.

If you do not stop it at the door, it reaches a dashboard, and someone makes a pricing decision on it.

The nightclub bouncer

Three bouncers, three different clubs.

Bouncer A turns away anyone suspicious and forgets them. The club is clean, but nobody can explain why Tuesday's crowd was small.

Bouncer B lets everyone in to avoid confrontation. Chaos inside, discovered at 3 AM.

Bouncer C turns people away and writes down who, when, and why. In the morning there is a list — and it turns out 400 people were rejected because one door scanner had the wrong date format.

Be Bouncer C. The rejection log is worth more than the rejection.

The shape to write

from dataclasses import dataclass
from datetime import datetime, timezone

REQUIRED = ("event_id", "ride_id", "event_type", "event_time")

@dataclass
class Rejection:
    event: dict
    reason: str

def validate(event: dict) -> Rejection | None:
    """Return None if the event is good, else why it was rejected."""
    for field in REQUIRED:
        if not event.get(field):
            return Rejection(event, f"missing_{field}")

    try:
        ts = datetime.fromisoformat(event["event_time"].replace("Z", "+00:00"))
    except (ValueError, AttributeError):
        return Rejection(event, "bad_event_time")

    if ts > datetime.now(timezone.utc):
        return Rejection(event, "future_timestamp")   # client clock skew

    if event["event_type"] == "ride_completed":
        fare = event.get("fare")
        if not isinstance(fare, (int, float)) or fare <= 0:
            return Rejection(event, "invalid_fare")

    return None

Note what it returns: not True/False, but the reason. A boolean throws away the only thing you will want at 3 AM.

Validate, dedupe, aggregate — in that order

def total_completed_fare(events: list[dict]) -> tuple[float, list[Rejection]]:
    total, seen, rejected = 0.0, set(), []

    for e in events:
        if (r := validate(e)) is not None:
            rejected.append(r)                  # → dead letter queue
            continue
        if e["event_id"] in seen:
            continue                            # retry duplicate, silently fine
        seen.add(e["event_id"])
        if e["event_type"] == "ride_completed":
            total += e["fare"]

    return round(total, 2), rejected

The order is not cosmetic. Dedupe before you aggregate, or one Kafka retry inflates revenue. Validate before you dedupe, or a malformed event without an event_id poisons the seen set.

Where this goes in production

Hand-rolled checks are fine to learn on; production usually reaches for declarative validation:

• Pydantic — typed models with coercion and clear errors, at the service boundary
• JSON Schema / Avro / Protobuf + a schema registry — the producer's contract is enforced before the event is even accepted
• Great Expectations / dbt tests / Soda — table-level assertions after landing: not-null, uniqueness, accepted values, row-count and freshness thresholds
• Delta/Iceberg constraints — reject bad rows at write time

And the rule underneath all of them: validate as early as the data enters your control, assert again where it is consumed. Cheap at the gate, expensive in the mart.

One bad record must not stop the line

The single most important design decision in a consumer: what happens on a record you cannot process.

• Crash — the partition stalls, lag climbs, the on-call phone rings. Correct only if the failure is systemic (the DB is down)
• Skip silently — data disappears with no evidence. Never acceptable
• Dead letter queue — park the record with its reason, timestamp and offset; keep consuming; alert on DLQ rate

The DLQ is not a graveyard. It needs an owner, a dashboard, and a replay path — fix the parser, re-drive the DLQ through the same idempotent consumer, and the data lands correctly. Without replay you have merely built a nicer way to lose data.

Validate → dedupe → aggregate. In that order, every time.

Reject with a reason, not a boolean.

A DLQ without a replay path is just a slower delete.

Gotchas

• if not event.get("fare") rejects a legitimate 0.0 — use is None when zero is valid
• float for money accumulates rounding error; use Decimal for anything that reconciles against a bank
• Naive datetime objects (no timezone) compared against aware ones raise TypeError — always parse to UTC-aware
• A bare except: swallows KeyboardInterrupt and hides real bugs; catch the specific exception
• Mutating the input dict during validation — the raw event should reach the DLQ exactly as it arrived

Words to know

In two minutes

• Validation is the front door: every record is checked for required fields, types, formats and sane ranges before it is trusted.
• A validator returns a reason (missing_ride_id, bad_event_time), never a bare boolean — the reason is what you route, count and debug on.
• Process in a fixed order: validate → dedupe → aggregate. Validate first so junk never enters the dedup set; dedupe before summing so retries cannot inflate revenue.
• A bad record must not stop the line: park it in a dead letter queue with its reason and offset, keep consuming, alert on the DLQ rate.
• A DLQ needs an owner and a replay path through the same idempotent consumer, or it is just a slower delete.
• Classic Python traps: if not fare rejects a valid 0.0, naive vs aware datetimes raise TypeError, float drifts for money.
• In production use declarative tools — Pydantic or a schema registry at the boundary, dbt/Great Expectations tests after landing.

Test yourself

There are 9 questions, 3 puzzles and 5 recall cards for this lesson in the app — the recall cards come back on later days, just before you would forget.