Python for Data Engineering: From CSV to a Clean Pipeline

Most data work starts the same way: someone hands you a CSV and asks for "just a quick summary." The quick script grows, breaks on the next file, and quietly corrupts a number that nobody notices for a month. A little structure turns that throwaway script into a pipeline you can trust.

The naive read is a trap

Reading a CSV straight into a dictionary feels fine until reality arrives: a missing column, a price written as "1,299.00", an empty string where a date should be. The script does not crash, it produces wrong numbers, which is worse.

• Types are inferred per chunk, so the same column can be text in one file and a float in the next.
• Silent missing values propagate through every downstream aggregation.
• Encoding surprises (UTF-8 versus Latin-1) corrupt text fields without warning.

Validate at the boundary

Treat the edge of your pipeline as a checkpoint: parse each raw row into a typed record and reject anything that does not fit, loudly, before it reaches your logic.

from dataclasses import dataclass
from datetime import date

@dataclass(frozen=True)
class Sale:
    order_id: str
    amount: float
    ordered_on: date

def parse_row(row: dict) -> Sale:
    amount = float(row["amount"].replace(",", ""))
    if amount < 0:
        raise ValueError(f"negative amount for {row['order_id']}")
    return Sale(
        order_id=row["order_id"].strip(),
        amount=amount,
        ordered_on=date.fromisoformat(row["ordered_on"]),
    )

Shape it into stages

A pipeline reads best as a sequence of small, pure functions - extract, transform, load - each testable on its own. Keep the messy I/O at the ends and the logic in the middle.

import csv
import logging
from collections import Counter

def run(path: str) -> Counter:
    totals = Counter()
    with open(path, newline="", encoding="utf-8") as fh:
        for raw in csv.DictReader(fh):
            try:
                sale = parse_row(raw)
            except (KeyError, ValueError) as exc:
                logging.warning("skipping bad row: %s", exc)
                continue
            totals[sale.ordered_on.isoformat()] += sale.amount
    return totals

Make it reproducible

• Pin dependencies with a lock file so the pipeline that ran yesterday runs the same today.
• Make each run idempotent, so re-processing the same input never double-counts.
• Write rejected rows to a dead-letter file instead of dropping them, so you can audit what was skipped.
• Log counts in and out: a run that reads 10,000 rows and writes 4 should trip an alarm.

The point is not any single library, it is the discipline of validating early, keeping stages small, and never trusting a CSV to be clean. That is what turns a script into a pipeline.