data-pipeline

name	data-pipeline
description	Design, build, or debug data processing pipelines. Use when asked to process a dataset, transform data, build an ETL pipeline, schedule batch jobs, or fix data quality issues.
license	MIT
compatibility	Requires python 3.10+

Overview

Data pipelines fail silently and corrupt downstream systems. Every pipeline must be observable, idempotent, and validated at the boundary.

Process

1. 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?

2. 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  # ISO 8601
       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)  # Never silently drop
       return transform(valid)
   

3. 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.

4. Log progress at meaningful checkpoints. After every major stage (extract, validate, transform, load), log the record count and any failures.

5. 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.

6. Run on the full dataset. Monitor progress. On completion, report: records in, records out, records failed, and time elapsed.

Rationalizations

Excuse	Rebuttal
"I'll add validation later"	Invalid data corrupts your database. Validate at the boundary now.
"Logging slows the pipeline down"	A pipeline that fails without logs requires a full rerun to debug. Log it.
"It worked on the sample"	Test samples are not representative. Always run a full-dataset dry run before writing to the destination.

Verification

• Input and output schemas are defined before any code is written
• Invalid records are logged, not silently dropped
• Pipeline was tested on a 100-record sample before full run
• Final report includes: records in, records out, records failed