Validating ETL Payloads with Pydantic

A Pydantic v2 model placed in front of client.insert turns ClickHouse’s unforgiving columnar typing into a per-record decision you control in Python: valid rows coerce cleanly to the destination types, invalid rows raise a structured ValidationError you can route to a dead-letter sink, and a malformed payload never aborts a whole block on the server. This guide builds that validation boundary — mapping Python types to ClickHouse types, handling coercion and validation failures, dead-lettering rejects, and keeping batch validation fast — as the Python-ETL companion to the broader schema validation and evolution control plane.

Prerequisites

• ClickHouse 23.8+ with a destination MergeTree table whose schema you own
• Python 3.9+ with pydantic>=2.5 and clickhouse-connect>=0.7
• A dead-letter destination — a separate ClickHouse table, a Kafka topic, or object storage
• The destination table’s exact column names, types, and nullability (from DESCRIBE TABLE)
• INSERT privilege on both the main and dead-letter tables
• A decision on strictness: reject-on-first-error vs. collect-all-errors per record

Where the validation boundary sits

Validation belongs before serialization, not after the insert fails. The model is the single choke point every record passes through: raw dicts (from a Kafka consumer, an API poll, a CSV reader) go in, and either a typed model instance comes out — ready to become a tuple for bulk insertion with clickhouse-connect — or a ValidationError diverts the record to a dead-letter path. Because the check runs in Python, a single bad record costs one record, not the block; the server only ever sees rows that already match its column types.

Step 1 — Model the ClickHouse schema as a Pydantic type

Every ClickHouse column maps to a Python type that Pydantic can validate and coerce. Get this mapping right and the model output is a drop-in tuple for client.insert; get it wrong and the server rejects the block.

from datetime import datetime
from typing import Literal, Optional
from uuid import UUID
from pydantic import BaseModel, Field, ConfigDict

class EventRow(BaseModel):
    model_config = ConfigDict(extra="forbid")  # reject unexpected fields loudly

    event_id:   UUID
    event_ts:   datetime
    user_id:    int = Field(ge=0)                       # UInt64 → non-negative
    event_type: Literal["view", "checkout", "signup"]   # LowCardinality guard
    revenue:    Optional[float] = None                  # Nullable(Float64)
    tags:       list[str] = Field(default_factory=list) # Array(String)

    def to_row(self) -> tuple:
        # Column order MUST match the client.insert column_names list.
        return (self.event_id, self.event_ts, self.user_id,
                self.event_type, self.revenue, self.tags)

extra="forbid" is the schema-drift tripwire: when an upstream producer adds an unannounced field, validation fails loudly here instead of silently dropping data — exactly the coordination failure the schema validation and evolution control plane exists to catch.

Step 2 — Coerce and catch errors per record

Pydantic v2 coerces compatible inputs (an ISO timestamp string to datetime, a numeric string to int) and raises a structured ValidationError — with the offending field, input value, and reason — when it cannot. Use model_validate and handle the error at the record boundary.

from pydantic import ValidationError

def validate_record(raw: dict) -> tuple[EventRow | None, dict | None]:
    try:
        return EventRow.model_validate(raw), None
    except ValidationError as exc:
        # exc.errors() is a list of dicts: loc, msg, type, input — ideal for a DLQ.
        return None, {"raw": raw, "errors": exc.errors(include_url=False)}

Decide strictness deliberately. The default collects all errors in a record before raising, which gives a complete rejection report; if you only need the first failure and want to shave latency on a hot path, that trade-off is discussed in Step 5.

Step 3 — Route rejects to a dead-letter table

A rejected record is data you cannot afford to lose — it is the evidence for the producer conversation. Persist it with enough context to replay: the raw payload, the structured error, and a timestamp. A ClickHouse dead-letter table keeps rejects queryable alongside the pipeline.

CREATE TABLE analytics.events_dlq
(
    `received_at` DateTime64(3) DEFAULT now64(3),
    `source`      LowCardinality(String),
    `raw`         String,          -- original payload as JSON text
    `errors`      String           -- serialized Pydantic error list
)
ENGINE = MergeTree
PARTITION BY toYYYYMMDD(received_at)
ORDER BY (source, received_at);

import json
import clickhouse_connect

client = clickhouse_connect.get_client(host="clickhouse", database="analytics")

def dead_letter(rejects: list[dict], source: str) -> None:
    if not rejects:
        return
    client.insert(
        "events_dlq",
        [(source, json.dumps(r["raw"], default=str), json.dumps(r["errors"], default=str))
         for r in rejects],
        column_names=["source", "raw", "errors"],
    )

Alarming on events_dlq row growth turns silent schema drift into a paged alert. A sustained spike almost always means a producer changed a field type or added a value outside a Literal set.

Step 4 — Validate a batch, then bulk-insert the survivors

Wire the pieces together: validate every record in an incoming batch, insert the valid rows in one bulk call, and dead-letter the rest. Partitioning happens once per batch, so the insert stays a single merge-friendly block.

def process_batch(raws: list[dict], source: str) -> tuple[int, int]:
    valid_rows, rejects = [], []
    for raw in raws:
        row, err = validate_record(raw)
        (valid_rows.append(row.to_row()) if row else rejects.append(err))

    if valid_rows:
        client.insert(
            "events",
            valid_rows,
            column_names=["event_id", "event_ts", "user_id",
                          "event_type", "revenue", "tags"],
        )
    dead_letter(rejects, source)
    return len(valid_rows), len(rejects)

One block for the good rows, one for the rejects — a poisoned record can never take down the batch, which is the whole point of validating before the server.

Step 5 — Keep batch validation fast

Pydantic v2’s core is compiled in Rust, so per-record validation is cheap, but you can still shave overhead on high-volume streams. Build the validator once, prefer model_validate over constructing from kwargs, and use TypeAdapter to validate a whole list in one call when you do not need per-record error isolation.

from pydantic import TypeAdapter

BatchAdapter = TypeAdapter(list[EventRow])  # build once, reuse across batches

def validate_fast(raws: list[dict]) -> list[EventRow]:
    # All-or-nothing: fastest path when upstream is already trusted/clean.
    return BatchAdapter.validate_python(raws)

TypeAdapter.validate_python on a list is the fastest option but fails the entire batch on the first bad record, so reserve it for trusted sources; keep the per-record loop from Step 4 where dead-lettering matters. Constructing the adapter or model once (not per record) is the single biggest win — repeated schema construction dominates the cost otherwise.

Verification

Confirm valid rows landed and rejects were captured by comparing the two tables over the same window:

SELECT
    (SELECT count() FROM analytics.events
       WHERE event_ts > now() - INTERVAL 15 MINUTE)      AS accepted,
    (SELECT count() FROM analytics.events_dlq
       WHERE received_at > now() - INTERVAL 15 MINUTE)    AS rejected;

A rejection rate that is non-zero but stable is healthy — it means the guard is working. Inspect why records are failing by aggregating the stored error types:

SELECT
    JSONExtractString(arrayJoin(JSONExtractArrayRaw(errors)), 'type') AS error_type,
    count() AS n
FROM analytics.events_dlq
WHERE received_at > now() - INTERVAL 1 DAY
GROUP BY error_type
ORDER BY n DESC;

A sudden new error_type — say literal_error on event_type — is your earliest signal that a producer shipped a value your Literal set does not know about.

Gotchas and edge cases

• Pydantic coercion can hide real problems. By default, int accepts the string "42" and even 42.0. If an upstream sending floats-as-ints signals a schema mismatch you care about, set model_config = ConfigDict(strict=True) (or Field(strict=True) per field) so coercion does not paper over the drift.
• datetime timezone handling bites at the ClickHouse boundary. A tz-aware datetime validated in Python is stored by ClickHouse in the column’s timezone; a naive one is assumed to already be in it. Normalize to UTC in the model (@field_validator) so DateTime64(3) values are unambiguous.
• extra="forbid" vs. extra="ignore" is a schema-evolution policy, not a style choice. forbid catches new upstream fields immediately (good for tight contracts); ignore tolerates additive changes silently (good for loose ones). Pick per stream, and pair forbid with the migration workflow in running ALTER TABLE migrations without downtime so the column exists before you start requiring it.
• A None for a non-Nullable column passes Pydantic but fails ClickHouse. If a field is Optional in the model but the column is not Nullable(T), the model accepts None and the insert then dies with Cannot insert NULL. Keep model nullability and column nullability identical, and validate that alignment in a test, not in production.

Related

• Schema Validation & Evolution — the parent control plane this validation boundary implements.
• Running ALTER TABLE Migrations Without Downtime — evolving the destination schema the model must track.
• Implementing Avro Schema Registry Validation in Python — the wire-format validator that pairs with these semantic checks.
• Bulk Inserting with clickhouse-connect — the insert path the validated rows flow into.
• Batch Insert Optimization — sizing the validated batches into merge-friendly parts.

Up: Schema Validation & Evolution