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

[Shabbirsheikh]

Summary

A CDC pipeline for an e-commerce order-management domain (customers,
orders, order_items, payments) that keeps a Parquet lake and a
DuckDB warehouse synchronized via Kafka + Spark Structured Streaming.
Full design rationale, trade-offs, and what's been verified live are in
submission/Shabbirsheikh/DESIGN.md
— this description is the condensed version.

Source Schema Design

• Domain: order-management e-commerce.
• Strong entities: customers, orders — independent identity/lifecycle.
• Weak entities: order_items (lifecycle entirely tied to its order),
  payments (tied to an order but looser — can arrive/retry after order
  creation, so it carries its own status/timestamps).
• Keys/relationships/indexes: PK per table, orders.customer_id is a
  real FK to customers; order_items.order_id/payments.order_id are
  deliberately not DB-level FKs — a real DuckDB 1.5.5 engine limitation
  (confirmed via an isolated repro) prevents UPDATEing a table that's
  simultaneously an FK child and FK parent, and orders.status changes
  constantly. That referential link is enforced in the data-quality layer
  instead. Indexes on every FK-shaped lookup column.
• Validation rules: enum-constrained status fields, non-negative
  amounts, order total reconciling with line items, payment timing,
  legal status-transition state machines — full list in source.md.

CDC Strategy

• Capture: SourceWriter writes real SQL to source.duckdb first,
  then — only after that succeeds — publishes to Kafka. This is
  application-level dual-write (simulated CDC), since DuckDB has no
  WAL/logical-replication slot to tail. Known risk: a crash between the
  two writes loses that event; a real Postgres source would use Debezium
  tailing the WAL instead.
• Replay/restart: Spark Structured Streaming's own checkpointLocation
  tracks processed Kafka offsets. I haven't run a dedicated
  kill-and-restart test against this exact pipeline, so I'm flagging that
  as reliance on a well-established framework guarantee rather than
  something I've independently proven here.
• Duplicates: the warehouse merge upserts with
  WHERE excluded._cdc_seq > current._cdc_seq, so a duplicate or
  out-of-order event can never overwrite a row already ahead of it. The
  lake itself isn't strictly exactly-once at the storage layer (a retried
  Spark batch can duplicate a Parquet row), but that's harmless for
  warehouse correctness.
• Deletes: captured with the row's last-known values; the warehouse
  soft-deletes (_deleted = TRUE) rather than removing the row.

Lake and Warehouse Modeling

• Lake: append-only Parquet, day-partitioned
  (data/lake_parquet/event_date=YYYY-MM-DD/), written by Spark's native
  streaming sink. Nothing here is ever updated or deleted.
• Warehouse: DuckDB tables mirroring source columns plus _cdc_seq
  and _deleted. No independent write path — always a function of "which
  lake events have been applied," via one function, apply_lake_events().
• Time travel: reconstruct_as_of(lake_rows_up_to_bound) replays a
  caller-bounded event list into a throwaway in-memory warehouse, never
  touching the live one. Verified live: for an order that moved
  pending -> paid -> shipped across three real Kafka offsets, bounding
  the replay to each earlier offset correctly reproduced pending and
  paid while the live warehouse correctly showed shipped.

Schema Change Safety

• Detection: check_schema_contracts.py diffs the live source schema
  against an expected-columns/types contract.
• Stop behavior: a breaking change (dropped/renamed column, changed
  type) raises SchemaContractViolation; ingestion should stop rather
  than write against a wrong assumption.
• Surfacing: caught at the top level, printed, non-zero exit. A purely
  additive column is a warning, not a failure. (Known gap: nothing
  automated currently watches for that exit code — see Limitations.)

Validation Parity

• System checks: PK uniqueness, not-null, referential integrity
  (including the link DuckDB itself couldn't enforce), enum domain.
• Business checks: non-negative amounts, order-total-matches-line-items,
  payment timing, and status-transition legality (needs the lake's full
  history per entity, not just the warehouse's latest state).
• Failure reporting: DataQualityViolation lists every failing
  category and row; non-zero exit, same pattern as the schema check.

Catalog Exposure

catalog/catalog.json registers 5 datasets (the lake plus four warehouse
tables), each with owner, intended consumers, update cadence, and full
schema. validate_catalog.py enforces every dataset has a complete entry
and that documented schemas haven't drifted from the code. A consumer
discovers datasets by reading catalog.json and queries the lake via
read_parquet(...) or the warehouse by connecting directly to
warehouse.duckdb.

Validation

python -m pytest -q                          # 21 tests, in-memory
python scripts/check_schema_contracts.py
python scripts/run_data_quality_checks.py
python scripts/validate_catalog.py
python scripts/seed_demo_data.py             # then run pipeline/spark_consumer.py via Docker

Full setup/run instructions: submission/Shabbirsheikh/README.md.

I've run every command above against this exact code: 21/21 tests pass,
schema contract passes, data quality checks pass with zero violations,
catalog validation passes.

Known Limitations / Next Steps

1. Real CDC (Debezium/Postgres WAL) instead of simulated dual-write.
2. Monitoring/alerting — failures currently just exit non-zero.
3. Orchestration (Airflow) instead of a manual Docker command.
4. Multi-partition Kafka for real-volume throughput.
5. A concurrent-capable production warehouse instead of DuckDB.
6. Richer, schema-registry-based schema-evolution handling.
7. No automated restart-recovery test — the most concrete gap given the
   assignment's testing requirements call for it directly.

Responsible AI Usage

I used Claude during implementation to speed up writing boilerplate and
draft code. The domain choice, the build order, and the trade-offs
(DuckDB over Postgres, on-demand as the default run mode) 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. The DuckDB FK limitation was confirmed with an isolated repro
before I accepted it as a real engine bug rather than a schema mistake.

Author Checklist

• Tests pass (21/21)
• Schema compatibility checks pass
• Data quality validations pass
• Catalog metadata validation passes
• End-to-end CDC flow validated (source → Kafka → Spark → lake + warehouse)