From dfbd96528b98035ca89481447618bbf2cebb1e25 Mon Sep 17 00:00:00 2001 From: Shabbir Sheikh Date: Fri, 24 Jul 2026 20:30:49 +0530 Subject: [PATCH 1/2] Initial approach doc for CDC lakehouse assignment --- submission/Shabbirsheikh/DESIGN.md | 36 ++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) create mode 100644 submission/Shabbirsheikh/DESIGN.md diff --git a/submission/Shabbirsheikh/DESIGN.md b/submission/Shabbirsheikh/DESIGN.md new file mode 100644 index 0000000..b783fd7 --- /dev/null +++ b/submission/Shabbirsheikh/DESIGN.md @@ -0,0 +1,36 @@ +# CDC Lakehouse — E-commerce Orders + +## Proposed Approach + +My understanding is that the objective is to build a reliable CDC pipeline that keeps a lake and a warehouse synchronized with a transactional source, while handling replay, schema drift, and recovery scenarios. + +Domain: an order-management slice of e-commerce — customers, orders, order_items, payments — chosen for a natural strong/weak entity split and a cross-table invariant (order total vs. line items) worth validating downstream. + +### Initial Technology Stack + +- Python +- DuckDB (source database, local — simulates the transactional OLTP system) +- Simulated CDC (DuckDB has no WAL/logical-replication slot to tail like Postgres, so change capture is emitted at the application layer as sequence-numbered insert/update/delete events, standing in for a real log) +- Kafka (event transport for change records, local broker) +- Spark Structured Streaming (consumes Kafka, writes lake + warehouse) +- Parquet (lake: append-only change history) +- Delta Lake (warehouse storage: latest-state snapshot via MERGE/upsert, with native time travel for restore — written by Spark, queried locally via DuckDB's `delta` extension so DuckDB remains the single query interface across source and warehouse) +- pytest (validation and correctness tests) +- Airflow (stretch goal: orchestration/scheduling wrapper once the core pipeline is correct) + +### High-Level Flow + +``` +Source (DuckDB: customers, orders, order_items, payments) + -> CDC capture (sequence-numbered insert/update/delete events, app-layer emitted) + -> Kafka (change event topic) + -> Spark Structured Streaming + -> Lake (Parquet, append-only, full history) + -> Warehouse (Delta Lake, latest-state via merge, time travel via Delta versioning) +``` + +Schema-contract checks run before ingestion to detect breaking source changes; the warehouse is rebuilt from the lake, so point-in-time reconstruction is possible either by replaying up to a given sequence/offset or via Delta's native `VERSION AS OF` / `TIMESTAMP AS OF`. + +### Notes + +This represents my initial approach before implementation. CDC capture itself is simulated at the application layer rather than a real WAL connector — DuckDB doesn't expose one — while Kafka, Spark, and Delta Lake are used for real as the transport, processing, and warehouse-storage layers. As development progresses, I may refine details such as schema-change detection, replay/duplicate handling, and warehouse time-travel modeling based on testing and practical considerations. From ea2ea56f44b6724f1659f5f4b9babe42d5acb22f Mon Sep 17 00:00:00 2001 From: Shabbir Sheikh Date: Sun, 26 Jul 2026 12:31:16 +0530 Subject: [PATCH 2/2] Implement CDC lakehouse pipeline: Kafka + Spark + DuckDB --- submission/Shabbirsheikh/.gitignore | 14 + submission/Shabbirsheikh/DESIGN.md | 345 ++++++++++++++++-- submission/Shabbirsheikh/README.md | 160 ++++++++ submission/Shabbirsheikh/catalog/catalog.json | 94 +++++ submission/Shabbirsheikh/docs/catalog.md | 61 ++++ .../Shabbirsheikh/docs/cdc_and_streaming.md | 101 +++++ .../docs/schema_safety_and_data_quality.md | 79 ++++ .../docs/warehouse_and_time_travel.md | 98 +++++ submission/Shabbirsheikh/feature.md | 120 ++++++ submission/Shabbirsheikh/pipeline/__init__.py | 0 .../Shabbirsheikh/pipeline/kafka_producer.py | 61 ++++ .../Shabbirsheikh/pipeline/source_writer.py | 40 ++ .../Shabbirsheikh/pipeline/spark_consumer.py | 136 +++++++ .../Shabbirsheikh/pipeline/warehouse.py | 101 +++++ submission/Shabbirsheikh/scripts/__init__.py | 0 .../scripts/check_schema_contracts.py | 112 ++++++ .../scripts/run_data_quality_checks.py | 205 +++++++++++ .../Shabbirsheikh/scripts/seed_demo_data.py | 108 ++++++ .../Shabbirsheikh/scripts/validate_catalog.py | 83 +++++ submission/Shabbirsheikh/source.md | 115 ++++++ submission/Shabbirsheikh/source/__init__.py | 0 submission/Shabbirsheikh/source/models.py | 173 +++++++++ submission/Shabbirsheikh/tests/__init__.py | 0 submission/Shabbirsheikh/tests/conftest.py | 25 ++ submission/Shabbirsheikh/tests/helpers.py | 76 ++++ .../Shabbirsheikh/tests/test_catalog.py | 39 ++ .../Shabbirsheikh/tests/test_data_quality.py | 84 +++++ .../tests/test_schema_contracts.py | 62 ++++ .../Shabbirsheikh/tests/test_warehouse.py | 54 +++ 29 files changed, 2523 insertions(+), 23 deletions(-) create mode 100644 submission/Shabbirsheikh/.gitignore create mode 100644 submission/Shabbirsheikh/README.md create mode 100644 submission/Shabbirsheikh/catalog/catalog.json create mode 100644 submission/Shabbirsheikh/docs/catalog.md create mode 100644 submission/Shabbirsheikh/docs/cdc_and_streaming.md create mode 100644 submission/Shabbirsheikh/docs/schema_safety_and_data_quality.md create mode 100644 submission/Shabbirsheikh/docs/warehouse_and_time_travel.md create mode 100644 submission/Shabbirsheikh/feature.md create mode 100644 submission/Shabbirsheikh/pipeline/__init__.py create mode 100644 submission/Shabbirsheikh/pipeline/kafka_producer.py create mode 100644 submission/Shabbirsheikh/pipeline/source_writer.py create mode 100644 submission/Shabbirsheikh/pipeline/spark_consumer.py create mode 100644 submission/Shabbirsheikh/pipeline/warehouse.py create mode 100644 submission/Shabbirsheikh/scripts/__init__.py create mode 100644 submission/Shabbirsheikh/scripts/check_schema_contracts.py create mode 100644 submission/Shabbirsheikh/scripts/run_data_quality_checks.py create mode 100644 submission/Shabbirsheikh/scripts/seed_demo_data.py create mode 100644 submission/Shabbirsheikh/scripts/validate_catalog.py create mode 100644 submission/Shabbirsheikh/source.md create mode 100644 submission/Shabbirsheikh/source/__init__.py create mode 100644 submission/Shabbirsheikh/source/models.py create mode 100644 submission/Shabbirsheikh/tests/__init__.py create mode 100644 submission/Shabbirsheikh/tests/conftest.py create mode 100644 submission/Shabbirsheikh/tests/helpers.py create mode 100644 submission/Shabbirsheikh/tests/test_catalog.py create mode 100644 submission/Shabbirsheikh/tests/test_data_quality.py create mode 100644 submission/Shabbirsheikh/tests/test_schema_contracts.py create mode 100644 submission/Shabbirsheikh/tests/test_warehouse.py diff --git a/submission/Shabbirsheikh/.gitignore b/submission/Shabbirsheikh/.gitignore new file mode 100644 index 0000000..b358bbd --- /dev/null +++ b/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 diff --git a/submission/Shabbirsheikh/DESIGN.md b/submission/Shabbirsheikh/DESIGN.md index b783fd7..9f8354e 100644 --- a/submission/Shabbirsheikh/DESIGN.md +++ b/submission/Shabbirsheikh/DESIGN.md @@ -1,36 +1,335 @@ # CDC Lakehouse — E-commerce Orders -## Proposed Approach +## Summary -My understanding is that the objective is to build a reliable CDC pipeline that keeps a lake and a warehouse synchronized with a transactional source, while handling replay, schema drift, and recovery scenarios. +This is a CDC pipeline that keeps a Parquet lake and a DuckDB warehouse +synchronized from a simulated transactional order-management source, using +Kafka as the transport and Spark Structured Streaming as the processing +layer. It covers all six of the assignment's core requirements: a source +schema with real strong/weak entities, a CDC flow that captures every +insert/update/delete, fail-closed schema-drift detection, point-in-time +restore via a bounded lake replay, a validation-parity layer that mirrors +the source's constraints against the (deliberately loosely-typed) +warehouse, and a catalog exposing every lake/warehouse dataset with +ownership and consumer metadata. -Domain: an order-management slice of e-commerce — customers, orders, order_items, payments — chosen for a natural strong/weak entity split and a cross-table invariant (order total vs. line items) worth validating downstream. +I initially planned this with Delta Lake as the warehouse storage layer +(see the very first version of this document, still visible in the git +history / the early PR commits). During implementation I simplified that +to plain DuckDB plus an explicit `reconstruct_as_of()` replay function +instead — this avoided taking on a table-format dependency for a +single-node, assignment-scoped warehouse, at the cost of not getting +Delta's built-in `VERSION AS OF` querying for free. I'd rather be upfront +about that change than leave the original plan looking like what actually +got built. -### Initial Technology Stack +I also want to flag early, rather than bury it: this is *simulated* CDC. +DuckDB has no WAL or logical-replication slot the way Postgres does, so +there's no real log to tail. Change capture happens at the application +layer instead — documented in detail below, including the specific +failure mode this introduces and how a real Postgres-backed deployment +would close that gap. -- Python -- DuckDB (source database, local — simulates the transactional OLTP system) -- Simulated CDC (DuckDB has no WAL/logical-replication slot to tail like Postgres, so change capture is emitted at the application layer as sequence-numbered insert/update/delete events, standing in for a real log) -- Kafka (event transport for change records, local broker) -- Spark Structured Streaming (consumes Kafka, writes lake + warehouse) -- Parquet (lake: append-only change history) -- Delta Lake (warehouse storage: latest-state snapshot via MERGE/upsert, with native time travel for restore — written by Spark, queried locally via DuckDB's `delta` extension so DuckDB remains the single query interface across source and warehouse) -- pytest (validation and correctness tests) -- Airflow (stretch goal: orchestration/scheduling wrapper once the core pipeline is correct) +This document is the design narrative. For quicker reference: `feature.md` +maps each requirement below to exactly what got built, `source.md` has the +full schema and every invariant, and `docs/` has one file per component +(CDC/streaming, warehouse/time-travel, schema/data-quality, catalog) with +implementation-level detail I didn't want to bloat this document with. +`README.md` has the setup and run commands. -### High-Level Flow +## Source Schema Design + +Domain: order-management, e-commerce. Customers place orders; orders +contain line items and are paid via one or more payment attempts. + +**Strong entities**: `customers`, `orders` — both have independent +identity and their own lifecycle, not owned by anything else. + +**Weak entities**: `order_items` (its lifecycle is entirely tied to its +parent order — a line item has no meaning outside an order) and +`payments` (tied to an order, but more loosely — a payment can arrive or +retry after the order was created, so it carries its own timestamps and +status independent of the order's). ``` -Source (DuckDB: customers, orders, order_items, payments) - -> CDC capture (sequence-numbered insert/update/delete events, app-layer emitted) - -> Kafka (change event topic) - -> Spark Structured Streaming - -> Lake (Parquet, append-only, full history) - -> Warehouse (Delta Lake, latest-state via merge, time travel via Delta versioning) +customers(customer_id PK, name, email, status enum, created_at, updated_at) +orders(order_id PK, customer_id FK -> customers, status enum, + total_amount DECIMAL(18,2), discount_code nullable, + created_at, updated_at) +order_items(order_item_id PK, order_id, sku, quantity INT >0, + unit_price DECIMAL(18,2) >=0) +payments(payment_id PK, order_id, amount DECIMAL(18,2) >0, status enum, + created_at, paid_at nullable) +``` + +Indexes: `idx_orders_customer_id`, `idx_orders_status`, +`idx_order_items_order_id`, `idx_payments_order_id` — on every FK-shaped +lookup column, since those are what both the referential-integrity check +and any downstream analytical query would filter/join on. + +**One deliberate deviation from a textbook FK design**: `order_items.order_id` +and `payments.order_id` are *not* declared with `REFERENCES orders(order_id)`, +even though that's the obviously "correct" relational choice. I hit a real +DuckDB engine limitation while building this — on DuckDB 1.5.5, you cannot +`UPDATE` a table that is simultaneously an FK child (`orders` references +`customers`) and an FK parent (`order_items`/`payments` reference `orders`): +any `UPDATE` on `orders` fails with "Violates foreign key constraint +because key is still referenced," even when the update doesn't touch +`order_id` at all. I didn't take this at face value — I built an isolated +minimal repro (a plain grandparent → parent → child FK chain, nothing else +in it) before concluding it was a genuine engine bug rather than something +wrong in my schema. Since `orders.status` changes constantly (the single +most common write in this domain), `orders` had to stay updatable, so I +dropped the FK declaration on the child side and moved that referential +check into the data-quality layer instead. On Postgres or MySQL this +workaround wouldn't be necessary. + +**Validation rules** (source-enforced via PK/CHECK/NOT NULL, re-verified +downstream — see Validation Parity): +- `orders.total_amount` should reconcile with + `sum(order_items.quantity * unit_price)` for that order. +- `order_items.quantity > 0`, `unit_price >= 0`; `payments.amount > 0`. +- `payments.paid_at >= payments.created_at` when set. +- Status fields restricted to a known enum per table. +- Status transitions follow a legal state machine (e.g. an order can go + `pending -> paid -> shipped`, but not `cancelled -> paid`). + +## CDC Strategy + +**How changes are captured**: a `SourceWriter` class does two things, in +this order, on every call: (1) runs the real SQL against `source.duckdb` +so the source's own constraints apply, and (2) — only after that succeeds +— publishes the corresponding event to Kafka. This is application-level +dual-write, not log-based capture, because DuckDB doesn't expose a WAL to +tail. I'm calling this out plainly rather than dressing it up: if the +process crashes between step 1 and step 2, that change is durably in the +source but was never published. In production, with a real Postgres +source, this would be replaced by Debezium tailing the database's WAL, +which guarantees capture regardless of what happens to any application +process in between. + +**How inserts/updates/deletes are handled**: all three map to a Kafka +event carrying an `operation` field (`insert`/`update`/`delete`), the +table name, the primary key, a full snapshot of the row's current values, +and a capture timestamp. Deletes are captured with the row's last-known +values (not just the key) so the lake retains what the row looked like +right before deletion. + +**Ordering**: the Kafka topic (`cdc-events-v4`) has exactly one partition, +so the message offset is a true global order across every table — that +offset becomes the `sequence` number used by every downstream component +(lake row, warehouse `_cdc_seq`, time-travel bounding), with no separate +manually-maintained counter. I chose durability confirmation over +throughput here: the producer uses `acks="all"` and blocks +(`future.get(timeout=10)`) until the broker fully acknowledges the write, +so a successful call is a real durability guarantee, not just "sent." + +**How replay/restart works**: Spark Structured Streaming's own +`checkpointLocation` tracks which Kafka offsets have already been +processed — that's the restart-safety mechanism, built into the framework +itself, so I didn't need a separate offset-tracking table for this path. +I'm relying on this being a well-established Spark guarantee rather than +having independently proven it here: I haven't run a dedicated +kill-and-restart test against this exact pipeline, and there's no +automated test for it either (see Known Limitations) — the `_cdc_seq` +guard in the warehouse merge is a second, independent safety net either +way, so a redelivered batch is a no-op regardless. + +**How duplicates are handled**: the warehouse merge upserts with +`WHERE excluded._cdc_seq > current._cdc_seq` — a duplicate or +out-of-order event can never overwrite a row that's already ahead of it, +so replaying the same batch twice is a no-op on the warehouse. The lake +itself isn't strictly exactly-once at the storage layer — I actually hit +this during testing: a Spark batch that crashed mid-way (after writing to +Parquet but before finishing the warehouse write) produced a duplicate row +in the lake once I reran it. Harmless for warehouse correctness, since the +merge guard dedupes regardless of how many times an event is replayed, but +worth being honest that the lake can carry small amounts of redundant +storage as a result. + +## Lake and Warehouse Modeling + +**Lake**: append-only Parquet, written by Spark's native streaming sink, +partitioned by day (`data/lake_parquet/event_date=YYYY-MM-DD/`) so a +specific day's history can be targeted directly for reprocessing, +investigation, or a bounded restore without scanning the whole lake. +Nothing here is ever updated or deleted — it's the durable, replayable +history everything else is built from. + +**Warehouse**: DuckDB tables (`wh_customers`, `wh_orders`, `wh_order_items`, +`wh_payments`), each mirroring its source table's columns plus `_cdc_seq` +(sequence of the last applied event) and `_deleted` (soft-delete flag — a +source delete never physically removes the warehouse row, since it's +still needed for history/restore). There's no separate write path into +the warehouse; it's always a function of "which lake events have been +applied so far," which is what keeps it consistent with the lake and what +makes rebuild/restore possible in the first place. + +**Time travel / restore**: `reconstruct_as_of(lake_rows_up_to_bound)` +takes a caller-bounded list of lake events (bounded by sequence or by +timestamp — the function itself doesn't care which), replays them through +the exact same merge logic into a brand-new, throwaway, in-memory +warehouse, and returns that connection. It never touches the live +warehouse — "restore" here means rebuild elsewhere, verify it, then decide +what to do with the live system, not overwrite it blind. + +I verified this live, not just in unit tests: for one order that moved +`pending -> paid -> shipped` across three real Kafka offsets (19, 20, 21), +bounding the replay to `sequence <= 19` correctly reproduced `pending`, +bounding to `sequence <= 20` correctly reproduced `paid`, and the live +warehouse correctly showed `shipped` — three different, independently +correct answers for three different bounds, none of which touched the +live warehouse. + +## Schema Change Safety + +`check_schema_contracts.py` diffs the live source's +`information_schema.columns` against an expected-columns/types contract. +A **breaking** change (dropped column, renamed column, changed type) +raises `SchemaContractViolation`; the script catches it at the top level, +prints every specific problem, and exits non-zero — ingestion should stop +rather than keep writing against a wrong assumption about the schema. A +purely **additive** change (a genuinely new column) is reported as a +warning instead of a failure, since old code simply won't reference a +column it doesn't know about yet — it's a signal the contract needs +updating, not a reason to halt. + +I'll be direct about the current gap here: nothing automated is watching +for that non-zero exit code today — it's a manually-run script. In +production this would be wired into an orchestrator (Airflow) so a +schema-contract failure fails the task and pages someone, or pauses the +pipeline until it's fixed. + +## Validation Parity + +The warehouse tables are typed loosely on purpose, so the merge/upsert +logic can stay generic across all four tables — which means none of the +source's constraints (`NOT NULL`, `CHECK`, the one FK) physically exist on +the warehouse side. `run_data_quality_checks.py` puts that parity back: + +- **System checks**: PK uniqueness, not-null, referential integrity + (including the `order_items`/`payments` → `orders` link that DuckDB + itself couldn't enforce), enum domain. +- **Business checks**: non-negative amounts, order-total-matches-line-items, + payment timing, and status-transition legality. The last one needs the + lake's *full* history per entity, not just the warehouse's latest state + — detecting an illegal transition (`cancelled -> paid`) requires knowing + the sequence of past statuses, which the warehouse structurally doesn't + retain. + +Failures are surfaced the same way as the schema check: an `enforce()` +function raises `DataQualityViolation` listing every failing category and +row, caught at the top level, non-zero exit. Same current gap as above — +nothing automated pages anyone on failure yet. + +## Catalog Exposure + +`catalog/catalog.json` registers 5 datasets: the lake dataset +(`lake_cdc_events`) and the four warehouse tables. There's no +raw/source-layer entry — the source database is the transactional system +of record, not something published for downstream consumption. Each entry +documents an owner, its intended consumers (these genuinely differ per +table — `wh_payments` lists `finance` and `audit`, `wh_order_items` +doesn't), an update cadence, a description, and the full column schema. + +This isn't just a document that can silently go stale: `validate_catalog.py` +checks that every dataset the pipeline actually produces has a complete +catalog entry, and that a warehouse entry's documented schema hasn't +drifted from what the code actually expects. A catalog that can drift +unnoticed is worse than no catalog, so I made that check automatic rather +than something to remember. + +A consumer discovers datasets by reading `catalog.json` directly (or a +production metadata platform indexing it — see Known Limitations) and +queries them accordingly: the lake via `read_parquet('data/lake_parquet/*/*.parquet')` +from any DuckDB connection, the warehouse by connecting directly to +`warehouse.duckdb` and querying `wh_*` tables with normal SQL. + +## How to Run / Validate + +```bash +# tests (in-memory, ~2s, never touches real source/warehouse data) +python -m pytest -q + +# source schema still matches the expected contract +python scripts/check_schema_contracts.py + +# warehouse + lake pass every system and business validation +python scripts/run_data_quality_checks.py + +# every lake/warehouse dataset is registered with complete metadata +python scripts/validate_catalog.py + +# seed a consistent demo dataset — writes to source.duckdb AND publishes +# the matching events to Kafka in the same run +python scripts/seed_demo_data.py + +# process those Kafka events into the lake + warehouse (on-demand, default — +# processes whatever's currently in Kafka, then exits) +docker run --rm --network host --user 1000:1000 \ + -v /etc/passwd:/etc/passwd:ro \ + -v /tmp/kafka-clients-3.9.0.jar:/opt/spark/jars/kafka-clients-4.0.0.jar:ro \ + -v /tmp/commons-pool2-2.12.0.jar:/opt/spark/jars/commons-pool2-2.12.0.jar:ro \ + -v /u01/container-pylibs:/opt/pylibs:ro \ + -v $(pwd):/workspace -w /workspace -e PYTHONPATH=/opt/pylibs \ + spark-kafka:4.0.0 /opt/spark/bin/spark-submit --master 'local[*]' pipeline/spark_consumer.py + +# or, to run continuously instead (polls Kafka every 5 seconds and keeps +# picking up new events automatically — stop with `docker stop` when done, +# it holds a resident Spark driver process): +docker run --rm --network host --user 1000:1000 \ + -v /etc/passwd:/etc/passwd:ro \ + -v /tmp/kafka-clients-3.9.0.jar:/opt/spark/jars/kafka-clients-4.0.0.jar:ro \ + -v /tmp/commons-pool2-2.12.0.jar:/opt/spark/jars/commons-pool2-2.12.0.jar:ro \ + -v /u01/container-pylibs:/opt/pylibs:ro \ + -v $(pwd):/workspace -w /workspace -e PYTHONPATH=/opt/pylibs \ + spark-kafka:4.0.0 /opt/spark/bin/spark-submit --master 'local[*]' pipeline/spark_consumer.py --continuous ``` -Schema-contract checks run before ingestion to detect breaking source changes; the warehouse is rebuilt from the lake, so point-in-time reconstruction is possible either by replaying up to a given sequence/offset or via Delta's native `VERSION AS OF` / `TIMESTAMP AS OF`. +I've run every command above, and the full end-to-end flow (seed data → +Kafka → Spark → lake + warehouse), against this exact code: 21/21 tests +pass, schema contract passes, data quality checks pass with zero +violations, and catalog validation passes. + +## Known Limitations / Next Steps + +Things I'd tackle next, in priority order, if this were going to +production: + +1. **Real CDC, not simulated** — replace the application-level dual-write + with Debezium tailing a real Postgres WAL, closing the crash-between-writes + gap described above. +2. **Monitoring and alerting** — right now, a schema-contract or + data-quality failure just exits non-zero; nothing pages anyone. I'd add + Kafka consumer-lag metrics, Spark `StreamingQueryListener` batch-lag + metrics, and wire validation failures into Slack/PagerDuty. +3. **Orchestration** — Spark currently runs via a manual Docker command; a + production setup needs Airflow (or similar) for scheduling and + restart-on-failure, especially for the on-demand run mode. +4. **Multi-partition Kafka** — a single partition is a throughput ceiling + at real volume. The fix is partitioning by entity key and relying on + the `_cdc_seq` guard for per-row order instead of a global order. +5. **A concurrent-capable production warehouse** — DuckDB is single-writer, + which is fine for this scale and for proving correctness, but I + wouldn't choose it for a production warehouse serving concurrent + analysts while Spark is also writing to it. +6. **Richer schema-evolution handling** — the current contract check only + distinguishes breaking vs. additive; a schema-registry-based + compatibility model (Avro/Protobuf-style) would handle more nuance. +7. **No automated restart-recovery test** — restart-safety currently rests + on Spark's own checkpointing (a well-established framework guarantee) + plus the warehouse's `_cdc_seq` guard, but I don't have a dedicated + kill-and-restart test proving it against this exact pipeline. This is + the most concrete gap I'd close first, since the assignment's testing + requirements call for it directly. -### Notes +## Responsible AI Usage -This represents my initial approach before implementation. CDC capture itself is simulated at the application layer rather than a real WAL connector — DuckDB doesn't expose one — while Kafka, Spark, and Delta Lake are used for real as the transport, processing, and warehouse-storage layers. As development progresses, I may refine details such as schema-change detection, replay/duplicate handling, and warehouse time-travel modeling based on testing and practical considerations. +I used Claude during implementation to speed up writing boilerplate and +draft code. The domain choice, the build order (a DuckDB-only core first +to validate the merge logic, then Kafka and Spark layered on top), and the +trade-offs (DuckDB over Postgres, on-demand as the default run mode given +the VM's memory constraints) were my decisions. I personally ran and +verified the infrastructure at every stage — records landing in +Kafka/lake/warehouse, continuous mode picking up a new record live, and +the time-travel reconstruction documented above. diff --git a/submission/Shabbirsheikh/README.md b/submission/Shabbirsheikh/README.md new file mode 100644 index 0000000..5440744 --- /dev/null +++ b/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://: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 `` 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. diff --git a/submission/Shabbirsheikh/catalog/catalog.json b/submission/Shabbirsheikh/catalog/catalog.json new file mode 100644 index 0000000..91caa62 --- /dev/null +++ b/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" + } + } + ] +} diff --git a/submission/Shabbirsheikh/docs/catalog.md b/submission/Shabbirsheikh/docs/catalog.md new file mode 100644 index 0000000..91953d1 --- /dev/null +++ b/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_` 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. diff --git a/submission/Shabbirsheikh/docs/cdc_and_streaming.md b/submission/Shabbirsheikh/docs/cdc_and_streaming.md new file mode 100644 index 0000000..60fb507 --- /dev/null +++ b/submission/Shabbirsheikh/docs/cdc_and_streaming.md @@ -0,0 +1,101 @@ +# CDC Capture and Streaming + +Covers: `pipeline/source_writer.py`, `pipeline/kafka_producer.py`, +`pipeline/spark_consumer.py`. + +## Capture: application-level dual-write, not log-based + +`SourceWriter.insert/update/delete()` does two things in sequence, on +every call: + +1. Runs the real SQL against `source.duckdb` (so the source's own + PK/FK/CHECK/NOT NULL constraints apply, same as production). +2. Only *after* that succeeds, calls `KafkaChangeCapture.emit()` to + publish the matching event to Kafka. + +This ordering matters: it guarantees the pipeline never emits an event for +something that didn't actually happen in the source. + +**This is simulated CDC, not log-based capture.** DuckDB has no +WAL / logical-replication slot to tail the way Postgres does, so there is +no way to do real log-based capture against it. The known risk: if the +process crashes between step 1 and step 2, that change is durably in the +source but was never published — the classic "dual-write problem." In +production, with a real Postgres (or similar) source, this would be +replaced by Debezium tailing the database's WAL, which guarantees that +whatever commits to the source is captured regardless of what happens to +any application process. + +## Transport: Kafka, single partition + +`pipeline/kafka_producer.py`: + +- `make_producer()` creates a `KafkaProducer` with `acks="all"` — the + `.send()` call blocks (`future.get(timeout=10)`) until the broker + fully acknowledges the write. That acknowledgment is the durability + point: once it returns, the event is durable on the broker even if + every downstream consumer is down. +- Topic `cdc-events-v4` is created with exactly one partition. This means + the Kafka message offset is a true global ordering across every event + from every table — that offset becomes the `sequence` number used by + every downstream component (lake row, warehouse `_cdc_seq`, time-travel + bounding), with no separate manually-maintained counter. +- Trade-off: a single partition caps throughput to what one partition can + handle. At real production volume, the fix is to partition by entity + key (e.g. `customer_id`) and drop the requirement for *global* order — + only *per-row* order matters for correctness, which the warehouse's + `_cdc_seq` guard already enforces independent of partition count. + +## Processing: Spark Structured Streaming + +`pipeline/spark_consumer.py`: + +- Reads the topic (`readStream.format("kafka")`), parses the JSON + envelope (`operation`, `table_name`, `primary_key`, `data`, + `captured_at`), and derives an `event_date` partition column from + `captured_at`. +- `foreachBatch(process_batch)` does two things per micro-batch: + 1. Appends the batch to the Parquet lake + (`data/lake_parquet/event_date=YYYY-MM-DD/*.parquet`), partitioned by + day so a specific day's history can be targeted directly later + (reprocessing, investigation, a bounded restore) without scanning + the whole lake. + 2. Collects the batch to the driver, in sequence order, and hands it to + `pipeline/warehouse.py`'s `apply_lake_events()` — the same merge + function the test suite exercises directly. Spark's job is only "get + Kafka data to the driver, in order"; the correctness guarantee + (last-write-wins via `_cdc_seq`) comes from code that's independently + tested. +- The DuckDB warehouse connection is opened and closed fresh **inside** + `process_batch()`, per batch — not held for the job's lifetime. This + was a deliberate fix after hitting a real cross-process test failure: + DuckDB holds an exclusive lock for as long as a connection stays open, + so in continuous mode (which can run for hours), a long-lived + connection would permanently lock out every other reader. Reconnecting + per batch means the lock is only held for the few milliseconds it takes + to apply that batch. + +## Run modes and latency + +- **On-demand (default)**: `trigger(availableNow=True)` — processes + whatever is currently in Kafka, then exits. Predictable, no background + process left running. +- **Continuous** (`--continuous` flag): `trigger(processingTime="5 + seconds")` — polls Kafka every 5 seconds indefinitely. Worst-case + latency for a single event is bounded by that interval (up to 5 + seconds); best case, an event arriving just before a scheduled poll is + picked up almost immediately. This leaves a resident Spark driver + process (JVM, a few hundred MB) — verified live that this needs to be + explicitly stopped after use, since an unattended continuous container + combined with a shared checkpoint directory across two different topic + names caused real memory pressure and a corrupted checkpoint during + development. + +## Restart-safety + +Spark's own `checkpointLocation` (`data/spark_checkpoints/main`) tracks +which Kafka offsets have already been processed — this is the +restart-safety mechanism for this path, built into Structured Streaming +itself. `apply_lake_events()`'s own `_cdc_seq` guard is a second, +independent safety net: even if Spark ever redelivered an already-applied +batch, re-applying it is a no-op on the warehouse. diff --git a/submission/Shabbirsheikh/docs/schema_safety_and_data_quality.md b/submission/Shabbirsheikh/docs/schema_safety_and_data_quality.md new file mode 100644 index 0000000..17a3e72 --- /dev/null +++ b/submission/Shabbirsheikh/docs/schema_safety_and_data_quality.md @@ -0,0 +1,79 @@ +# Schema Safety and Data Quality + +Covers: `scripts/check_schema_contracts.py`, `scripts/run_data_quality_checks.py`. + +## Schema-contract check: fail-closed on breaking changes + +`check_schema_contracts.py` compares the live source's +`information_schema.columns` against an expected-columns/types contract +defined in `source/models.py` (`EXPECTED_COLUMNS`, `COLUMN_TYPES`). + +Two categories of difference, handled differently: + +- **Breaking** — a dropped column, a renamed column (which looks like a + drop), or a changed type. `enforce()` raises `SchemaContractViolation`; + the top-level script catches it, prints every specific problem to + stderr, and exits with a non-zero code. Ingestion should stop here + rather than keep writing against a wrong assumption about the schema. +- **Non-breaking** — a genuinely new column that isn't in the contract + yet. Reported as a warning, not a failure: old code simply won't + reference a column it doesn't know about, so nothing crashes. It's a + signal that the contract needs updating, not a reason to stop. + +**Current gap**: nothing automated currently watches for that non-zero +exit code — it's a manually-run script. In production this would be +wired into an orchestrator (e.g. Airflow) so a schema-contract failure +fails the task and pages someone, or pauses the downstream pipeline. + +## Data-quality checks: putting back the constraints the warehouse doesn't have + +The warehouse tables are typed loosely on purpose (see +`warehouse_and_time_travel.md`) so the merge logic stays generic — which +means none of the source's constraints (`NOT NULL`, `CHECK`, `FK`) +physically exist on the warehouse side. `run_data_quality_checks.py` puts +that parity back: every rule the source enforces gets re-checked here +against the warehouse, so a CDC bug or a late/out-of-order write can't +silently corrupt it without anyone noticing. + +### System checks + +- `check_pk_uniqueness` — no duplicate primary keys per table. +- `check_not_null` — required columns aren't null. +- `check_referential_integrity` — every FK-shaped column resolves to a + real parent row, including `orders -> customers` (which *is* a real DB + FK) and `order_items/payments -> orders` (which is *not* a DB FK, for + the DuckDB engine-limitation reason documented in `source.md` — this is + the one place that link's integrity is actually enforced). +- `check_enum_domain` — status columns only contain known values. + +### Business checks + +- `check_non_negative_amounts` — order totals, item quantities/prices, + payment amounts are all within their valid range. +- `check_order_totals_match_items` — `orders.total_amount` reconciles + with `sum(order_items.quantity * unit_price)` for that order. +- `check_payment_timing` — `payments.paid_at >= payments.created_at`. +- `check_status_transitions` — the one check that needs more than the + warehouse's latest state. It replays the lake's *full* history per + entity (via `_status_history_per_entity`, sorted by `sequence`) and + flags any transition that isn't in the allowed state machine (e.g. + `cancelled -> paid` is illegal for an order). This structurally + requires the lake, not the warehouse, because the warehouse has no + memory of what a status used to be. + +### Where the lake events come from + +`_read_lake_events()` reads directly from the Parquet lake via +`read_parquet('data/lake_parquet/*/*.parquet')`, reusing the same DuckDB +connection as the warehouse check (any open DuckDB connection can query a +Parquet file by path — it doesn't have to be "the lake's own" +connection). `run_all()`/`enforce()` take this as a plain list of event +dicts, not a database connection, which is also what lets the test suite +build the same shape of list by hand without touching Parquet at all. + +### Failure surfacing + +Same pattern as the schema-contract check: `enforce()` raises +`DataQualityViolation` listing every failure by category, caught at the +top level, exits non-zero. Same current gap: nothing automated currently +watches for that exit code. diff --git a/submission/Shabbirsheikh/docs/warehouse_and_time_travel.md b/submission/Shabbirsheikh/docs/warehouse_and_time_travel.md new file mode 100644 index 0000000..a43a449 --- /dev/null +++ b/submission/Shabbirsheikh/docs/warehouse_and_time_travel.md @@ -0,0 +1,98 @@ +# Warehouse and Time Travel + +Covers: `pipeline/warehouse.py`. + +## Model + +The warehouse is the latest known state per row, built entirely by +replaying the lake — there is no separate write path into it. It is +always a function of "which lake events have been applied so far," which +is what keeps it consistent with the lake and makes rebuild/restore +possible. + +Each warehouse table (`wh_customers`, `wh_orders`, `wh_order_items`, +`wh_payments`) mirrors its source table's columns/types, plus two extras: + +- `_cdc_seq` — the sequence (Kafka offset) of the last lake event applied + to this row. +- `_deleted` — soft-delete flag. A source delete never physically removes + the warehouse row, since the row is still needed for history/restore. + +Warehouse tables are typed loosely (no `NOT NULL`/`CHECK`/`FK`) so the +merge logic can stay generic across all four tables — this is why a +separate validation-parity layer exists (see +`schema_safety_and_data_quality.md`). + +## Merge logic: last-write-wins + +`apply_lake_events(conn, lake_rows)` does a full-column upsert per event: + +```sql +INSERT INTO wh_x (...) VALUES (...) +ON CONFLICT (pk) DO UPDATE SET ... +WHERE excluded._cdc_seq > wh_x._cdc_seq +``` + +The `WHERE` clause on the upsert is the entire correctness guarantee: an +out-of-order or duplicate event can never clobber a row that's already +ahead of it, because the write is simply rejected if its sequence isn't +strictly newer than what's already there. Each event's `data` is the full +row snapshot at capture time, not a partial diff — applying it is always +a full-column overwrite, never a merge of individual fields. + +This function is storage-agnostic by design: it takes a plain list of +event dicts, and doesn't care whether they came from a Parquet file, a +Kafka microbatch, or a test building them by hand. That's what lets +`pipeline/spark_consumer.py` reuse this exact function as its merge logic, +and lets the test suite exercise it without touching Kafka, Spark, or +Parquet at all. + +## Time travel: `reconstruct_as_of` + +```python +def reconstruct_as_of(lake_rows_up_to_bound: list[dict]) -> DuckDBPyConnection: + snapshot_conn = duckdb.connect(":memory:") + create_warehouse_tables(snapshot_conn) + apply_lake_events(snapshot_conn, lake_rows_up_to_bound) + return snapshot_conn +``` + +Given a list of lake events already bounded by the caller (e.g. +`sequence <= N`, or `captured_at <= some_timestamp`), this builds a +brand-new, throwaway, in-memory warehouse from scratch, using only those +events, and returns it. It never touches the live warehouse — this is +deliberately what "restore to a prior point in time" means operationally +here: rebuild the state somewhere else, verify it, then decide what (if +anything) to do with the live system. + +The function itself doesn't know or care how the bound was chosen — that +responsibility belongs entirely to the caller, which keeps the function +simple and reusable for either a sequence-based or timestamp-based +restore. + +### Verified live + +For a single order (`tt_order1`) that moved `pending -> paid -> shipped` +across three real Kafka offsets (19, 20, 21): + +| Query | Result | +|---|---| +| Live warehouse (current) | `shipped`, `_cdc_seq = 21` | +| `reconstruct_as_of(events <= 19)` | `pending` | +| `reconstruct_as_of(events <= 20)` | `paid` | + +Three different, independently correct answers for three different +bounds, none of which touched the live warehouse. + +## Note on the original design + +The initial approach note (`DESIGN.md`) proposed Delta Lake as the +warehouse storage layer, using its native transaction-log-based version +history (`VERSION AS OF` / `TIMESTAMP AS OF`) for time travel. During +implementation this was simplified to plain DuckDB plus this explicit +`reconstruct_as_of` replay function instead. Trade-off: no ACID +transaction log or built-in version querying the way Delta gives you, but +no extra table-format dependency either, for a warehouse that's +single-node and assignment-scoped. The lake genuinely is just Parquet +(not Delta/Iceberg), so this isn't a "lakehouse" in the strict sense — +it's a data lake plus a replay-based restore mechanism. diff --git a/submission/Shabbirsheikh/feature.md b/submission/Shabbirsheikh/feature.md new file mode 100644 index 0000000..4f441e0 --- /dev/null +++ b/submission/Shabbirsheikh/feature.md @@ -0,0 +1,120 @@ +# Feature Specification — CDC Lakehouse Pipeline (E-commerce Orders) + +This maps the assignment's six core requirements to what's actually built +and tested, not to an idealized version of it. `DESIGN.md` has the fuller +narrative and the reasoning behind each choice; this is the shorter, +requirement-by-requirement version. + +## 1. Source Data Model + +A transactional order-management schema — `customers`, `orders`, +`order_items`, `payments` — with a real strong/weak entity split and a +cross-table invariant (order total vs. line items) that's worth validating +downstream. Defined in `source/models.py` (`init_source_schema`), backed +by DuckDB with PK/FK/CHECK/NOT NULL constraints matching what a real OLTP +system would enforce. Full schema, entity rationale, and every invariant: +`source.md`. + +## 2. CDC Pipeline + +`pipeline/source_writer.py`'s `SourceWriter` writes the real SQL to +`source.duckdb` first, and only after that succeeds does it publish a +change event to Kafka via `pipeline/kafka_producer.py`. This is +application-level dual-write — simulated CDC, not log-based capture, +since DuckDB has no WAL or logical-replication slot to tail. The known +risk is a crash between the two writes losing an event; a production +source (Postgres) would use Debezium tailing the WAL instead, which +doesn't have that gap. + +The Kafka topic (`cdc-events-v4`) has exactly one partition, so the +message offset is a true global order — that offset becomes the +`sequence` used everywhere downstream, with no separate manual counter. +`pipeline/spark_consumer.py` (Spark Structured Streaming) reads the topic, +writes every event to the Parquet lake, then feeds the batch to +`pipeline/warehouse.py`'s `apply_lake_events()`, which upserts the +warehouse with a `WHERE excluded._cdc_seq > current._cdc_seq` guard so a +duplicate or out-of-order replay can never overwrite a row that's already +ahead of it. It runs either on-demand (processes what's currently in +Kafka, then exits — the default) or continuously (polls every 5 seconds). +Full mechanics: `docs/cdc_and_streaming.md`. + +## 3. Schema Change Detection and Safe Stop + +`scripts/check_schema_contracts.py` diffs the live source's +`information_schema.columns` against an expected-columns/types contract +in `source/models.py`. A breaking difference — a dropped column, a +renamed column, a changed type — raises `SchemaContractViolation`, caught +at the top level and surfaced as a non-zero exit code listing every +specific problem. A purely additive change (a genuinely new column) is a +warning instead, since nothing downstream references it yet. Full +mechanics: `docs/schema_safety_and_data_quality.md`. + +## 4. Historical Recovery and Time Travel + +`pipeline/warehouse.py`'s `reconstruct_as_of(lake_rows_up_to_bound)` takes +a caller-bounded list of lake events (bounded by sequence or timestamp), +replays them through the same merge logic into a brand-new in-memory +DuckDB, and returns that connection without ever touching the live +warehouse. I verified this live rather than trusting the unit tests alone: +for an order that moved `pending -> paid -> shipped` across three real +Kafka offsets, bounding the replay to each earlier offset in turn +correctly reproduced `pending` and then `paid`, while the live warehouse +correctly showed `shipped`. + +Worth noting: the original plan (see the early version of `DESIGN.md`) +proposed Delta Lake as the warehouse storage layer, using its native +version history for time travel. I simplified that to plain DuckDB plus +this explicit replay function during implementation, since it avoided a +table-format dependency for a warehouse this small — at the cost of not +getting Delta's built-in `VERSION AS OF` querying for free. Full +mechanics: `docs/warehouse_and_time_travel.md`. + +## 5. Validation Parity + +The warehouse tables are typed loosely, so none of the source's +constraints (PK uniqueness, not-null, referential integrity, enum domain) +physically exist on the warehouse side. `scripts/run_data_quality_checks.py` +re-checks them, plus the business rules — non-negative amounts, order +total vs. line items, payment timing, legal status transitions. The +status-transition check is the one that needs more than the warehouse's +latest state: detecting an illegal transition (`cancelled -> paid`) +requires the lake's full history per entity, since the warehouse has no +memory of what a status used to be. Full mechanics: +`docs/schema_safety_and_data_quality.md`. + +## 6. Catalog Exposure + +`catalog/catalog.json` registers 5 datasets — the lake plus the four +warehouse tables — each with an owner, intended consumers, update +cadence, and full column schema. `scripts/validate_catalog.py` enforces +that every dataset the pipeline actually produces has a complete entry, +and that a warehouse entry's documented schema hasn't drifted from the +code. Full details: `docs/catalog.md`. + +## Reliability Behavior + +Duplicate events are idempotent at the warehouse via the `_cdc_seq` +guard — a duplicate can still appear in the Parquet lake itself (e.g. +after a retried Spark batch), but that's harmless for warehouse +correctness. Out-of-order arrival is mostly prevented by the +single-partition Kafka topic, with the `_cdc_seq` guard as a second, +independent safety net either way. Restart safety rests on Spark's own +`checkpointLocation` tracking processed offsets, so there's no separate +offset-tracking table for this path. Deletes are soft — the warehouse row +is flagged `_deleted = TRUE` rather than removed, so history and restore +stay intact. + +## Known Limitations (Not Yet Built) + +- No orchestration — Spark runs manually via a Docker command, no Airflow + scheduling or restart-on-failure. +- No monitoring or alerting — schema/data-quality failures currently just + exit non-zero, nothing pages anyone. +- The lake isn't exactly-once at the storage layer — a retried batch can + duplicate Parquet rows, though warehouse correctness is unaffected. +- Schema-evolution detection is basic — dropped/renamed/type-changed vs. + additive only, no schema-registry-based compatibility negotiation. +- No automated restart-recovery test — restart safety relies on Spark's + own checkpointing (a well-established framework guarantee), but there's + no dedicated kill-and-restart test proving it against this exact + pipeline. diff --git a/submission/Shabbirsheikh/pipeline/__init__.py b/submission/Shabbirsheikh/pipeline/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/submission/Shabbirsheikh/pipeline/kafka_producer.py b/submission/Shabbirsheikh/pipeline/kafka_producer.py new file mode 100644 index 0000000..998a17a --- /dev/null +++ b/submission/Shabbirsheikh/pipeline/kafka_producer.py @@ -0,0 +1,61 @@ +""" +CDC transport: every source change is pushed straight to Kafka and we block +until the broker acknowledges it (acks="all") - that ack is the durability +point, same guarantee a WAL gives you, via a message broker instead. + +Kafka's per-partition offset is the sequence number for ordering - the +topic has exactly one partition so offset order is true global order. +""" + +import json +from datetime import datetime, timezone + +from kafka import KafkaProducer + +KNOWN_OPS = ("insert", "update", "delete") +TOPIC = "cdc-events-v4" +BOOTSTRAP_SERVERS = "192.168.1.11:9092" + + +def make_producer(bootstrap_servers: str = BOOTSTRAP_SERVERS) -> KafkaProducer: + return KafkaProducer( + bootstrap_servers=bootstrap_servers, + value_serializer=lambda v: json.dumps(v, default=str).encode("utf-8"), + key_serializer=lambda k: k.encode("utf-8") if k is not None else None, + acks="all", # wait for full broker commit - this is our durability point + ) + + +class KafkaChangeCapture: + """insert/update/delete interface for capturing a change, backed by Kafka.""" + + def __init__(self, producer: KafkaProducer, topic: str = TOPIC): + self.producer = producer + self.topic = topic + + def emit(self, op: str, table: str, pk: str, row: dict) -> dict: + if op not in KNOWN_OPS: + raise ValueError(f"'{op}' is not a change op, expected one of {KNOWN_OPS}") + + payload = { + "operation": op, + "table_name": table, + "primary_key": pk, + # pre-serialize to a JSON string here (not a nested object) so the + # Spark side can treat it as a plain string column and parse it + # per-table later, same pattern the DuckDB lake used + "data": json.dumps(row, default=str), + "captured_at": datetime.now(timezone.utc).isoformat(), + } + future = self.producer.send(self.topic, key=pk, value=payload) + future.get(timeout=10) # blocks until acked - turns "sent" into "durable" + return payload + + def insert(self, table: str, pk: str, row: dict) -> dict: + return self.emit("insert", table, pk, row) + + def update(self, table: str, pk: str, row: dict) -> dict: + return self.emit("update", table, pk, row) + + def delete(self, table: str, pk: str, row: dict) -> dict: + return self.emit("delete", table, pk, row) diff --git a/submission/Shabbirsheikh/pipeline/source_writer.py b/submission/Shabbirsheikh/pipeline/source_writer.py new file mode 100644 index 0000000..41a8b2a --- /dev/null +++ b/submission/Shabbirsheikh/pipeline/source_writer.py @@ -0,0 +1,40 @@ +""" +Writes to the real source.duckdb (so its PK/FK/CHECK/NOT NULL constraints +apply) and only after that succeeds emits the matching event to Kafka - +source write first, capture second, so we never emit an event for something +that didn't actually happen in the source. +""" + +from source.models import EXPECTED_COLUMNS, PRIMARY_KEYS + + +class SourceWriter: + def __init__(self, source_conn, capture): + self.source_conn = source_conn + self.capture = capture + + def insert(self, table: str, pk: str, row: dict) -> dict: + columns = EXPECTED_COLUMNS[table] + placeholders = ", ".join("?" for _ in columns) + values = [row.get(c) for c in columns] + self.source_conn.execute( + f"INSERT INTO {table} ({', '.join(columns)}) VALUES ({placeholders})", values + ) + return self.capture.insert(table, pk, row) + + def update(self, table: str, pk: str, row: dict) -> dict: + columns = EXPECTED_COLUMNS[table] + pk_col = PRIMARY_KEYS[table] + set_cols = [c for c in columns if c != pk_col] + set_clause = ", ".join(f"{c} = ?" for c in set_cols) + values = [row.get(c) for c in set_cols] + [pk] + self.source_conn.execute(f"UPDATE {table} SET {set_clause} WHERE {pk_col} = ?", values) + return self.capture.update(table, pk, row) + + def delete(self, table: str, pk: str, row: dict) -> dict: + # row still carries the last-known values, same as capture.delete() + # expects - the source row is really removed, but Kafka/lake keep + # what it looked like right before deletion + pk_col = PRIMARY_KEYS[table] + self.source_conn.execute(f"DELETE FROM {table} WHERE {pk_col} = ?", [pk]) + return self.capture.delete(table, pk, row) diff --git a/submission/Shabbirsheikh/pipeline/spark_consumer.py b/submission/Shabbirsheikh/pipeline/spark_consumer.py new file mode 100644 index 0000000..0be8c8d --- /dev/null +++ b/submission/Shabbirsheikh/pipeline/spark_consumer.py @@ -0,0 +1,136 @@ +""" +Processing layer: Kafka -> Spark Structured Streaming -> Parquet lake + +DuckDB warehouse. + +Lake: events appended to Parquet via Spark's native streaming sink, +partitioned by day (from captured_at), so a single day can be targeted +directly later without scanning the whole lake. + +Warehouse: DuckDB is single-writer, so Spark executors can't write to it +directly. Each microbatch is instead collected to the driver and handed to +the same apply_lake_events() the test suite already covers - Spark's job +here is just "get Kafka data to the driver, in order". + +Restart-safety comes from Spark's own checkpoint (CHECKPOINT_PATH); the +_cdc_seq guard in apply_lake_events is a second, independent safety net. + +Two run modes: on-demand (default, trigger(availableNow=True), processes +what's in Kafka and exits) or continuous (--continuous, keeps polling and +leaves a Spark driver process resident - stop it when done). +""" + +import sys +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PROJECT_ROOT)) # spark-submit only puts the script's own dir on sys.path + +import duckdb +from pyspark.sql import SparkSession, functions as F, types as T + +from pipeline import warehouse as wh + +KAFKA_BOOTSTRAP = "192.168.1.11:9092" +TOPIC = "cdc-events-v4" + +LAKE_PATH = str(PROJECT_ROOT / "data" / "lake_parquet") +WAREHOUSE_DB = str(PROJECT_ROOT / "data" / "warehouse.duckdb") +CHECKPOINT_PATH = str(PROJECT_ROOT / "data" / "spark_checkpoints" / "main") + +# `data` stays a raw JSON string - apply_lake_events() parses it per-table +ENVELOPE_SCHEMA = T.StructType([ + T.StructField("operation", T.StringType()), + T.StructField("table_name", T.StringType()), + T.StructField("primary_key", T.StringType()), + T.StructField("data", T.StringType()), + T.StructField("captured_at", T.StringType()), +]) + + +def build_spark() -> SparkSession: + return SparkSession.builder.appName("cdc-lakehouse").getOrCreate() + + +def run( + spark: SparkSession | None = None, + continuous: bool = False, + trigger_interval: str = "5 seconds", +): + owns_spark = spark is None + spark = spark or build_spark() + spark.sparkContext.setLogLevel("WARN") + + raw = ( + spark.readStream.format("kafka") + .option("kafka.bootstrap.servers", KAFKA_BOOTSTRAP) + .option("subscribe", TOPIC) + .option("startingOffsets", "earliest") + .load() + ) + + events = raw.select( + F.col("offset").alias("sequence"), + F.from_json(F.col("value").cast("string"), ENVELOPE_SCHEMA).alias("event"), + ).select( + "sequence", + F.col("event.operation").alias("operation"), + F.col("event.table_name").alias("table_name"), + F.col("event.primary_key").alias("primary_key"), + F.col("event.data").alias("data"), + F.col("event.captured_at").alias("captured_at"), + F.to_date(F.to_timestamp(F.col("event.captured_at"))).alias("event_date"), # lake partition key + ) + + bootstrap_conn = duckdb.connect(WAREHOUSE_DB) + wh.create_warehouse_tables(bootstrap_conn) + bootstrap_conn.close() + + def process_batch(batch_df, batch_id: int): + batch_df.persist() + try: + if batch_df.take(1): + ( + batch_df.write + .format("parquet") + .mode("append") + .partitionBy("event_date") + .save(LAKE_PATH) + ) + + rows = [ + r.asDict() + for r in batch_df.drop("event_date").orderBy("sequence").collect() + ] + + # fresh connection per batch, not one held for the job's + # lifetime - DuckDB locks the file for as long as a + # connection is open, and continuous mode can run for hours + wh_conn = duckdb.connect(WAREHOUSE_DB) + try: + wh.apply_lake_events(wh_conn, rows) + finally: + wh_conn.close() + print(f"[batch {batch_id}] applied {len(rows)} event(s)", flush=True) + finally: + batch_df.unpersist() + + writer = ( + events.writeStream + .foreachBatch(process_batch) + .option("checkpointLocation", CHECKPOINT_PATH) + ) + + if continuous: + print(f"Starting CONTINUOUS mode, checking every {trigger_interval} - Ctrl+C to stop", flush=True) + query = writer.trigger(processingTime=trigger_interval).start() + else: + query = writer.trigger(availableNow=True).start() + + query.awaitTermination() + + if owns_spark: + spark.stop() + + +if __name__ == "__main__": + run(continuous="--continuous" in sys.argv) diff --git a/submission/Shabbirsheikh/pipeline/warehouse.py b/submission/Shabbirsheikh/pipeline/warehouse.py new file mode 100644 index 0000000..459d3cc --- /dev/null +++ b/submission/Shabbirsheikh/pipeline/warehouse.py @@ -0,0 +1,101 @@ +""" +Warehouse = latest known state per row, rebuilt by replaying the lake. +No separate write path - the warehouse is always "which lake events have +been applied so far", which keeps it consistent with the lake and makes +rebuild/restore possible. + +Each warehouse table mirrors its source columns plus: + _cdc_seq - sequence of the last lake event applied to this row + _deleted - soft-delete flag (history/restore still needs the row) + +Last-write-wins is enforced with `WHERE excluded._cdc_seq > wh_x._cdc_seq` +on the upsert itself, so an out-of-order or duplicate event can never +clobber a row that's already ahead of it. + +apply_lake_events() is storage-agnostic - it just takes a list of event +dicts, whether they came from Parquet, a Kafka microbatch, or a test. That's +what lets pipeline/spark_consumer.py reuse this exact merge logic. +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +# duckdb is only needed inside reconstruct_as_of() - kept out of the +# module-level imports so this stays importable in the Spark consumer's +# environment, which doesn't have duckdb installed. +if TYPE_CHECKING: + import duckdb + +from source.models import EXPECTED_COLUMNS, PRIMARY_KEYS, COLUMN_TYPES + + +def _warehouse_table(source_table: str) -> str: + return f"wh_{source_table}" + + +def create_warehouse_tables(conn: duckdb.DuckDBPyConnection) -> None: + for table, columns in EXPECTED_COLUMNS.items(): + pk = PRIMARY_KEYS[table] + types = COLUMN_TYPES[table] + col_defs = ", ".join(f"{c} {types[c]}" for c in columns) + conn.execute(f""" + CREATE TABLE IF NOT EXISTS {_warehouse_table(table)} ( + {col_defs}, + _cdc_seq BIGINT NOT NULL, + _deleted BOOLEAN NOT NULL DEFAULT FALSE, + PRIMARY KEY ({pk}) + ) + """) + + +def _bind_expr(sql_type: str) -> str: + # decimals/timestamps round-trip through JSON as plain strings, so they + # need an explicit cast back on the way into the warehouse + if sql_type.startswith("DECIMAL") or sql_type == "TIMESTAMP": + return f"CAST(? AS {sql_type})" + return "?" + + +def apply_lake_events(conn: duckdb.DuckDBPyConnection, lake_rows: list[dict]) -> None: + """lake_rows must be sorted by sequence. Each `data` is a full row + snapshot, not a diff - applying it is always a full-column overwrite.""" + for row in lake_rows: + table = row["table_name"] + if table not in EXPECTED_COLUMNS: + continue # event for a table we don't have a warehouse model for + + columns = EXPECTED_COLUMNS[table] + types = COLUMN_TYPES[table] + pk = PRIMARY_KEYS[table] + data = json.loads(row["data"]) if isinstance(row["data"], str) else row["data"] + is_delete = row["operation"] == "delete" + + insert_cols = columns + ["_cdc_seq", "_deleted"] + values = [data.get(c) for c in columns] + [row["sequence"], is_delete] + + placeholders = ", ".join(_bind_expr(types[c]) for c in columns) + ", ?, ?" + update_clause = ", ".join(f"{c} = excluded.{c}" for c in insert_cols) + wh_table = _warehouse_table(table) + + conn.execute( + f""" + INSERT INTO {wh_table} ({", ".join(insert_cols)}) + VALUES ({placeholders}) + ON CONFLICT ({pk}) DO UPDATE SET {update_clause} + WHERE excluded._cdc_seq > {wh_table}._cdc_seq + """, + values, + ) + + +def reconstruct_as_of(lake_rows_up_to_bound: list[dict]) -> duckdb.DuckDBPyConnection: + """Time travel: build a throwaway warehouse from lake events up to some + bound. Never touches the live warehouse.""" + import duckdb # see the TYPE_CHECKING note up top + + snapshot_conn = duckdb.connect(":memory:") + create_warehouse_tables(snapshot_conn) + apply_lake_events(snapshot_conn, lake_rows_up_to_bound) + return snapshot_conn diff --git a/submission/Shabbirsheikh/scripts/__init__.py b/submission/Shabbirsheikh/scripts/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/submission/Shabbirsheikh/scripts/check_schema_contracts.py b/submission/Shabbirsheikh/scripts/check_schema_contracts.py new file mode 100644 index 0000000..c8d0328 --- /dev/null +++ b/submission/Shabbirsheikh/scripts/check_schema_contracts.py @@ -0,0 +1,112 @@ +""" +Compares whatever the source database actually looks like right now against +EXPECTED_COLUMNS / COLUMN_TYPES in source/models.py. A dropped column, +renamed column, or changed type means the contract is broken - ingestion +should stop rather than keep writing against a wrong assumption about the +schema. + +A brand new column that we don't know about yet is reported too, but it's +not treated as breaking on its own - old code just won't reference it, so +nothing crashes. It's a signal the contract needs updating, not a fail. +""" + +import sys +from pathlib import Path + +import duckdb + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PROJECT_ROOT)) + +from source.models import EXPECTED_COLUMNS, COLUMN_TYPES, init_source_schema + +DEFAULT_DB_PATH = PROJECT_ROOT / "data" / "source.duckdb" + + +class SchemaContractViolation(Exception): + pass + + +def _live_columns(conn: duckdb.DuckDBPyConnection, table: str) -> dict[str, str]: + rows = conn.execute( + """ + SELECT column_name, data_type + FROM information_schema.columns + WHERE table_name = ? + ORDER BY ordinal_position + """, + [table], + ).fetchall() + return dict(rows) + + +def check_table(conn: duckdb.DuckDBPyConnection, table: str) -> list[str]: + """List of human-readable problems for this table. Empty list = compatible.""" + live = _live_columns(conn, table) + if not live: + return [f"{table}: table is missing entirely from the source"] + + expected_cols = EXPECTED_COLUMNS[table] + expected_types = COLUMN_TYPES[table] + problems = [] + + for col in expected_cols: + if col not in live: + problems.append(f"{table}.{col}: column missing (dropped, or renamed to something we don't recognize)") + continue + if live[col] != expected_types[col]: + problems.append(f"{table}.{col}: type changed - expected {expected_types[col]}, found {live[col]}") + + unexpected = sorted(set(live) - set(expected_cols)) + if unexpected: + problems.append(f"{table}: new column(s) {unexpected} not in the contract yet (not breaking, but update EXPECTED_COLUMNS)") + + return problems + + +def check_all(conn: duckdb.DuckDBPyConnection) -> dict[str, list[str]]: + report = {} + for table in EXPECTED_COLUMNS: + problems = check_table(conn, table) + if problems: + report[table] = problems + return report + + +def enforce(conn: duckdb.DuckDBPyConnection) -> None: + """Raises SchemaContractViolation on any missing column or type change. + A purely additive new column does not fail this - see check_table.""" + report = check_all(conn) + breaking = { + table: [p for p in problems if "not breaking" not in p] + for table, problems in report.items() + } + breaking = {t: p for t, p in breaking.items() if p} + + if breaking: + lines = [f" - {p}" for problems in breaking.values() for p in problems] + raise SchemaContractViolation( + "Source schema is incompatible with the expected contract, ingestion stopped:\n" + + "\n".join(lines) + ) + + +if __name__ == "__main__": + DEFAULT_DB_PATH.parent.mkdir(parents=True, exist_ok=True) + conn = duckdb.connect(str(DEFAULT_DB_PATH)) + init_source_schema(conn) # no-op if tables already exist + + try: + enforce(conn) + except SchemaContractViolation as exc: + print(f"SCHEMA CONTRACT VIOLATION\n{exc}", file=sys.stderr) + sys.exit(1) + + report = check_all(conn) # anything left here is non-breaking (new columns) + if report: + print("Schema check passed, but with warnings:") + for table, problems in report.items(): + for p in problems: + print(f" - {p}") + else: + print("Schema check passed - source matches the expected contract.") diff --git a/submission/Shabbirsheikh/scripts/run_data_quality_checks.py b/submission/Shabbirsheikh/scripts/run_data_quality_checks.py new file mode 100644 index 0000000..30764a3 --- /dev/null +++ b/submission/Shabbirsheikh/scripts/run_data_quality_checks.py @@ -0,0 +1,205 @@ +""" +The warehouse tables are typed loosely (see warehouse.py) so the merge logic +stays generic - none of the source's constraints (NOT NULL, CHECK, FK) +exist on the warehouse side. This script puts that parity back: every rule +the source enforces gets re-checked here, so a CDC bug or an out-of-order +write can't silently corrupt the warehouse unnoticed. + +System checks: PK uniqueness, not-null, referential integrity, enums. +Business checks: non-negative amounts, order total vs line items, payment +timing, and (needs the lake's full history, not just latest state) status +transitions that shouldn't be possible. +""" + +import json +import sys +from pathlib import Path + +import duckdb + +PROJECT_ROOT = Path(__file__).resolve().parents[1] + +LAKE_PATH = PROJECT_ROOT / "data" / "lake_parquet" +WAREHOUSE_DB = PROJECT_ROOT / "data" / "warehouse.duckdb" + +# only forward transitions are legal; anything not listed here is a violation +ALLOWED_ORDER_TRANSITIONS = { + "pending": {"paid", "cancelled"}, + "paid": {"shipped", "refunded"}, + "shipped": {"refunded"}, + "cancelled": set(), + "refunded": set(), +} +ALLOWED_PAYMENT_TRANSITIONS = { + "pending": {"settled", "failed"}, + "settled": {"refunded"}, + "failed": set(), + "refunded": set(), +} + + +class DataQualityViolation(Exception): + pass + + +# ---------------------------------------------------------------- system -- + +def check_pk_uniqueness(conn: duckdb.DuckDBPyConnection, table: str, pk: str) -> list[str]: + dupes = conn.execute( + f"SELECT {pk}, COUNT(*) c FROM {table} GROUP BY {pk} HAVING COUNT(*) > 1" + ).fetchall() + return [f"{table}.{pk}={row[0]}: duplicated {row[1]} times" for row in dupes] + + +def check_not_null(conn: duckdb.DuckDBPyConnection, table: str, columns: list[str]) -> list[str]: + problems = [] + for col in columns: + n = conn.execute(f"SELECT COUNT(*) FROM {table} WHERE {col} IS NULL").fetchone()[0] + if n: + problems.append(f"{table}.{col}: {n} row(s) with NULL where a value is required") + return problems + + +def check_referential_integrity( + conn: duckdb.DuckDBPyConnection, child_table: str, fk_col: str, parent_table: str, parent_pk: str +) -> list[str]: + orphans = conn.execute(f""" + SELECT c.{fk_col} FROM {child_table} c + LEFT JOIN {parent_table} p ON c.{fk_col} = p.{parent_pk} + WHERE p.{parent_pk} IS NULL + """).fetchall() + return [f"{child_table}.{fk_col}={row[0]}: no matching {parent_table}.{parent_pk}" for row in orphans] + + +def check_enum_domain(conn: duckdb.DuckDBPyConnection, table: str, col: str, allowed: set[str]) -> list[str]: + placeholders = ", ".join(f"'{v}'" for v in allowed) + bad = conn.execute(f"SELECT DISTINCT {col} FROM {table} WHERE {col} NOT IN ({placeholders})").fetchall() + return [f"{table}.{col}: unexpected value {row[0]!r}, allowed are {sorted(allowed)}" for row in bad] + + +# -------------------------------------------------------------- business -- + +def check_non_negative_amounts(conn: duckdb.DuckDBPyConnection) -> list[str]: + problems = [] + bad_orders = conn.execute("SELECT order_id FROM wh_orders WHERE total_amount < 0").fetchall() + problems += [f"wh_orders.total_amount negative for {r[0]}" for r in bad_orders] + + bad_items = conn.execute( + "SELECT order_item_id FROM wh_order_items WHERE unit_price < 0 OR quantity <= 0" + ).fetchall() + problems += [f"wh_order_items bad quantity/unit_price for {r[0]}" for r in bad_items] + + bad_payments = conn.execute("SELECT payment_id FROM wh_payments WHERE amount <= 0").fetchall() + problems += [f"wh_payments.amount not positive for {r[0]}" for r in bad_payments] + return problems + + +def check_order_totals_match_items(conn: duckdb.DuckDBPyConnection) -> list[str]: + rows = conn.execute(""" + SELECT o.order_id, o.total_amount, COALESCE(SUM(i.quantity * i.unit_price), 0) AS items_total + FROM wh_orders o + LEFT JOIN wh_order_items i ON i.order_id = o.order_id AND i._deleted = FALSE + WHERE o._deleted = FALSE + GROUP BY o.order_id, o.total_amount + HAVING ROUND(o.total_amount, 2) != ROUND(items_total, 2) + """).fetchall() + return [f"order {r[0]}: total_amount={r[1]} but line items sum to {r[2]}" for r in rows] + + +def check_payment_timing(conn: duckdb.DuckDBPyConnection) -> list[str]: + rows = conn.execute(""" + SELECT payment_id FROM wh_payments + WHERE paid_at IS NOT NULL AND paid_at < created_at + """).fetchall() + return [f"wh_payments.{r[0]}: paid_at is before created_at" for r in rows] + + +def _status_history_per_entity(lake_events: list[dict], table: str) -> dict[str, list[str]]: + history: dict[str, list[str]] = {} + for event in sorted(lake_events, key=lambda e: e["sequence"]): + if event["table_name"] != table or event["operation"] == "delete": + continue + data = json.loads(event["data"]) if isinstance(event["data"], str) else event["data"] + status = data.get("status") + if status is not None: + history.setdefault(event["primary_key"], []).append(status) + return history + + +def check_status_transitions(lake_events: list[dict], table: str, allowed: dict[str, set[str]]) -> list[str]: + problems = [] + for pk, statuses in _status_history_per_entity(lake_events, table).items(): + for before, after in zip(statuses, statuses[1:]): + if before == after: + continue # a re-emitted same-status update isn't a transition + if after not in allowed.get(before, set()): + problems.append(f"{table} {pk}: illegal status transition {before} -> {after}") + return problems + + +# ------------------------------------------------------------------ main -- + +def run_all(wh_conn: duckdb.DuckDBPyConnection, lake_events: list[dict]) -> dict[str, list[str]]: + results = { + "pk_uniqueness": ( + check_pk_uniqueness(wh_conn, "wh_customers", "customer_id") + + check_pk_uniqueness(wh_conn, "wh_orders", "order_id") + + check_pk_uniqueness(wh_conn, "wh_order_items", "order_item_id") + + check_pk_uniqueness(wh_conn, "wh_payments", "payment_id") + ), + "not_null": ( + check_not_null(wh_conn, "wh_customers", ["name", "email", "status"]) + + check_not_null(wh_conn, "wh_orders", ["customer_id", "status", "total_amount"]) + + check_not_null(wh_conn, "wh_payments", ["amount", "status"]) + ), + "referential_integrity": ( + check_referential_integrity(wh_conn, "wh_orders", "customer_id", "wh_customers", "customer_id") + + check_referential_integrity(wh_conn, "wh_order_items", "order_id", "wh_orders", "order_id") + + check_referential_integrity(wh_conn, "wh_payments", "order_id", "wh_orders", "order_id") + ), + "enum_domain": ( + check_enum_domain(wh_conn, "wh_customers", "status", {"active", "suspended", "closed"}) + + check_enum_domain(wh_conn, "wh_orders", "status", {"pending", "paid", "shipped", "cancelled", "refunded"}) + + check_enum_domain(wh_conn, "wh_payments", "status", {"pending", "settled", "failed", "refunded"}) + ), + "non_negative_amounts": check_non_negative_amounts(wh_conn), + "order_totals_match_items": check_order_totals_match_items(wh_conn), + "payment_timing": check_payment_timing(wh_conn), + "order_status_transitions": check_status_transitions(lake_events, "orders", ALLOWED_ORDER_TRANSITIONS), + "payment_status_transitions": check_status_transitions(lake_events, "payments", ALLOWED_PAYMENT_TRANSITIONS), + } + return {k: v for k, v in results.items() if v} + + +def enforce(wh_conn: duckdb.DuckDBPyConnection, lake_events: list[dict]) -> None: + failures = run_all(wh_conn, lake_events) + if failures: + lines = [f"[{category}] {p}" for category, problems in failures.items() for p in problems] + raise DataQualityViolation("Data quality checks failed:\n" + "\n".join(f" - {l}" for l in lines)) + + +def _read_lake_events(conn: duckdb.DuckDBPyConnection) -> list[dict]: + """Reads every event out of the Parquet lake, reusing the warehouse + connection since read_parquet() works from any open connection.""" + rows = conn.execute( + f"SELECT * FROM read_parquet('{LAKE_PATH.as_posix()}/*/*.parquet') ORDER BY sequence" + ).fetchall() + cols = [d[0] for d in conn.description] + return [dict(zip(cols, row)) for row in rows] + + +if __name__ == "__main__": + if not WAREHOUSE_DB.exists() or not LAKE_PATH.exists(): + print(f"No warehouse/lake found at {WAREHOUSE_DB} / {LAKE_PATH} - nothing to check yet.") + sys.exit(0) + + wh_conn = duckdb.connect(str(WAREHOUSE_DB)) + lake_events = _read_lake_events(wh_conn) + + try: + enforce(wh_conn, lake_events) + except DataQualityViolation as exc: + print(str(exc), file=sys.stderr) + sys.exit(1) + + print("All data quality checks passed.") diff --git a/submission/Shabbirsheikh/scripts/seed_demo_data.py b/submission/Shabbirsheikh/scripts/seed_demo_data.py new file mode 100644 index 0000000..2291fa1 --- /dev/null +++ b/submission/Shabbirsheikh/scripts/seed_demo_data.py @@ -0,0 +1,108 @@ +""" +End-to-end demo: writes a small, realistic set of changes through +SourceWriter, which does two things per call - runs the real SQL against +source.duckdb (so source's own PK/FK/CHECK/NOT NULL constraints apply) and +publishes the matching event to Kafka. This is the "real" way to generate +data for this pipeline: source, Kafka, lake, and warehouse all end up +consistent with each other, unlike calling KafkaChangeCapture directly +(which only touches Kafka, and was fine for testing pipeline mechanics but +left source.duckdb empty). + +Run pipeline/spark_consumer.py afterward (see README/DESIGN.md for the +docker command) to process these into the lake and warehouse. +""" +import sys +sys.path.insert(0, "/u01/kulu-assignment/data-assignments/submission/Shabbirsheikh") + +import duckdb +from decimal import Decimal +from datetime import datetime, timezone + +from source.models import init_source_schema +from pipeline.kafka_producer import make_producer, KafkaChangeCapture +from pipeline.source_writer import SourceWriter + +SOURCE_DB = "/u01/kulu-assignment/data-assignments/submission/Shabbirsheikh/data/source.duckdb" + +source_conn = duckdb.connect(SOURCE_DB) +init_source_schema(source_conn) + +producer = make_producer() +capture = KafkaChangeCapture(producer) +writer = SourceWriter(source_conn, capture) + +now = datetime.now(timezone.utc) + +# customer 1: Asha, places an order with 2 line items, pays, ships +writer.insert("customers", "c100", { + "customer_id": "c100", "name": "Asha Verma", "email": "asha@example.com", + "status": "active", "created_at": now, "updated_at": now, +}) +writer.insert("orders", "o100", { + "order_id": "o100", "customer_id": "c100", "status": "pending", + "total_amount": Decimal("64.97"), "discount_code": None, + "created_at": now, "updated_at": now, +}) +writer.insert("order_items", "i100", { + "order_item_id": "i100", "order_id": "o100", "sku": "TSHIRT-BLUE-M", + "quantity": 2, "unit_price": Decimal("19.99"), +}) +writer.insert("order_items", "i101", { + "order_item_id": "i101", "order_id": "o100", "sku": "CAP-BLACK", + "quantity": 1, "unit_price": Decimal("24.99"), +}) +writer.insert("payments", "p100", { + "payment_id": "p100", "order_id": "o100", "amount": Decimal("64.97"), + "status": "pending", "created_at": now, "paid_at": None, +}) +writer.update("orders", "o100", { + "order_id": "o100", "customer_id": "c100", "status": "paid", + "total_amount": Decimal("64.97"), "discount_code": None, + "created_at": now, "updated_at": now, +}) +writer.update("payments", "p100", { + "payment_id": "p100", "order_id": "o100", "amount": Decimal("64.97"), + "status": "settled", "created_at": now, "paid_at": now, +}) +writer.update("orders", "o100", { + "order_id": "o100", "customer_id": "c100", "status": "shipped", + "total_amount": Decimal("64.97"), "discount_code": None, + "created_at": now, "updated_at": now, +}) + +# customer 2: Ravi, cancels his order before payment +writer.insert("customers", "c101", { + "customer_id": "c101", "name": "Ravi Kumar", "email": "ravi@example.com", + "status": "active", "created_at": now, "updated_at": now, +}) +writer.insert("orders", "o101", { + "order_id": "o101", "customer_id": "c101", "status": "pending", + "total_amount": Decimal("19.99"), "discount_code": "WELCOME10", + "created_at": now, "updated_at": now, +}) +writer.insert("order_items", "i102", { + "order_item_id": "i102", "order_id": "o101", "sku": "CAP-BLACK", + "quantity": 1, "unit_price": Decimal("19.99"), +}) +writer.update("orders", "o101", { + "order_id": "o101", "customer_id": "c101", "status": "cancelled", + "total_amount": Decimal("19.99"), "discount_code": "WELCOME10", + "created_at": now, "updated_at": now, +}) + +# customer 3: gets soft-deleted (closed account) after being created - +# source row is REALLY deleted, warehouse will show it as _deleted=true +writer.insert("customers", "c102", { + "customer_id": "c102", "name": "Meera Iyer", "email": "meera@example.com", + "status": "active", "created_at": now, "updated_at": now, +}) +writer.delete("customers", "c102", { + "customer_id": "c102", "name": "Meera Iyer", "email": "meera@example.com", + "status": "active", "created_at": now, "updated_at": now, +}) + +producer.flush() +producer.close() +source_conn.close() +print("Seeded demo events through SourceWriter - source.duckdb and Kafka topic 'cdc-events-v4' are now consistent.") +print("Now run pipeline/spark_consumer.py (see README/DESIGN.md for the docker command) to process them into lake + warehouse.") diff --git a/submission/Shabbirsheikh/scripts/validate_catalog.py b/submission/Shabbirsheikh/scripts/validate_catalog.py new file mode 100644 index 0000000..a059569 --- /dev/null +++ b/submission/Shabbirsheikh/scripts/validate_catalog.py @@ -0,0 +1,83 @@ +""" +Makes sure catalog/catalog.json actually reflects what the pipeline +produces - every lake and warehouse dataset must have an entry, and that +entry has to carry enough metadata for someone to actually use it without +reading pipeline code. +""" + +import json +import sys +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PROJECT_ROOT)) + +from source.models import EXPECTED_COLUMNS + +CATALOG_PATH = PROJECT_ROOT / "catalog" / "catalog.json" +REQUIRED_FIELDS = ["name", "layer", "description", "owner", "consumers", "update_cadence", "schema"] + + +class CatalogValidationError(Exception): + pass + + +def load_catalog() -> dict: + with open(CATALOG_PATH) as f: + return json.load(f) + + +def expected_dataset_names() -> set[str]: + return {"lake_cdc_events"} | {f"wh_{table}" for table in EXPECTED_COLUMNS} + + +def validate(catalog: dict) -> list[str]: + problems = [] + datasets = {d.get("name"): d for d in catalog.get("datasets", [])} + + missing = expected_dataset_names() - set(datasets) + for name in sorted(missing): + problems.append(f"no catalog entry for dataset '{name}'") + + for name, entry in datasets.items(): + for field in REQUIRED_FIELDS: + if not entry.get(field): + problems.append(f"'{name}': missing or empty required field '{field}'") + + # warehouse entries should at least document every source column plus the two + # bookkeeping columns the warehouse always adds - catches the catalog going + # stale after a schema change without anyone updating catalog.json to match + for table in EXPECTED_COLUMNS: + wh_name = f"wh_{table}" + entry = datasets.get(wh_name) + if not entry: + continue + documented = set(entry.get("schema", {})) + expected = set(EXPECTED_COLUMNS[table]) | {"_cdc_seq", "_deleted"} + undocumented = expected - documented + if undocumented: + problems.append(f"'{wh_name}': catalog schema is missing column(s) {sorted(undocumented)}") + + return problems + + +def enforce() -> None: + catalog = load_catalog() + problems = validate(catalog) + if problems: + raise CatalogValidationError( + "Catalog validation failed:\n" + "\n".join(f" - {p}" for p in problems) + ) + + +if __name__ == "__main__": + try: + enforce() + except CatalogValidationError as exc: + print(str(exc), file=sys.stderr) + sys.exit(1) + except FileNotFoundError: + print(f"No catalog found at {CATALOG_PATH}", file=sys.stderr) + sys.exit(1) + + print("Catalog is valid - every lake/warehouse dataset is registered with full metadata.") diff --git a/submission/Shabbirsheikh/source.md b/submission/Shabbirsheikh/source.md new file mode 100644 index 0000000..9db7e54 --- /dev/null +++ b/submission/Shabbirsheikh/source.md @@ -0,0 +1,115 @@ +# Source Data Model + +Domain: an order-management slice of e-commerce. Customers place orders; +orders contain line items and are paid via one or more payment attempts. + +Defined in `source/models.py` (`EXPECTED_COLUMNS`, `PRIMARY_KEYS`, +`COLUMN_TYPES`, `init_source_schema`). + +## Entities + +| Table | Kind | Why | +|---|---|---| +| `customers` | Strong | Independent identity/lifecycle, not owned by anything else. | +| `orders` | Strong | Independent identity/lifecycle; references a customer but exists as its own entity with its own status lifecycle. | +| `order_items` | Weak | Lifecycle is entirely tied to its parent order — an order line item has no meaning outside an order. | +| `payments` | Weak | Tied to an order, but more loosely than `order_items` — a payment can arrive or retry after the order was created, so it carries its own timestamps and status independent of the order's. | + +## Schema + +``` +customers + customer_id VARCHAR PRIMARY KEY + name VARCHAR NOT NULL + email VARCHAR NOT NULL + status VARCHAR NOT NULL CHECK (status IN ('active','suspended','closed')) + created_at TIMESTAMP NOT NULL + updated_at TIMESTAMP NOT NULL + +orders + order_id VARCHAR PRIMARY KEY + customer_id VARCHAR NOT NULL REFERENCES customers(customer_id) + status VARCHAR NOT NULL CHECK (status IN ('pending','paid','shipped','cancelled','refunded')) + total_amount DECIMAL(18,2) NOT NULL CHECK (total_amount >= 0) + discount_code VARCHAR NULL + created_at TIMESTAMP NOT NULL + updated_at TIMESTAMP NOT NULL + INDEX idx_orders_customer_id (customer_id) + INDEX idx_orders_status (status) + +order_items + order_item_id VARCHAR PRIMARY KEY + order_id VARCHAR NOT NULL -- see FK note below + sku VARCHAR NOT NULL + quantity INTEGER NOT NULL CHECK (quantity > 0) + unit_price DECIMAL(18,2) NOT NULL CHECK (unit_price >= 0) + INDEX idx_order_items_order_id (order_id) + +payments + payment_id VARCHAR PRIMARY KEY + order_id VARCHAR NOT NULL -- see FK note below + amount DECIMAL(18,2) NOT NULL CHECK (amount > 0) + status VARCHAR NOT NULL CHECK (status IN ('pending','settled','failed','refunded')) + created_at TIMESTAMP NOT NULL + paid_at TIMESTAMP NULL + INDEX idx_payments_order_id (order_id) +``` + +## Why `order_items.order_id` / `payments.order_id` have no `REFERENCES` clause + +This is a deliberate workaround for a real DuckDB engine limitation, not a +modeling gap. DuckDB 1.5.5 cannot `UPDATE` a table that is simultaneously an +FK child (`orders` references `customers`) and an FK parent (`order_items` +and `payments` reference `orders`) — any `UPDATE` on `orders` fails with +"Violates foreign key constraint because key is still referenced," even +when the update doesn't touch `order_id` at all. + +This was confirmed with an isolated minimal repro (a grandparent -> parent +-> child FK chain) before concluding it was an engine bug rather than a +mistake in this schema. Since `orders.status` changes constantly (the most +common write in this domain), `orders` has to stay updatable, so the FK +declaration was dropped on the child side. Referential integrity for this +link is enforced instead in the data-quality layer +(`check_referential_integrity` in `scripts/run_data_quality_checks.py`) — +same as every other business rule that can't be expressed as a +single-table constraint. + +On Postgres or MySQL, this workaround would not be necessary — the FK +would be declared normally. + +## Invariants + +- `orders.total_amount` should reconcile with + `sum(order_items.quantity * order_items.unit_price)` for that order — + checked downstream (`check_order_totals_match_items`), not enforceable + as a single-table `CHECK`. +- `order_items.quantity > 0`, `unit_price >= 0`. +- `payments.amount > 0`. +- `payments.paid_at >= payments.created_at` when `paid_at` is set — + checked downstream (`check_payment_timing`). +- Status fields are restricted to known enum values at the source, and + re-verified downstream (`check_enum_domain`) since the warehouse doesn't + physically enforce the same `CHECK` constraints. +- Status transitions follow a legal state machine per entity — checked + downstream using the lake's full history (`check_status_transitions`), + since this needs the sequence of past statuses, not just the current one: + - orders: `pending -> {paid, cancelled}`, `paid -> {shipped, refunded}`, + `shipped -> {refunded}`; `cancelled`/`refunded` are terminal. + - payments: `pending -> {settled, failed}`, `settled -> {refunded}`; + `failed`/`refunded` are terminal. +- `order_items.order_id` / `payments.order_id` should reference a real + order — not a DuckDB-enforced FK for the reason above, checked instead + in the data-quality layer. + +## Change Patterns + +- `customers`: created once, status updated occasionally (active → + suspended/closed); rarely hard-deleted in practice (soft-deletion via + status is the realistic pattern, though the pipeline supports a real + source-side delete too, captured as a soft-delete in the warehouse). +- `orders`: created once, status updated multiple times over its + lifecycle (this is the highest-frequency write pattern in the domain). +- `order_items`: created with the order, essentially immutable afterward. +- `payments`: created when a payment attempt starts, updated as it + settles/fails; a single order can have more than one payment row + (retries). diff --git a/submission/Shabbirsheikh/source/__init__.py b/submission/Shabbirsheikh/source/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/submission/Shabbirsheikh/source/models.py b/submission/Shabbirsheikh/source/models.py new file mode 100644 index 0000000..4267413 --- /dev/null +++ b/submission/Shabbirsheikh/source/models.py @@ -0,0 +1,173 @@ +""" +Source schema for an e-commerce order-management system. + +Domain: customers place orders; orders contain line items and are paid +via one or more payment attempts. + +Strong entities : customers, orders +Weak entities : order_items (lifecycle tied to its order), payments + (tied to an order, but can arrive/retry after order + creation, so it carries its own timestamps/status) + +Invariants: +- orders.total_amount should reconcile with sum(order_items.quantity * unit_price) + for that order (checked downstream, not enforceable as a single-table CHECK) +- order_items.quantity > 0, unit_price >= 0 +- payments.amount > 0 +- payments.paid_at >= payments.created_at when paid_at is set +- status fields restricted to known enum values +- order_items.order_id / payments.order_id should reference a real order, + but this is NOT a DuckDB-enforced FK (see init_source_schema below for + why - a real DuckDB engine limitation, not an oversight) - checked + instead in the data-quality layer +""" + +from __future__ import annotations + +# duckdb is only needed inside init_source_schema() - kept out of the +# module-level imports so this stays importable without duckdb installed +# (e.g. in the Spark consumer's environment). + +# schema-contract check diffs the live source against this to catch +# dropped/renamed columns before they break anything downstream +EXPECTED_COLUMNS: dict[str, list[str]] = { + "customers": [ + "customer_id", + "name", + "email", + "status", + "created_at", + "updated_at", + ], + "orders": [ + "order_id", + "customer_id", + "status", + "total_amount", + "discount_code", + "created_at", + "updated_at", + ], + "order_items": [ + "order_item_id", + "order_id", + "sku", + "quantity", + "unit_price", + ], + "payments": [ + "payment_id", + "order_id", + "amount", + "status", + "created_at", + "paid_at", + ], +} + +# PK per table - the warehouse layer upserts on this +PRIMARY_KEYS: dict[str, str] = { + "customers": "customer_id", + "orders": "order_id", + "order_items": "order_item_id", + "payments": "payment_id", +} + +# column -> SQL type; the warehouse layer reuses this so amounts/timestamps +# keep their real types instead of collapsing into VARCHAR via JSON +COLUMN_TYPES: dict[str, dict[str, str]] = { + "customers": { + "customer_id": "VARCHAR", + "name": "VARCHAR", + "email": "VARCHAR", + "status": "VARCHAR", + "created_at": "TIMESTAMP", + "updated_at": "TIMESTAMP", + }, + "orders": { + "order_id": "VARCHAR", + "customer_id": "VARCHAR", + "status": "VARCHAR", + "total_amount": "DECIMAL(18,2)", + "discount_code": "VARCHAR", + "created_at": "TIMESTAMP", + "updated_at": "TIMESTAMP", + }, + "order_items": { + "order_item_id": "VARCHAR", + "order_id": "VARCHAR", + "sku": "VARCHAR", + "quantity": "INTEGER", + "unit_price": "DECIMAL(18,2)", + }, + "payments": { + "payment_id": "VARCHAR", + "order_id": "VARCHAR", + "amount": "DECIMAL(18,2)", + "status": "VARCHAR", + "created_at": "TIMESTAMP", + "paid_at": "TIMESTAMP", + }, +} + + +def init_source_schema(conn: "duckdb.DuckDBPyConnection") -> None: + """Create source tables with constraints and indexes on the given connection.""" + + conn.execute(""" + CREATE TABLE IF NOT EXISTS customers ( + customer_id VARCHAR PRIMARY KEY, + name VARCHAR NOT NULL, + email VARCHAR NOT NULL, + status VARCHAR NOT NULL + CHECK (status IN ('active', 'suspended', 'closed')), + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL + ) + """) + + conn.execute(""" + CREATE TABLE IF NOT EXISTS orders ( + order_id VARCHAR PRIMARY KEY, + customer_id VARCHAR NOT NULL REFERENCES customers(customer_id), + status VARCHAR NOT NULL + CHECK (status IN ('pending', 'paid', 'shipped', 'cancelled', 'refunded')), + total_amount DECIMAL(18, 2) NOT NULL CHECK (total_amount >= 0), + discount_code VARCHAR, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL + ) + """) + conn.execute("CREATE INDEX IF NOT EXISTS idx_orders_customer_id ON orders(customer_id)") + conn.execute("CREATE INDEX IF NOT EXISTS idx_orders_status ON orders(status)") + + # order_items.order_id / payments.order_id deliberately skip + # `REFERENCES orders(order_id)`: DuckDB 1.5.5 can't UPDATE a table that's + # both an FK child (orders -> customers) and FK parent (order_items/ + # payments -> orders) - any UPDATE on orders fails, even one that + # doesn't touch order_id. orders.status changes constantly, so orders + # has to stay updatable. Referential integrity for this link is enforced + # in the data-quality layer instead (check_referential_integrity). + conn.execute(""" + CREATE TABLE IF NOT EXISTS order_items ( + order_item_id VARCHAR PRIMARY KEY, + order_id VARCHAR NOT NULL, + sku VARCHAR NOT NULL, + quantity INTEGER NOT NULL CHECK (quantity > 0), + unit_price DECIMAL(18, 2) NOT NULL CHECK (unit_price >= 0) + ) + """) + conn.execute("CREATE INDEX IF NOT EXISTS idx_order_items_order_id ON order_items(order_id)") + + conn.execute(""" + CREATE TABLE IF NOT EXISTS payments ( + payment_id VARCHAR PRIMARY KEY, + order_id VARCHAR NOT NULL, + amount DECIMAL(18, 2) NOT NULL CHECK (amount > 0), + status VARCHAR NOT NULL + CHECK (status IN ('pending', 'settled', 'failed', 'refunded')), + created_at TIMESTAMP NOT NULL, + paid_at TIMESTAMP + ) + """) + conn.execute("CREATE INDEX IF NOT EXISTS idx_payments_order_id ON payments(order_id)") diff --git a/submission/Shabbirsheikh/tests/__init__.py b/submission/Shabbirsheikh/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/submission/Shabbirsheikh/tests/conftest.py b/submission/Shabbirsheikh/tests/conftest.py new file mode 100644 index 0000000..94ad265 --- /dev/null +++ b/submission/Shabbirsheikh/tests/conftest.py @@ -0,0 +1,25 @@ +import duckdb +import pytest + +from source.models import init_source_schema +from pipeline import warehouse +from tests.helpers import EventBuilder + + +@pytest.fixture +def source_conn(): + conn = duckdb.connect(":memory:") + init_source_schema(conn) + return conn + + +@pytest.fixture +def warehouse_conn(): + conn = duckdb.connect(":memory:") + warehouse.create_warehouse_tables(conn) + return conn + + +@pytest.fixture +def change_log(): + return EventBuilder() diff --git a/submission/Shabbirsheikh/tests/helpers.py b/submission/Shabbirsheikh/tests/helpers.py new file mode 100644 index 0000000..5652f56 --- /dev/null +++ b/submission/Shabbirsheikh/tests/helpers.py @@ -0,0 +1,76 @@ +"""Small row builders shared across test files, just to keep test bodies short.""" + +from datetime import datetime, timezone +from decimal import Decimal + + +class EventBuilder: + """Test-only: builds a sequence-numbered event list in the shape + apply_lake_events() expects. Production uses Kafka's own offset as the + sequence instead (see pipeline.kafka_producer.KafkaChangeCapture).""" + + def __init__(self): + self.events: list[dict] = [] + self._next_seq = 1 + + def insert(self, table: str, pk: str, row: dict) -> dict: + return self._emit("insert", table, pk, row) + + def update(self, table: str, pk: str, row: dict) -> dict: + return self._emit("update", table, pk, row) + + def delete(self, table: str, pk: str, row: dict) -> dict: + return self._emit("delete", table, pk, row) + + def _emit(self, operation: str, table: str, pk: str, row: dict) -> dict: + event = { + "sequence": self._next_seq, + "operation": operation, + "table_name": table, + "primary_key": pk, + "data": row, + "captured_at": datetime.now(timezone.utc), + } + self.events.append(event) + self._next_seq += 1 + return event + + +def customer_row(customer_id="c1", status="active", **overrides): + now = datetime.now(timezone.utc) + row = { + "customer_id": customer_id, "name": "Test User", "email": f"{customer_id}@example.com", + "status": status, "created_at": now, "updated_at": now, + } + row.update(overrides) + return row + + +def order_row(order_id="o1", customer_id="c1", status="pending", total_amount="10.00", **overrides): + now = datetime.now(timezone.utc) + row = { + "order_id": order_id, "customer_id": customer_id, "status": status, + "total_amount": Decimal(total_amount), "discount_code": None, + "created_at": now, "updated_at": now, + } + row.update(overrides) + return row + + +def order_item_row(order_item_id="i1", order_id="o1", quantity=1, unit_price="10.00", **overrides): + row = { + "order_item_id": order_item_id, "order_id": order_id, "sku": "SKU1", + "quantity": quantity, "unit_price": Decimal(unit_price), + } + row.update(overrides) + return row + + +def payment_row(payment_id="p1", order_id="o1", amount="10.00", status="pending", paid_at=None, **overrides): + now = datetime.now(timezone.utc) + row = { + "payment_id": payment_id, "order_id": order_id, "amount": Decimal(amount), + "status": status, "created_at": now, "paid_at": paid_at, + } + row.update(overrides) + return row diff --git a/submission/Shabbirsheikh/tests/test_catalog.py b/submission/Shabbirsheikh/tests/test_catalog.py new file mode 100644 index 0000000..372b163 --- /dev/null +++ b/submission/Shabbirsheikh/tests/test_catalog.py @@ -0,0 +1,39 @@ +from scripts.validate_catalog import load_catalog, validate, expected_dataset_names + + +def test_committed_catalog_is_valid(): + """The actual catalog/catalog.json in this repo should always pass.""" + catalog = load_catalog() + assert validate(catalog) == [] + + +def test_every_expected_dataset_is_registered(): + catalog = load_catalog() + registered = {d["name"] for d in catalog["datasets"]} + assert expected_dataset_names() <= registered + + +def test_missing_dataset_is_detected(): + broken = {"datasets": []} + problems = validate(broken) + assert any("lake_cdc_events" in p for p in problems) + assert any("wh_customers" in p for p in problems) + + +def test_incomplete_metadata_is_detected(): + broken = { + "datasets": [ + { + "name": "wh_customers", + "layer": "warehouse", + "description": "", # empty - should be flagged + "owner": "data-platform", + "consumers": ["analytics"], + "update_cadence": "near real-time", + "schema": {"customer_id": "VARCHAR"}, # missing most columns + } + ] + } + problems = validate(broken) + assert any("description" in p for p in problems) + assert any("missing column" in p for p in problems) diff --git a/submission/Shabbirsheikh/tests/test_data_quality.py b/submission/Shabbirsheikh/tests/test_data_quality.py new file mode 100644 index 0000000..eb6040d --- /dev/null +++ b/submission/Shabbirsheikh/tests/test_data_quality.py @@ -0,0 +1,84 @@ +from datetime import timedelta + +import pytest + +from pipeline import warehouse +from scripts.run_data_quality_checks import run_all, enforce, DataQualityViolation + +from tests.helpers import customer_row, order_row, order_item_row, payment_row + + +def _apply(warehouse_conn, change_log): + warehouse.apply_lake_events(warehouse_conn, change_log.events) + + +def test_clean_data_has_no_violations(warehouse_conn, change_log): + change_log.insert("customers", "c1", customer_row("c1")) + change_log.insert("orders", "o1", order_row("o1", "c1", total_amount="20.00")) + change_log.insert("order_items", "i1", order_item_row("i1", "o1", quantity=2, unit_price="10.00")) + _apply(warehouse_conn, change_log) + + assert run_all(warehouse_conn, change_log.events) == {} + + +def test_null_required_field_is_caught(warehouse_conn, change_log): + change_log.insert("customers", "c1", customer_row("c1", name=None)) + _apply(warehouse_conn, change_log) + + failures = run_all(warehouse_conn, change_log.events) + assert "not_null" in failures + + +def test_orphaned_foreign_key_is_caught(warehouse_conn, change_log): + change_log.insert("order_items", "i1", order_item_row("i1", order_id="no-such-order")) + _apply(warehouse_conn, change_log) + + failures = run_all(warehouse_conn, change_log.events) + assert "referential_integrity" in failures + + +def test_unexpected_enum_value_is_caught(warehouse_conn, change_log): + change_log.insert("orders", "o1", order_row("o1")) + change_log.insert("payments", "p1", payment_row("p1", "o1", status="not_a_real_status")) + _apply(warehouse_conn, change_log) + + failures = run_all(warehouse_conn, change_log.events) + assert "enum_domain" in failures + + +def test_order_total_mismatch_with_items_is_caught(warehouse_conn, change_log): + change_log.insert("orders", "o1", order_row("o1", total_amount="999.00")) + change_log.insert("order_items", "i1", order_item_row("i1", "o1", quantity=1, unit_price="10.00")) + _apply(warehouse_conn, change_log) + + failures = run_all(warehouse_conn, change_log.events) + assert "order_totals_match_items" in failures + + +def test_payment_before_order_creation_is_caught(warehouse_conn, change_log): + row = payment_row("p1", "o1") + row["paid_at"] = row["created_at"] - timedelta(days=1) + change_log.insert("orders", "o1", order_row("o1")) + change_log.insert("payments", "p1", row) + _apply(warehouse_conn, change_log) + + failures = run_all(warehouse_conn, change_log.events) + assert "payment_timing" in failures + + +def test_illegal_status_transition_is_caught(warehouse_conn, change_log): + change_log.insert("orders", "o1", order_row("o1", status="pending")) + change_log.update("orders", "o1", order_row("o1", status="cancelled")) + change_log.update("orders", "o1", order_row("o1", status="paid")) # cancelled -> paid is illegal + _apply(warehouse_conn, change_log) + + failures = run_all(warehouse_conn, change_log.events) + assert "order_status_transitions" in failures + + +def test_enforce_raises_when_anything_fails(warehouse_conn, change_log): + change_log.insert("customers", "c1", customer_row("c1", name=None)) + _apply(warehouse_conn, change_log) + + with pytest.raises(DataQualityViolation): + enforce(warehouse_conn, change_log.events) diff --git a/submission/Shabbirsheikh/tests/test_schema_contracts.py b/submission/Shabbirsheikh/tests/test_schema_contracts.py new file mode 100644 index 0000000..34f26f1 --- /dev/null +++ b/submission/Shabbirsheikh/tests/test_schema_contracts.py @@ -0,0 +1,62 @@ +import duckdb +import pytest + +from scripts.check_schema_contracts import check_all, enforce, SchemaContractViolation + +# Building each scenario as a standalone in-memory schema (rather than ALTERing +# a real table) sidesteps a real DuckDB limitation: ALTER TABLE DROP/RENAME +# COLUMN refuses to run once a table has an index or is referenced by a FK, +# which ours always are. A live source doing a real migration wouldn't have +# that restriction, so this is the more portable way to prove the check works. +ORDERS_OK = "CREATE TABLE orders (order_id VARCHAR, customer_id VARCHAR, status VARCHAR, total_amount DECIMAL(18,2), discount_code VARCHAR, created_at TIMESTAMP, updated_at TIMESTAMP)" +ITEMS_OK = "CREATE TABLE order_items (order_item_id VARCHAR, order_id VARCHAR, sku VARCHAR, quantity INTEGER, unit_price DECIMAL(18,2))" +PAYMENTS_OK = "CREATE TABLE payments (payment_id VARCHAR, order_id VARCHAR, amount DECIMAL(18,2), status VARCHAR, created_at TIMESTAMP, paid_at TIMESTAMP)" +CUSTOMERS_OK = "CREATE TABLE customers (customer_id VARCHAR, name VARCHAR, email VARCHAR, status VARCHAR, created_at TIMESTAMP, updated_at TIMESTAMP)" + + +def _conn_with_customers(customers_ddl: str) -> duckdb.DuckDBPyConnection: + conn = duckdb.connect(":memory:") + conn.execute(customers_ddl) + conn.execute(ORDERS_OK) + conn.execute(ITEMS_OK) + conn.execute(PAYMENTS_OK) + return conn + + +def test_fully_compatible_schema_passes_clean(): + conn = _conn_with_customers(CUSTOMERS_OK) + enforce(conn) # must not raise + assert check_all(conn) == {} + + +def test_dropped_column_is_detected(): + conn = _conn_with_customers( + "CREATE TABLE customers (customer_id VARCHAR, email VARCHAR, status VARCHAR, created_at TIMESTAMP, updated_at TIMESTAMP)" + ) + with pytest.raises(SchemaContractViolation, match="name"): + enforce(conn) + + +def test_renamed_column_is_detected_as_missing(): + conn = _conn_with_customers( + "CREATE TABLE customers (customer_id VARCHAR, name VARCHAR, email_address VARCHAR, status VARCHAR, created_at TIMESTAMP, updated_at TIMESTAMP)" + ) + with pytest.raises(SchemaContractViolation, match="email"): + enforce(conn) + + +def test_changed_type_is_detected(): + conn = _conn_with_customers( + "CREATE TABLE customers (customer_id VARCHAR, name VARCHAR, email VARCHAR, status INTEGER, created_at TIMESTAMP, updated_at TIMESTAMP)" + ) + with pytest.raises(SchemaContractViolation, match="type changed"): + enforce(conn) + + +def test_additive_column_does_not_break_ingestion(): + conn = _conn_with_customers( + "CREATE TABLE customers (customer_id VARCHAR, name VARCHAR, email VARCHAR, status VARCHAR, created_at TIMESTAMP, updated_at TIMESTAMP, loyalty_tier VARCHAR)" + ) + enforce(conn) # must not raise - purely additive is not breaking + report = check_all(conn) + assert "loyalty_tier" in str(report) # still surfaced as a warning though diff --git a/submission/Shabbirsheikh/tests/test_warehouse.py b/submission/Shabbirsheikh/tests/test_warehouse.py new file mode 100644 index 0000000..6cb4545 --- /dev/null +++ b/submission/Shabbirsheikh/tests/test_warehouse.py @@ -0,0 +1,54 @@ +from pipeline import warehouse + +from tests.helpers import customer_row, order_row + + +def test_apply_lake_events_upserts_to_latest_state(warehouse_conn, change_log): + change_log.insert("orders", "o1", order_row(status="pending")) + change_log.update("orders", "o1", order_row(status="paid")) + + warehouse.apply_lake_events(warehouse_conn, change_log.events) + + row = warehouse_conn.execute( + "SELECT status, _cdc_seq FROM wh_orders WHERE order_id = 'o1'" + ).fetchone() + assert row == ("paid", 2) + + +def test_stale_out_of_order_write_cannot_overwrite_a_newer_row(warehouse_conn, change_log): + change_log.insert("orders", "o1", order_row(status="pending")) + change_log.update("orders", "o1", order_row(status="paid")) + warehouse.apply_lake_events(warehouse_conn, change_log.events) + + stale_event = dict(change_log.events[0]) # the original "pending" insert, sequence 1 + warehouse.apply_lake_events(warehouse_conn, [stale_event]) # already applied - must not undo the seq-2 "paid" write + + status = warehouse_conn.execute( + "SELECT status FROM wh_orders WHERE order_id = 'o1'" + ).fetchone()[0] + assert status == "paid" + + +def test_delete_soft_deletes_instead_of_removing_the_row(warehouse_conn, change_log): + change_log.insert("customers", "c1", customer_row()) + change_log.delete("customers", "c1", customer_row()) + + warehouse.apply_lake_events(warehouse_conn, change_log.events) + + row = warehouse_conn.execute( + "SELECT customer_id, _deleted FROM wh_customers" + ).fetchone() + assert row == ("c1", True) + + +def test_reconstruct_as_of_shows_state_before_a_later_change(change_log): + change_log.insert("orders", "o1", order_row(status="pending")) + change_log.update("orders", "o1", order_row(status="paid")) + + bounded_events = [e for e in change_log.events if e["sequence"] <= 1] + past_conn = warehouse.reconstruct_as_of(bounded_events) + + status = past_conn.execute( + "SELECT status FROM wh_orders WHERE order_id = 'o1'" + ).fetchone()[0] + assert status == "pending"