CDC Lakehouse Pipeline — E-commerce Orders (Kafka + Spark + DuckDB)

submission/Shabbirsheikh/.gitignore

@@ -0,0 +1,14 @@
+.venv/
+__pycache__/
+*.pyc
+.pytest_cache/
+
+# all pipeline-generated/local runtime state - source db, lake parquet,
+# warehouse db, spark checkpoints. Fully rebuildable from source code +
+# seed_demo_data.py + spark_consumer.py, so none of it belongs in git.
+data/
+spark-warehouse/
+artifacts/
+
+# personal scratch/debug script, not part of the submission
+scripts/data_load.py

submission/Shabbirsheikh/README.md

@@ -0,0 +1,160 @@
+# CDC Lakehouse — E-commerce Orders
+
+Setup and run instructions. For the design rationale, trade-offs, and what's
+been verified, see [`DESIGN.md`](./DESIGN.md) — this file is deliberately
+just the practical steps.
+
+## Prerequisites
+
+- Python 3.12
+- Docker (for Kafka and for Spark — Spark 4.0 needs Java 17, so it runs in
+  a container rather than requiring a specific host Java version)
+- A Kafka broker reachable from this machine (see "Kafka broker" below if
+  you don't already have one running)
+
+## 1. Python environment
+
+```bash
+cd submission/Shabbirsheikh
+python3.12 -m venv .venv
+source .venv/bin/activate
+pip install duckdb==1.5.5 kafka-python==3.0.9 pyspark==4.0.0 pytest==9.1.1
+```
+
+## 2. Kafka broker
+
+If you don't already have one running, a single-broker KRaft-mode instance
+is enough:
+
+```bash
+docker run -d --name broker --network host \
+  -e KAFKA_NODE_ID=1 \
+  -e KAFKA_PROCESS_ROLES=broker,controller \
+  -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 \
+  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://<your-host-ip>:9092 \
+  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
+  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
+  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
+  apache/kafka:latest
+```
+
+Replace `<your-host-ip>` with an address other processes (including the
+Spark container) can actually reach — `localhost` only works if every
+component runs in the same network namespace as the broker.
+
+Create the topic (single partition — see `DESIGN.md`'s CDC Strategy
+section for why):
+
+```bash
+docker exec broker /opt/kafka/bin/kafka-topics.sh \
+  --bootstrap-server localhost:9092 \
+  --create --topic cdc-events-v4 --partitions 1 --replication-factor 1
+```
+
+If you use a different bootstrap address or topic name, update
+`BOOTSTRAP_SERVERS`/`TOPIC` in `pipeline/kafka_producer.py` and
+`KAFKA_BOOTSTRAP`/`TOPIC` in `pipeline/spark_consumer.py` to match.
+
+## 3. Spark image (for the consumer)
+
+The Spark consumer runs inside a Docker image built on `apache/spark:4.0.0`
+with the Kafka connector JARs it needs — `apache/spark`'s default image
+doesn't ship a version of `kafka-clients` that matches what
+`spark-sql-kafka-0-10` expects, so this rebuilds it with matching versions.
+
+```dockerfile
+# Dockerfile
+FROM apache/spark:4.0.0
+
+COPY jars/kafka-clients-4.0.0.jar /opt/spark/jars/
+COPY jars/spark-sql-kafka-0-10_2.13-4.0.0.jar /opt/spark/jars/
+COPY jars/spark-token-provider-kafka-0-10_2.13-4.0.0.jar /opt/spark/jars/
+```
+
+Download the three JARs above (matching versions for Spark 4.0.0 / Scala
+2.13) into a `jars/` folder next to the Dockerfile, then:
+
+```bash
+docker build -t spark-kafka:4.0.0 .
+```
+
+You also need `kafka-clients-3.9.0.jar` and `commons-pool2-2.12.0.jar`
+available on the host to mount over the pre-staged versions at run time
+(see the run command below) — the versions baked into the base image don't
+match what the Kafka connector actually expects.
+
+## 4. Run the tests
+
+```bash
+cd submission/Shabbirsheikh
+python -m pytest -q
+```
+
+21 tests, in-memory, no dependency on Kafka/Spark/any persisted file —
+should complete in a couple of seconds.
+
+## 5. Run the validation scripts
+
+```bash
+python scripts/check_schema_contracts.py    # source schema vs. expected contract
+python scripts/run_data_quality_checks.py   # warehouse + lake vs. system/business rules
+python scripts/validate_catalog.py          # catalog completeness vs. actual datasets
+```
+
+The data-quality and catalog scripts expect `data/warehouse.duckdb` and
+`data/lake_parquet/` to already exist — run step 6 first if this is a
+clean checkout.
+
+## 6. Seed data and run the pipeline end to end
+
+```bash
+# writes to source.duckdb and publishes to Kafka
+python scripts/seed_demo_data.py
+
+# process what's in Kafka into the lake + warehouse (on-demand, default)
+docker run --rm --network host --user "$(id -u):$(id -g)" \
+  -v /etc/passwd:/etc/passwd:ro \
+  -v /path/to/kafka-clients-3.9.0.jar:/opt/spark/jars/kafka-clients-4.0.0.jar:ro \
+  -v /path/to/commons-pool2-2.12.0.jar:/opt/spark/jars/commons-pool2-2.12.0.jar:ro \
+  -v "$(pwd)":/workspace -w /workspace \
+  spark-kafka:4.0.0 /opt/spark/bin/spark-submit --master 'local[*]' pipeline/spark_consumer.py
+```
+
+To run continuously instead (polls Kafka every 5 seconds, stays running —
+stop it with `docker stop` when you're done, it holds a resident JVM
+process):
+
+```bash
+# append --continuous to the spark-submit line above
+```
+
+## 7. Verify a record end to end
+
+```bash
+duckdb data/source.duckdb -c "SELECT * FROM customers WHERE customer_id='c100';"
+duckdb -c "SELECT * FROM read_parquet('data/lake_parquet/*/*.parquet') WHERE primary_key='c100';"
+duckdb data/warehouse.duckdb -c "SELECT * FROM wh_customers WHERE customer_id='c100';"
+```
+
+## Project layout
+
+```
+source/models.py               source schema (DDL, expected-column contract)
+pipeline/source_writer.py      writes to source.duckdb, then publishes to Kafka
+pipeline/kafka_producer.py     Kafka producer / change-event envelope
+pipeline/spark_consumer.py     Kafka -> Parquet lake + DuckDB warehouse
+pipeline/warehouse.py          merge logic (apply_lake_events), time travel (reconstruct_as_of)
+scripts/check_schema_contracts.py   schema-drift detection
+scripts/run_data_quality_checks.py  system + business validation
+scripts/validate_catalog.py         catalog completeness check
+scripts/seed_demo_data.py           consistent demo dataset
+catalog/catalog.json                dataset metadata (lake + warehouse)
+tests/                               21 pytest tests, in-memory
+feature.md, source.md, docs/         supporting documentation (see below)
+```
+
+`DESIGN.md` is the primary design document. `feature.md` maps each of the
+assignment's six requirements to what was actually built; `source.md` has
+the full data model and invariants; `docs/` has one file per component
+(CDC/streaming, warehouse/time-travel, schema/data-quality, catalog) with
+implementation-level detail.

submission/Shabbirsheikh/catalog/catalog.json

@@ -0,0 +1,94 @@
+{
+  "datasets": [
+    {
+      "name": "lake_cdc_events",
+      "layer": "lake",
+      "description": "Append-only log of every change captured from the order-management source (customers, orders, order_items, payments), published to Kafka and written out as Parquet by the Spark Structured Streaming consumer (pipeline/spark_consumer.py). One row per insert/update/delete, keyed by Kafka's own message offset as `sequence`. This is the durable, replayable history the warehouse is rebuilt from - nothing here is ever updated or deleted.",
+      "owner": "data-platform",
+      "consumers": ["data-platform", "analytics", "audit"],
+      "update_cadence": "streaming - applied within one trigger interval of the source change in continuous mode, or on-demand via trigger(availableNow=True)",
+      "storage": "Parquet files under data/lake_parquet/event_date=YYYY-MM-DD/*.parquet, partitioned by day",
+      "schema": {
+        "sequence": "BIGINT - Kafka offset for this event; monotonic because the topic has a single partition",
+        "operation": "VARCHAR - insert | update | delete",
+        "table_name": "VARCHAR - which source table this event came from",
+        "primary_key": "VARCHAR - the source row's PK value",
+        "data": "VARCHAR - full row snapshot at capture time, as a JSON string",
+        "captured_at": "VARCHAR - UTC capture time, ISO-8601 string",
+        "event_date": "DATE - partition column, derived from captured_at"
+      }
+    },
+    {
+      "name": "wh_customers",
+      "layer": "warehouse",
+      "description": "Latest known state of each customer. Rows are never physically removed - a source delete sets _deleted instead, so history stays intact for restore/audit.",
+      "owner": "data-platform",
+      "consumers": ["analytics", "product"],
+      "update_cadence": "near real-time, derived from lake_cdc_events",
+      "schema": {
+        "customer_id": "VARCHAR - primary key",
+        "name": "VARCHAR",
+        "email": "VARCHAR",
+        "status": "VARCHAR - active | suspended | closed",
+        "created_at": "TIMESTAMP",
+        "updated_at": "TIMESTAMP",
+        "_cdc_seq": "BIGINT - sequence of the last applied change",
+        "_deleted": "BOOLEAN - true if the source row was deleted"
+      }
+    },
+    {
+      "name": "wh_orders",
+      "layer": "warehouse",
+      "description": "Latest known state of each order. total_amount is expected to reconcile with the sum of that order's wh_order_items - checked in scripts/run_data_quality_checks.py, not enforced by the table itself.",
+      "owner": "data-platform",
+      "consumers": ["analytics", "product", "finance"],
+      "update_cadence": "near real-time, derived from lake_cdc_events",
+      "schema": {
+        "order_id": "VARCHAR - primary key",
+        "customer_id": "VARCHAR - references wh_customers",
+        "status": "VARCHAR - pending | paid | shipped | cancelled | refunded",
+        "total_amount": "DECIMAL(18,2)",
+        "discount_code": "VARCHAR - nullable",
+        "created_at": "TIMESTAMP",
+        "updated_at": "TIMESTAMP",
+        "_cdc_seq": "BIGINT",
+        "_deleted": "BOOLEAN"
+      }
+    },
+    {
+      "name": "wh_order_items",
+      "layer": "warehouse",
+      "description": "Latest known state of each order line item. Has no lifecycle of its own - always tied to a parent order in wh_orders.",
+      "owner": "data-platform",
+      "consumers": ["analytics", "finance"],
+      "update_cadence": "near real-time, derived from lake_cdc_events",
+      "schema": {
+        "order_item_id": "VARCHAR - primary key",
+        "order_id": "VARCHAR - references wh_orders",
+        "sku": "VARCHAR",
+        "quantity": "INTEGER - > 0",
+        "unit_price": "DECIMAL(18,2) - >= 0",
+        "_cdc_seq": "BIGINT",
+        "_deleted": "BOOLEAN"
+      }
+    },
+    {
+      "name": "wh_payments",
+      "layer": "warehouse",
+      "description": "Latest known state of each payment attempt against an order. A single order can have more than one payment row (retries).",
+      "owner": "data-platform",
+      "consumers": ["analytics", "finance", "audit"],
+      "update_cadence": "near real-time, derived from lake_cdc_events",
+      "schema": {
+        "payment_id": "VARCHAR - primary key",
+        "order_id": "VARCHAR - references wh_orders",
+        "amount": "DECIMAL(18,2) - > 0",
+        "status": "VARCHAR - pending | settled | failed | refunded",
+        "created_at": "TIMESTAMP",
+        "paid_at": "TIMESTAMP - nullable until settled, must be >= created_at",
+        "_cdc_seq": "BIGINT",
+        "_deleted": "BOOLEAN"
+      }
+    }
+  ]
+}

submission/Shabbirsheikh/docs/catalog.md

@@ -0,0 +1,61 @@
+# Catalog
+
+Covers: `catalog/catalog.json`, `scripts/validate_catalog.py`.
+
+## What's registered
+
+`catalog.json` has 5 entries — the lake dataset and the four warehouse
+tables. There is no source/raw-layer entry: the source database itself is
+the transactional system of record, not a published, consumer-facing
+dataset.
+
+| Dataset | Layer |
+|---|---|
+| `lake_cdc_events` | lake |
+| `wh_customers` | warehouse |
+| `wh_orders` | warehouse |
+| `wh_order_items` | warehouse |
+| `wh_payments` | warehouse |
+
+## What each entry documents
+
+- `owner` — the team responsible for the dataset (`data-platform` for all
+  of them here).
+- `consumers` — which downstream teams are expected to use it, and this
+  differs per table: `wh_payments` lists `finance` and `audit` as
+  consumers, `wh_order_items` doesn't, reflecting who actually needs that
+  data.
+- `update_cadence` — how fresh the data is expected to be (e.g.
+  "streaming — applied within one trigger interval of the source change
+  in continuous mode, or on-demand").
+- `description` — what the dataset is, in plain language.
+- `schema` — every column, with type and a one-line meaning, including
+  the two warehouse-only bookkeeping columns (`_cdc_seq`, `_deleted`).
+
+`lake_cdc_events`'s entry specifically documents that it's Parquet,
+day-partitioned under `data/lake_parquet/event_date=YYYY-MM-DD/`, and that
+`sequence` is Kafka's own message offset — someone reading the catalog
+shouldn't need to read the pipeline code to understand what they're
+querying.
+
+## Enforcement, not just documentation
+
+`validate_catalog.py`:
+
+- `expected_dataset_names()` — the lake dataset name plus `wh_<table>` for
+  every table in `source/models.py`'s `EXPECTED_COLUMNS`. Any dataset the
+  pipeline actually produces that's missing from `catalog.json` is a
+  failure.
+- Every entry must have all required fields (`name`, `layer`,
+  `description`, `owner`, `consumers`, `update_cadence`, `schema`) —
+  non-empty.
+- For every warehouse table, the catalog's documented schema must include
+  every column `source/models.py` actually expects, plus `_cdc_seq` and
+  `_deleted`. This specifically catches the catalog going stale after a
+  schema change without anyone remembering to update `catalog.json` to
+  match — a documentation file that can silently drift is worse than no
+  documentation, so this is checked automatically rather than trusted.
+
+`enforce()` raises `CatalogValidationError` listing every problem;
+`__main__` catches it and exits non-zero, same pattern as the other two
+validation scripts.