Data cleaning pipelines

Data work beyond the pandas taster from the intro tutorials: cleaning messy inputs, building small pipelines, and keeping transformations testable. You do not need a data warehouse to benefit from pipeline thinking – a folder of CSV exports and a report script counts.

Stages, not one giant script

Split the work into explicit stages: ingest, validate, transform, aggregate, export. Each stage takes structured input and returns structured output. That makes failures easier to locate and stages easier to test in isolation.

from dataclasses import dataclass
from pathlib import Path
import csv

@dataclass
class Sale:
    sku: str
    quantity: int
    unit_price: float

def load_sales(path: Path) -> list[Sale]:
    rows = []
    with path.open(newline="", encoding="utf-8") as handle:
        for row in csv.DictReader(handle):
            rows.append(Sale(
                sku=row["sku"].strip(),
                quantity=int(row["qty"]),
                unit_price=float(row["unit_price"]),
            ))
    return rows

def clean_sales(sales: list[Sale]) -> list[Sale]:
    return [s for s in sales if s.quantity > 0 and s.unit_price >= 0]

def total_revenue(sales: list[Sale]) -> float:
    return sum(s.quantity * s.unit_price for s in sales)

Validation at the boundary

Validate when data enters your system – file read, API payload, spreadsheet import. Fail loudly with a useful message rather than letting bad rows silently become NaN.

def parse_positive_int(value: str, field: str) -> int:
    try:
        number = int(value)
    except ValueError as exc:
        raise ValueError(f"{field} must be an integer, got {value!r}") from exc
    if number < 0:
        raise ValueError(f"{field} must be >= 0")
    return number

pandas when the table grows teeth

For columnar cleaning, joins, and groupbys, pandas still earns its keep. Keep business rules in named functions so the notebook or script does not become an unreadable chain of method calls.

import pandas as pd

def load_frame(path: str) -> pd.DataFrame:
    df = pd.read_csv(path, parse_dates=["sold_at"])
    df["sku"] = df["sku"].str.strip().str.upper()
    df = df.dropna(subset=["sku", "qty"])
    df = df[df["qty"] > 0]
    return df

def monthly_totals(df: pd.DataFrame) -> pd.DataFrame:
    df = df.assign(revenue=df["qty"] * df["unit_price"])
    return (
        df.groupby(df["sold_at"].dt.to_period("M"))["revenue"]
        .sum()
        .reset_index(name="total")
    )

Idempotent outputs

Write outputs to a temp file, then rename into place. If the job crashes halfway, you do not leave a half-written report that looks finished.

import json
import tempfile
from pathlib import Path

def write_json_atomic(path: Path, payload: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile("w", delete=False, dir=path.parent) as tmp:
        json.dump(payload, tmp, indent=2)
        tmp_path = Path(tmp.name)
    tmp_path.replace(path)

Logging and reproducibility

Log row counts at each stage, source file paths, and timestamps. When a stakeholder asks why March’s total changed, you want an audit trail – not a vague memory of editing a notebook at midnight.

When not to use pandas

Small CSV, simple rules, no joins – the standard library and dataclasses may be faster to run and easier to deploy. Pandas adds weight. Reach for it when the table operations genuinely save you time, not because it is the default hammer.