FibbieBanks — PySpark ETL pipeline
& star schema warehouse

Five stages, each one its own module.

The pipeline is composed of five Python modules under an etl/ package. Each stage is independently runnable for debugging (python -m etl.clean, python -m etl.transform, etc.), and each can be imported by the next via a clean main() entry point that returns its outputs plus the shared Spark session.

The explore module is its own step rather than a notebook afterthought. Knowing your data is a first-class engineering concern: profiling row counts, null distributions, and distinct values catches problems before they propagate downstream. Most pipelines bury this in commented-out notebook cells. Here it has a dedicated module with its own section() banners in the log output.

Composable, not monolithic

Each module imports the previous one's main() function and chains forward. Running python -m etl.load walks the entire pipeline: extract → clean → transform → load, because each stage's main() calls into the previous one. This means every stage is independently testable and every intermediate output is available without re-reading the source CSV.

def main():
    df_cleaned, spark = clean_main()

    # Build the date dimension first — others depend on it
    dim_date_df = dim_date(df_cleaned)

    # Build the surrogate-keyed dimensions in parallel
    dim_cust  = dim_customer(df_cleaned)
    dim_trans = dim_transaction(df_cleaned, dim_date_df)
    dim_emp   = dim_employee(df_cleaned)

    # Fact joins back to all four dims on natural keys
    fact = fact_table(df_cleaned, dim_cust, dim_trans, dim_emp)

    return {
        "dim_date": dim_date_df,
        "dim_customer": dim_cust,
        "dim_transaction": dim_trans,
        "dim_employee": dim_emp,
        "fact_transactions": fact
    }, spark

Validation at every stage. Type discipline end-to-end.

A warehouse is only as trustworthy as the discipline that built it. Three engineering patterns run through every stage of FibbieBanks.

1. Schema validation as a contract

Every transformation in transform.py calls validate_schema() before and after it runs. If a column is missing, the pipeline halts with a ValueError listing exactly which columns weren't found. This means a schema drift in the source CSV gets caught at the start of transform, not three stages later when something tries to load NULLs into a NOT NULL column.

def validate_schema(df, expected_columns: list[str]) -> None:
    missing_cols = [c for c in expected_columns if c not in df.columns]
    if missing_cols:
        logging.error(f"Missing columns: {missing_cols}")
        raise ValueError(f"Missing required columns: {missing_cols}")
    logging.info("✅ Schema validation passed.")

2. Money stays money

The source CSV's amount column arrives as a string (with "$1,250.50" formatting in places). In clean.py it's stripped of non-numeric characters via regex and cast to Spark's DecimalType(18,2). This type flows through every subsequent stage unchanged and lands in PostgreSQL as NUMERIC(18,2). The cast logs how many rows became NULL during the conversion. Silent data corruption is impossible.

3. Observability built in

A custom logger.py module provides a @timed decorator that wraps every pipeline stage with execution timing, and a section() utility that prints visual banners between stages. Every cleaning operation logs before/after row counts. Every load operation logs how many rows were appended (or that none were, on idempotent re-runs). The whole story of any run is in a single pipeline.log file.

2026-05-17 15:47:30 INFO 🔷 Creating dim_date table
2026-05-17 15:47:31 INFO ✅ dim_date table created.
2026-05-17 15:47:31 INFO 🔷 Creating dim_transaction table
2026-05-17 15:47:31 INFO ✅ Schema validation passed.
2026-05-17 15:47:36 INFO ✅ dim_transaction table created — amount is NUMERIC.
2026-05-17 15:47:36 INFO ⏱️ Step completed in 2m 6s
2026-05-17 15:47:42 INFO 🔷 Loading dimension & fact tables — INCREMENTAL MODE
2026-05-17 15:47:54 INFO ℹ️ No new rows to insert — analytics.dim_date is already up to date
2026-05-17 15:48:44 INFO ℹ️ No new rows to insert — analytics.dim_customer is already up to date
2026-05-17 15:49:50 INFO ℹ️ No new rows to insert — analytics.dim_transaction is already up to date
2026-05-17 15:50:26 INFO ℹ️ No new rows to insert — analytics.dim_employee is already up to date
2026-05-17 15:54:26 INFO ℹ️ No new rows to insert — analytics.fact_transactions is already up to date
2026-05-17 15:54:27 INFO 🎉 ALL TABLES LOADED SUCCESSFULLY — ONLY NEW DATA APPENDED

That sequence of "No new rows to insert" lines (one per table) is the visible proof that idempotency is working. The deterministic hash keys mean the same source data produces the same surrogate keys, and the LEFT JOIN merge correctly identifies that nothing is actually new.

What this pipeline produces.

Measurable outcomes from end-to-end runs on the 1M-row source dataset.