Data Cleaning Workflows

A one-off cleaning script is a liability — you run it once, forget what it does, and can't reproduce the result. A cleaning workflow is a documented, reusable pipeline that takes raw data in and delivers clean data out, with every decision logged.

Learning Objectives

• Structure a data cleaning pipeline as a series of named functions
• Use method chaining for readable Pandas workflows
• Document cleaning decisions as code comments and audit logs
• Validate cleaned data with assertions
• Build a reusable cleaning module

---

The Three-Stage Cleaning Pattern

Every cleaning workflow follows the same structure:

1. AUDIT — understand the raw data before touching it
2. CLEAN — apply transformations in a logical order
3. VALIDATE — verify the result matches expectations

---

Stage 1 — Audit

import pandas as pd
import numpy as np

def audit_dataframe(df, name="DataFrame"):
    """Print a comprehensive data quality report."""
    print(f"\n{'='*60}")
    print(f"AUDIT: {name}")
    print(f"{'='*60}")
    print(f"Shape: {df.shape[0]:,} rows × {df.shape[1]} columns")
    print(f"\nData types:")
    print(df.dtypes.to_string())
    print(f"\nMissing values:")
    missing = df.isnull().sum()
    missing_pct = df.isnull().mean() * 100
    missing_df = pd.DataFrame({"count": missing, "pct": missing_pct.round(1)})
    print(missing_df[missing_df["count"] > 0].to_string())
    print(f"\nDuplicate rows: {df.duplicated().sum():,}")
    print(f"\nNumeric summary:")
    print(df.describe().round(2).to_string())

---

Stage 2 — Clean (Pipeline Pattern)

Build cleaning as a series of functions that each take and return a DataFrame. Chain them with .pipe().

def fix_dtypes(df):
    """Convert columns to correct data types."""
    df = df.copy()
    df["order_date"] = pd.to_datetime(df["order_date"], errors="coerce")
    df["quantity"] = pd.to_numeric(df["quantity"], errors="coerce")
    df["unit_price"] = (
        df["unit_price"]
        .astype(str)
        .str.replace("£", "").str.replace(",", "")
        .pipe(pd.to_numeric, errors="coerce")
    )
    return df

def standardise_strings(df):
    """Normalise all string columns."""
    df = df.copy()
    df["status"] = df["status"].str.lower().str.strip()
    df["country"] = df["country"].str.title().str.strip()
    df["product"] = df["product"].str.strip().str.replace(r"\s+", " ", regex=True)
    return df

def handle_missing_values(df):
    """Apply business rules for missing data."""
    df = df.copy()
    df["discount_pct"] = df["discount_pct"].fillna(0)
    df["delivery_date"] = df["delivery_date"]   # keep as NaN — not yet delivered
    df = df.dropna(subset=["order_id", "customer_id"])   # these are critical
    return df

def remove_duplicates(df):
    """Remove duplicate orders, keeping the most recently updated."""
    df = df.copy()
    if "updated_at" in df.columns:
        df = df.sort_values("updated_at", ascending=False)
    df = df.drop_duplicates(subset=["order_id"], keep="first")
    return df.reset_index(drop=True)

def add_calculated_columns(df):
    """Add derived columns used in all downstream analyses."""
    df = df.copy()
    df["total"] = df["quantity"] * df["unit_price"]
    df["net_total"] = df["total"] * (1 - df["discount_pct"] / 100)
    df["year"] = df["order_date"].dt.year
    df["month"] = df["order_date"].dt.month
    df["quarter"] = df["order_date"].dt.quarter
    df["is_weekend"] = df["order_date"].dt.dayofweek >= 5
    return df

def filter_valid_rows(df):
    """Remove rows that fail business logic validation."""
    df = df.copy()
    before = len(df)
    df = df[df["quantity"] > 0]                    # no zero-quantity orders
    df = df[df["unit_price"] > 0]                  # no zero-price orders
    df = df[df["unit_price"] <= 10000]             # price sanity check
    after = len(df)
    print(f"filter_valid_rows: removed {before - after} invalid rows")
    return df

Chaining with .pipe()

def clean_orders(filepath):
    """Main cleaning pipeline — takes raw CSV path, returns clean DataFrame."""
    return (
        pd.read_csv(filepath)
        .pipe(fix_dtypes)
        .pipe(standardise_strings)
        .pipe(handle_missing_values)
        .pipe(remove_duplicates)
        .pipe(add_calculated_columns)
        .pipe(filter_valid_rows)
    )

# Usage
df_raw = pd.read_csv("orders_raw.csv")
audit_dataframe(df_raw, "Raw Orders")

df_clean = clean_orders("orders_raw.csv")
audit_dataframe(df_clean, "Cleaned Orders")

---

Stage 3 — Validate

After cleaning, assert that the result meets expectations:

def validate_orders(df, name="orders"):
    """Assert data quality constraints are met."""
    errors = []

    if df["order_id"].duplicated().any():
        errors.append("Duplicate order_ids found")

    if df["order_id"].isnull().any():
        errors.append("Null order_ids found")

    if (df["quantity"] <= 0).any():
        errors.append("Non-positive quantities found")

    if (df["unit_price"] <= 0).any():
        errors.append("Non-positive unit prices found")

    if df["order_date"].isnull().any():
        errors.append("Null order dates found")

    if len(errors) == 0:
        print(f"✓ {name}: all validations passed ({len(df):,} rows)")
    else:
        for error in errors:
            print(f"✗ {name}: {error}")
        raise ValueError(f"Data validation failed with {len(errors)} errors")

---

Audit Log — Document What You Did

import datetime

cleaning_log = []

def log_step(step_name, before_rows, after_rows, note=""):
    """Record each cleaning step for reproducibility."""
    entry = {
        "step": step_name,
        "rows_before": before_rows,
        "rows_after": after_rows,
        "rows_removed": before_rows - after_rows,
        "pct_removed": f"{(before_rows - after_rows) / before_rows:.1%}",
        "note": note,
        "timestamp": datetime.datetime.now().isoformat(),
    }
    cleaning_log.append(entry)
    print(f"  {step_name}: {before_rows:,} → {after_rows:,} rows ({entry['rows_removed']:,} removed)")

# Usage inside a cleaning function
def remove_duplicates_logged(df):
    before = len(df)
    df = df.drop_duplicates(subset=["order_id"], keep="first").reset_index(drop=True)
    log_step("remove_duplicates", before, len(df), "kept most recent version")
    return df

---

Practice Exercises

Warm-up 1. Write a function fix_dtypes(df) that converts: order_date to datetime, quantity to int, unit_price to float (removing £ signs and commas). 2. Chain fix_dtypes, standardise_strings, and handle_missing_values using .pipe(). 3. Write a validate(df) function that checks: no duplicate order_id, no null customer_id, all quantity > 0.

Main 4. Build a complete cleaning pipeline for a sales CSV. Log the row count before and after each step. 5. Add a cleaning_summary() function that prints a before/after comparison: row count, missing values, duplicate count. 6. Add a validation step that raises an error if any assertion fails, and returns the clean DataFrame if all pass.

Stretch 7. Write a profile_data(df) function that returns a full HTML report (using df.describe(), df.dtypes, missing values, sample values per column) — something you'd share with a stakeholder. 8. Turn your cleaning pipeline into a class DataCleaner with methods for each cleaning step, a run() method that chains them all, and a report() method that prints the audit log.

---

← Data Transformation · Next: Case Study →