Work02Data engineering

Retail Lakehouse ETL

Schema-enforced PySpark batch warehouse over a million defective sales lines.

Role
Solo — design, engineering, deployment
Language
Python 99.1% · Makefile 0.9%
Timeline
Sep 2026 → Sep 2026
Last push
12 days ago · 2 commits
Source
raw sales lines read
1,009,989
end-to-end (~19.8K rows/s)
50.9 s
duplicates removed
9,993
quarantined, per-rule breakdown
1.30 %

A production-shaped batch ETL: 1M+ semi-structured retail sales lines through explicit schemas, harmonisation, window-function dedupe, an 8-rule quality gate and into a star schema with an SCD2 customer dimension and Spark SQL marts.

Problem

Raw retail feeds arrive as month-partitioned JSON with malformed lines, duplicate sale IDs, null foreign keys and two timestamp formats. A warehouse built on them silently mis-attributes revenue unless bad data is caught, quarantined and reported — and reruns must be byte-identical so retries are safe.

Approach

  1. 01

    Every source is read with an explicit StructType in PERMISSIVE mode; corrupt JSON is captured in _corrupt_record and schema drift is detected on a sample and logged before it becomes nulls.

  2. 02

    Harmonisation normalises casing and timestamps; row_number() with a total ordering removes duplicates deterministically.

  3. 03

    Eight declarative data-quality rules produce a per-rule breakdown and a quarantine table; the run fails only if the quarantine share crosses a threshold.

  4. 04

    Dimensions use SCD Type 2 validity windows and the fact table joins on the version valid at order time; the fact is Hive-partitioned by year/month so Hive, Athena, Synapse or Databricks can query it in place. Marts use LAG, DENSE_RANK and cumulative windows; MongoDB serving loads are idempotent upserts.

Results

raw sales lines read
1,009,989
end-to-end (~19.8K rows/s)
50.9 s
duplicates removed
9,993
quarantined, per-rule breakdown
1.30 %
Created
3 Sep 2026
Commits
2
Default branch
main
License
unlicensed

Stack

  • PySpark
  • Spark SQL
  • Parquet
  • MongoDB
  • Airflow
  • pytest
  • Python 3.12

From the notes