CDC Lakehouse Medallion Pipeline Submission - Sagar Bala

[sagarbalaai-code]

Submission Notes: CDC Medallion Lakehouse Pipeline

Hey Team,

Here is my submission for the CDC reliability assignment. I chose to model the E-Commerce domain (customers, orders, line items, and payment attempts) since it resembles the transactional systems I work with on a daily basis.

My Tech Stack & Approach

Since my core skillset is focused on the Databricks and Delta Lakehouse ecosystem (Python, SQL, PySpark, Auto Loader, and Delta Tables), I modeled the pipeline as a classic Medallion Architecture (Bronze → Silver → Gold). This allowed me to leverage Delta's native transaction logs and Change Data Feed (CDF) to build a reliable, incremental replication pipeline without writing heavy, custom state-tracking code.

---

Folder & Notebook Structure

All submission files are organized inside the submission/sagarbalaai-code/ directory:

• source/source_db.sql: The upstream transactional relational DDL schema (with indexes, keys, check constraints) and seed records representing our source data of record.
• pipeline/01_bronze_ingestion.py: Streams raw CDC events, runs schema drift checks, and appends to Bronze.
• pipeline/02_silver_merge.py: Reads the Bronze CDF incrementally and executes a deduplicated Delta MERGE.
• pipeline/03_gold_metrics.py: Combines Silver dimensions and facts to output business aggregates.
• tests/04_data_quality.py: Runs validation checks (system and business rules) to guarantee parity.

---

Core Technical Solutions & Code Design

1. Ingestion & Schema Drift ("Stop-the-Line")

• Location: 01_bronze_ingestion.py
• How it works: Before committing raw events to Bronze, check_schema_compatibility compares incoming batch structures against the target table. If columns were dropped or datatypes were changed incompatibly, the code raises an exception and halts the write. This prevents corrupt data from contaminating the lakehouse.
• Delta CDF: Enabled on the Bronze tables (delta.enableChangeDataFeed = true) to enable incremental downstream syncs.

2. Silver Deduplication & Delta MERGE

• Location: 02_silver_merge.py
• How it works: To prevent Spark's "Multiple source rows matched" MERGE error (when one record has multiple status changes in a single micro-batch), I deduplicate the batch by applying a window function partition on the primary key, ordering by _commit_timestamp DESC.
• Deletes: Hard deletes in the source DB are captured as DELETE operations and merged into Silver:
  whenMatchedDelete(condition="updates._change_type = 'delete'")

3. Data Quality & Validation Parity

• Location: 04_data_quality.py
• System Audits: Enforces PK uniqueness and referential integrity (child FKs mapping back to existing parents) using anti-joins.
• Business Audits: Checks chronology (shipped_at >= created_at) and aggregates order item subtotals to verify they match the order header total_amount.
• State-Machine transitions check: Queries the Bronze Change Data Feed history to trace status updates. It flags violations if a transaction attempted to revert a terminal status (e.g. changing an order from SHIPPED or CANCELLED back to PENDING).

4. Time Travel & Recovery Demo

• Location: 02_silver_merge.py
• How it works: Demonstrates how to query version 0 of the table using .option("versionAsOf", 0) and executes a metadata rollback using deltaTable.restoreToVersion(0).

---

Architecture Tradeoffs

• Halt Ingestion vs. Auto-Evolution: I chose a strict schema validation check ("Stop-the-line") rather than letting Spark automatically merge schemas. The tradeoff is that breaking changes require manual engineer intervention, but this guarantees data quality and prevents silent corruption.

Let me know if you have any questions during the review!