A Pandas Data-Cleaning Pipeline That Stays Debuggable

Cleaning a real dataset with pandas almost always means the same handful of operations in some order: standardize column names, fix data types, handle missing values, resolve duplicates, and validate the result before it goes anywhere else. Writing all of it as one dense block of chained method calls is tempting, because pandas makes chaining easy — but the same density that makes the code short also makes it nearly impossible to tell, after the fact, which line changed the row count or introduced a bad value.

Start with the shape of the problem, not the code

Before writing any cleaning logic, it is worth actually looking at what is wrong. df.info() shows dtypes and null counts per column in one call; df.describe(include="all") surfaces obviously wrong values — a "weight" column with a maximum of 4,200,000, for instance, which is either a unit mistake or a data-entry error, not a real freight load. Profiling first turns "clean the data" into a specific, checkable list: these three columns have the wrong dtype, this one has nulls that need a decision, these two have inconsistent casing. That list is what the pipeline should actually be built to fix — not a generic cleaning routine applied on faith.

Structuring the pipeline as named, single-purpose steps

Instead of one chain, each transformation becomes its own function that takes a dataframe and returns one. The pipeline itself becomes a short, readable list of steps:

import pandas as pd

def normalize_columns(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df.columns = df.columns.str.strip().str.lower().str.replace(" ", "_")
    return df

def fix_dtypes(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df["pickup_date"] = pd.to_datetime(df["pickup_date"], errors="coerce")
    df["weight_lbs"] = pd.to_numeric(df["weight_lbs"], errors="coerce")
    return df

def handle_missing(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df = df.dropna(subset=["load_id"])
    df["weight_lbs"] = df["weight_lbs"].fillna(df["weight_lbs"].median())
    return df

def deduplicate(df: pd.DataFrame) -> pd.DataFrame:
    return df.drop_duplicates(subset=["load_id"], keep="last")

def run_pipeline(df: pd.DataFrame) -> pd.DataFrame:
    steps = [normalize_columns, fix_dtypes, handle_missing, deduplicate]
    for step in steps:
        df = step(df)
    return df

Each function does one thing and can be tested in isolation with a small, deliberately messy input — which is exactly the shape of test that pytest fixtures and parametrize are built for: one fixture representing a realistic messy dataframe, and a parametrized case per edge value a given step needs to handle correctly. The .copy() at the top of each function is not cosmetic — pandas' chained-assignment warnings exist precisely because it is easy to mutate a view of the original dataframe without meaning to, and a cleaning step that silently mutates its input makes the pipeline's behavior depend on call order in ways that are hard to reason about later.

Making failures visible between steps, not just at the end

A pipeline that only reports its final row count hides exactly the information needed to debug it: which step dropped rows, and how many. A small logging wrapper around each step turns that into something visible without changing any step's logic:

import logging

logger = logging.getLogger("pipeline")

def log_step(step):
    def wrapped(df):
        before = len(df)
        result = step(df)
        after = len(result)
        logger.info(
            "%s: %d -> %d rows (%+d)",
            step.__name__, before, after, after - before,
        )
        return result
    return wrapped

def run_pipeline(df: pd.DataFrame) -> pd.DataFrame:
    steps = [normalize_columns, fix_dtypes, handle_missing, deduplicate]
    for step in steps:
        df = log_step(step)(df)
    return df

This is a small application of the same idea covered separately in structured logging in Python: a log line with named fields (step, before, after) is something a script can later parse to answer "did deduplication drop more rows than expected today," rather than a person needing to notice a suspicious number by eye during a run.

Deciding what to do with bad rows, on purpose

Every cleaning pipeline eventually meets a row that cannot be fixed automatically — a load with no ID, a weight that parsed to NaN and has no reasonable default. There are three honest options, and silently picking one without deciding is how bad data reaches production: drop the row and count how many were dropped, fill it with a documented default and flag that the value was imputed, or route it to a separate "needs review" output instead of either cleaning or discarding it. Which option is right depends entirely on what happens downstream — a missing load ID is probably unrecoverable and should be dropped and logged, while a missing weight might be reasonably filled with a placeholder as long as anything reading the cleaned data later knows that field can be an estimate rather than a measurement:

def handle_missing(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    missing_ids = df["load_id"].isna().sum()
    if missing_ids:
        logger.warning("dropping %d rows with no load_id", missing_ids)
    df = df.dropna(subset=["load_id"])

    df["weight_estimated"] = df["weight_lbs"].isna()
    df["weight_lbs"] = df["weight_lbs"].fillna(df["weight_lbs"].median())
    return df

Adding the weight_estimated flag column is a small change that keeps the imputation honest — anything consuming weight_lbs later can filter out or separately account for estimated values instead of treating every row as equally measured.

Validating the output, not just the input

Profiling at the start catches problems in the raw data; an explicit check at the end catches problems the pipeline itself introduced. A short set of assertions after the pipeline runs is cheap insurance against a future edit silently breaking an invariant the rest of the system depends on:

def validate(df: pd.DataFrame) -> pd.DataFrame:
    assert df["load_id"].is_unique, "duplicate load_id survived deduplication"
    assert df["weight_lbs"].notna().all(), "unfilled weight_lbs after cleaning"
    assert (df["weight_lbs"] > 0).all(), "non-positive weight_lbs after cleaning"
    return df

Placing validate as the last step in the same steps list used for the rest of the pipeline means it runs, and fails loudly, every time — rather than being a notebook cell someone remembers to run manually before an important export and forgets the rest of the time.

A short checklist

Is each cleaning operation its own named function, testable with a small representative input?
Does each step return a new dataframe rather than mutating its input in place?
Is the row count logged before and after each step, so a step that drops more rows than expected is visible immediately?
For every category of bad row, was drop-and-count, fill-and-flag, or route-to-review chosen on purpose, rather than left to whatever a default pandas method happens to do?
Does an explicit validation step run after cleaning and fail loudly if an invariant does not hold?

This is the same shape of problem as designing a freight intake schema for messages that were never structured to begin with — both are about deciding, deliberately and visibly, what happens to data that does not arrive in the shape the rest of the system expects.