Data pipelines fail silently and corrupt downstream systems. Every pipeline must be observable, idempotent, and validated at the boundary.
-
Define the contract. Before writing any transformation code, specify:
- Input schema: What fields, types, and constraints does the data arrive with?
- Output schema: What fields, types, and constraints must the output satisfy?
- Volume: How many records? Per-run? Per-day?
- Frequency: One-time, scheduled, or event-driven?
-
Validate at the boundary. The first thing any pipeline stage does is validate its input:
from pydantic import BaseModel, ValidationError
class InputRecord(BaseModel):
user_id: int
event_type: str
timestamp: str
value: float | None = None
def process(raw_records: list[dict]) -> list[dict]:
valid, invalid = [], []
for r in raw_records:
try:
valid.append(InputRecord(**r).model_dump())
except ValidationError as e:
invalid.append({"record": r, "error": str(e)})
if invalid:
log_invalid_records(invalid)
return transform(valid)
-
Make it idempotent. Running the pipeline twice on the same input must produce the same output. Use upserts, not inserts. Use deterministic IDs based on input content, not auto-increment.
-
Log progress at meaningful checkpoints. After every major stage (extract, validate, transform, load), log the record count and any failures.
-
Test with a sample. Before running on the full dataset, run on 100 records. Confirm the output schema, record count, and that no records were silently dropped.
-
Run on the full dataset. Monitor progress. On completion, report: records in, records out, records failed, and time elapsed.
