diff --git a/DATA_ASSIGNMENT.md b/DATA_ASSIGNMENT.md deleted file mode 100644 index a33191d..0000000 --- a/DATA_ASSIGNMENT.md +++ /dev/null @@ -1,547 +0,0 @@ -# CDC Lakehouse Reliability Assignment — Data Engineering - -## Overview - -Design and implement a small **change data capture (CDC) pipeline** that keeps a data lake and a warehouse synchronized from a relational source system. - -This assignment is intended to evaluate senior data engineering candidates on the kind of problems that matter in real systems: - -- schema and entity modeling -- CDC correctness -- durability and replayability -- near real-time propagation -- handling schema drift and incompatible source changes -- reproducibility and operational safety -- consistency between source validations and downstream models -- historical recovery and point-in-time restore - -This is intentionally **not** a toy ETL exercise. The goal is to assess whether you can design a data platform workflow that behaves safely when data changes, schemas evolve, and downstream consumers require both **full history** and **current-state analytics**. - -Focus on **clarity, correctness, and robustness** over breadth. - ---- - -## Problem Statement - -You are given a transactional source system with **3 to 10 tables**. You should design the source schema yourself, but it must contain: - -- a mix of strong and weak entities -- relationships between tables -- indexes -- a mix of data types, including at least: - - currency / decimal - - dates / timestamps - - enum-like fields - - identifiers / foreign keys - - nullable fields where appropriate - -Your task is to build a CDC-based pipeline that: - -1. captures all source changes -2. writes every change to a **lake** -3. keeps a **warehouse** updated for near real-time availability -4. stops ingestion and warns users if source schema changes incompatibly -5. preserves enough history to move backward in time and restore data via the warehouse -6. mirrors source system and business validations in the warehouse models -7. exposes both the lake and warehouse datasets in a **catalog** for access - -You may choose the domain, but it should resemble a realistic transactional system rather than a flat reporting dataset. - -Examples of suitable domains: - -- wallet / payments / transfers -- e-commerce orders and line items -- ride-hailing trips and settlements -- subscriptions and invoices -- ticketing and reservations - ---- - -## Time Expectation - -Please spend **3–5 hours** on this assignment. - -We do not expect a production-complete data platform. We care more about: - -- correct reasoning -- sensible modeling -- reliable CDC design -- explicit assumptions and tradeoffs -- reproducibility -- clear documentation - -A smaller, well-reasoned solution is better than a broad but shallow one. - ---- - -## Core Requirements - -### 1) Source Data Model - -Design a source relational model with **3 to 10 tables**. - -Your model must include: - -- strong entities -- weak entities -- one-to-many or many-to-one relationships -- appropriate primary keys and foreign keys -- indexes where lookup or change-capture performance would reasonably matter -- a mix of data types such as: - - amount / currency - - date / timestamp - - enum or status - - nullable optional attributes - -Examples of strong entities: -- customer -- order -- wallet -- merchant - -Examples of weak entities: -- order_item -- wallet_balance_history -- payment_attempt -- reservation_hold_event - -You should document: - -- why the entities were chosen -- entity relationships -- what makes a table strong vs weak in your model -- important invariants -- expected change patterns - -### 2) CDC Pipeline - -Build a CDC flow from the source into both a **lake** and a **warehouse**. - -Required behavior: - -- all inserts, updates, and deletes in the source must be captured -- the **lake** must retain **every data change** -- the **warehouse** must reflect the **latest snapshot** -- the warehouse should become available in **near real time** -- the design must support replay and recovery - -You may choose the exact implementation approach, for example: - -- database CDC log / WAL-based ingestion -- timestamp/version-based incremental extraction -- simulated CDC if necessary for local scope - -If you simulate CDC, document what is simulated and what would change in a production setup. - -### 3) Schema Change Detection and Safe Stop - -Your pipeline must explicitly detect source schema changes. - -Assume there is **no backward compatibility guarantee** from source systems. - -Required behavior: - -- if an incompatible source schema change is detected, ingestion must stop -- the system must emit a clear warning / failure signal -- the lake and warehouse should not continue ingesting silently under broken assumptions -- the detection strategy must be documented clearly - -Examples of incompatible changes: - -- dropped column -- renamed column -- changed data type -- changed enum domain -- nullable to non-nullable change where current logic breaks -- changed primary/foreign-key assumptions - -You do not need to solve automatic migration of breaking changes. We care that you fail safely and observably. - -### 4) Historical Recovery and Time Travel - -We need the ability to move backward in time and restore data via the warehouse. - -Your design should explain and, where practical, demonstrate: - -- how historical states are preserved -- how a prior point in time can be reconstructed -- how restore / rollback would work operationally -- what data model in the warehouse enables this - -A strong solution will distinguish clearly between: - -- event/change history in the lake -- current-state analytical snapshot in the warehouse -- historical reconstruction support via versioned or temporal modeling - -### 5) Validation Parity - -All important source validations should exist in the warehouse as well. - -This includes both: - -- **system validations** - - primary key uniqueness - - referential integrity - - not-null rules - - valid types / domains - -- **business validations** - - allowed status transitions - - non-negative monetary amounts where required - - line item totals matching order totals, if modeled - - settlement dates not preceding creation dates, if modeled - -Document: - -- which validations exist in the source -- how the same validations are asserted or tested in the warehouse -- how validation failures are surfaced - -### 6) Catalog Exposure - -Expose both the lake and warehouse datasets in a **catalog**. - -You may implement this minimally for the scope of the assignment, but the design should make access paths clear. - -At minimum, document: - -- dataset names / schemas -- which datasets are lake vs warehouse -- who they are intended for -- how a consumer discovers and queries them -- what metadata is published - -If you implement a lightweight local catalog approach, explain the production analogue. - ---- - -## Data Platform Expectations - -We expect a clean separation of concerns across your data platform design. - -A strong submission will usually make the following distinctions clear: - -### Source Layer -- transactional source schema -- authoritative system of record -- source validations and constraints - -### Ingestion / CDC Layer -- change capture logic -- schema change detection -- replay / checkpointing strategy -- failure handling - -### Lake Layer -- immutable or append-oriented change history -- raw or normalized change records -- enough detail to reconstruct every change - -### Warehouse Layer -- curated latest-state snapshot -- analytical tables or models -- versioned history strategy if used for restore / time travel - -### Quality / Validation Layer -- tests, constraints, or assertions -- parity checks with source assumptions -- data contract or schema checks - -### Catalog / Access Layer -- discoverability -- published metadata -- access boundaries or intended consumers - ---- - -## Documentation-First Workflow - -Before writing implementation code, document the behavior of your system. - -At minimum, include: - -- source schema -- CDC contracts -- assumptions about change capture -- lake data model -- warehouse data model -- schema evolution policy -- stop-the-line behavior for incompatible changes -- time-travel / restore strategy -- validation strategy -- catalog exposure strategy - -We are intentionally looking for candidates who can make the non-happy-path behavior of a data platform explicit before building it. - ---- - -## Functional Expectations - -Your submission should include the following artifacts or equivalents. - -### A) Source Schema Definition -Provide DDL, migration files, model definitions, or equivalent. - -Should include: -- keys -- constraints -- indexes -- representative enum/status fields -- sample data if helpful - -### B) CDC Design / Implementation -Provide code, SQL, pipeline definitions, orchestration, or scripts that show how: - -- source changes are detected -- changes are written to the lake -- latest-state data is published to the warehouse -- checkpoints / offsets / extraction boundaries are managed -- schema changes are detected and handled safely - -### C) Lake Output -The lake should preserve **every change**. - -Examples of acceptable representations: -- append-only change table -- bronze/raw CDC records -- JSON or parquet change files -- event log with operation type and timestamps - -The important point is that change history is durable and queryable. - -### D) Warehouse Output -The warehouse should represent the **latest snapshot** for downstream use. - -A strong solution may also include: -- SCD2 or temporal history support -- current-state dimensions/facts -- versioned snapshot tables -- restore-oriented modeling - -### E) Validation Layer -Include tests, assertions, expectations, or SQL checks that demonstrate: - -- system constraint parity -- business rule parity -- freshness / completeness checks where relevant -- schema compatibility checks - -### F) Catalog Metadata -Provide a minimal catalog artifact, metadata file, registry entry, or documented equivalent that exposes: - -- lake datasets -- warehouse datasets -- owners / intended users -- schema and description -- update cadence - ---- - -## Reliability Requirements - -Your design should explicitly account for the following: - -- duplicate CDC events -- retries after partial failure -- out-of-order change arrival where relevant -- ingestion restart after checkpointed progress -- incompatible schema changes -- deletes in the source -- late-arriving updates -- recovery from historical data - -We are not testing for a perfect enterprise platform. We are testing whether you can reason carefully about correctness and safe operations. - ---- - -## Architecture Expectations - -Use a clean, layered approach. - -We expect to see something close to the following, adapted to your chosen stack: - -### 1) Source Modeling Layer -- DDL / source entities / seed data -- explicit constraints and indexes - -### 2) CDC / Ingestion Layer -- extraction or log consumption logic -- checkpointing / offsets -- schema compatibility checks - -### 3) Lake Writing Layer -- durable append-only change persistence -- full change retention - -### 4) Warehouse Modeling Layer -- current-state snapshot logic -- historical restore strategy -- downstream-friendly schema - -### 5) Validation / Quality Layer -- data tests -- rule assertions -- schema checks - -### 6) Catalog Layer -- dataset registration / metadata exposure - -You do not need an elaborate platform framework, but the flow should be easy to reason about and maintain. - ---- - -## Testing Requirements - -We expect tests or validation steps that prove platform behavior, not only happy-path execution. - -### Required Testing Categories - -#### 1) Modeling and Constraint Tests -Cover: -- key constraints -- relationship integrity -- index assumptions where relevant -- source validation rules - -#### 2) CDC Correctness Tests -Cover: -- insert capture -- update capture -- delete capture -- duplicate or replay safety -- restart/recovery behavior - -#### 3) Schema Change Safety Tests -Cover: -- incompatible schema change detection -- ingestion stop behavior -- warning / failure signal emitted - -#### 4) Warehouse Correctness Tests -Cover: -- latest snapshot correctness -- validation parity with source -- historical reconstruction or restore logic - -#### 5) Catalog Tests or Validation -Cover: -- catalog entries exist for lake and warehouse datasets -- schema / metadata exposure is correct - -### Red, Blue, Green Discipline - -Please follow a **Red, Blue, Green** workflow: - -- **Red**: write a failing test or failing assertion for a required behavior -- **Blue**: implement the minimum necessary logic to make it pass -- **Green**: refactor for clarity, reliability, and maintainability while preserving behavior - -You do not need to submit every intermediate step, but your solution should reflect disciplined iteration. - ---- - -## Technology Choices - -You may use any language, framework, orchestration tool, or local data stack you are comfortable with. - -Examples include: -- Python / SQL -- dbt -- Spark -- Airflow / Dagster / Prefect -- PostgreSQL / DuckDB / ClickHouse / BigQuery-like local analogues -- Delta / Iceberg / Hudi-like local analogues -- simple scripts with well-structured SQL models - -Please choose a stack that makes your correctness story easy to understand. - -If you make simplifying assumptions, document them. - ---- - -## Non-Goals - -You do **not** need to build: - -- a full production deployment platform -- complex IAM or security controls -- a polished UI -- enterprise-scale orchestration -- exhaustive BI dashboards - -Keep the scope tight. Depth is more important than breadth. - ---- - -## Deliverables - -Please submit your solution as a **Pull Request**. - -Your PR should include: - -- source schema definitions -- CDC implementation or simulation -- lake and warehouse outputs/models -- validation tests/checks -- catalog metadata or equivalent -- setup instructions -- how to run validations/tests -- a short architecture/design explanation -- assumptions, tradeoffs, and limitations - -### PR Description Requirements - -Your PR description must explicitly explain: - -1. **Source schema design** - - tables - - strong vs weak entities - - keys - - relationships - - indexes - - validation rules - -2. **CDC strategy** - - how changes are captured - - how replay/restart works - - how duplicates are handled - - how deletes are handled - -3. **Lake and warehouse modeling** - - how the lake captures every change - - how the warehouse maintains latest snapshot - - how time travel / restore is supported - -4. **Schema change safety** - - how incompatible changes are detected - - how ingestion is stopped - - how warnings/failures are surfaced - -5. **Validation parity** - - how source validations are represented downstream - - how failures are checked and reported - -6. **Catalog exposure** - - how lake and warehouse datasets are published/discovered - -7. **Responsible AI usage** - - whether you used AI tools - - where they helped - - what you personally reviewed, validated, or corrected - -Please be candid. AI usage is allowed, but we care about engineering judgment, not generated volume. - ---- - -## What We Are Optimizing For - -A strong submission is one that is: - -- correct under replay, restart, and schema change -- explicit about invariants and tradeoffs -- easy to reason about -- layered cleanly -- backed by meaningful tests and validations - -A smaller but robust solution is preferred over a broader but fragile one. diff --git a/README.md b/README.md index cdaca35..cbd5363 100644 --- a/README.md +++ b/README.md @@ -1,50 +1,656 @@ -# Data Engineering Assignment +# CDC Data Platform Assignment ## Overview -This repository contains the take-home assignment for data engineering candidates at Kulu. Full assignment details, requirements, and evaluation criteria are documented in [DATA_ASSIGNMENT.md](./DATA_ASSIGNMENT.md). +This project implements a CDC-based data platform for an e-commerce transactional system using: + +* PySpark +* PostgreSQL (simulated locally using Spark datasets) +* Apache Airflow +* Parquet Data Lake +* SCD Type 2 Warehouse Modeling + +The platform captures all source changes, stores immutable history in a data lake, maintains a near real-time analytical warehouse, detects incompatible schema changes, supports historical recovery, enforces validation parity, and exposes datasets through a catalog. --- -## How to Submit +# Architecture + +```text +Source System +(PostgreSQL) + + | + v + +Schema Validation +(Fail Fast) + + | + v + +CDC Capture +(WAL Based CDC) + + | + v + +Bronze Layer +Immutable CDC Events + + | + v + +Silver Layer +Latest Operational Snapshot + + | + v + +Warehouse Layer + + - dim_customer (SCD2) + - dim_product (SCD2) -### 1. Fork this repository + - fact_orders + - fact_order_items + - fact_payment_attempts -Click the **Fork** button at the top-right of this GitHub repository page to create your own copy under your GitHub account. + | + v -### 2. Clone your fork +Data Quality Checks -```bash -git clone https://github.com//data-assignments.git -cd data-assignments + | + v + +Catalog Publication ``` -### 3. Complete the assignment +--- + +# Domain + +The chosen domain is an e-commerce platform. + +The model contains both strong and weak entities. + +## Strong Entities + +### Customers + +Stores customer information. + +### Products + +Stores product catalog information. + +### Orders + +Represents customer purchases. + +--- -Work on your solution inside your forked repository. Refer to [DATA_ASSIGNMENT.md](./DATA_ASSIGNMENT.md) for the full problem statement, requirements, and evaluation criteria. +## Weak Entities -### 4. Push your changes +### Order Items -Commit and push your work to your fork: +Dependent on Orders. + +Represents individual products within an order. + +### Payment Attempts + +Dependent on Orders. + +Represents payment gateway attempts and outcomes. + +--- + +# Source Data Model + +## customers + +| Column | Type | +| --------------- | ----------------- | +| customer_id | bigint | +| email | string | +| first_name | string | +| last_name | string | +| phone | string (nullable) | +| customer_status | enum | +| created_at | timestamp | +| updated_at | timestamp | + +Indexes: + +* PK(customer_id) +* email + +--- + +## products + +| Column | Type | +| -------------- | ------------- | +| product_id | bigint | +| sku | string | +| product_name | string | +| category | string | +| unit_price | decimal(18,2) | +| product_status | enum | +| created_at | timestamp | +| updated_at | timestamp | + +Indexes: + +* PK(product_id) +* sku + +--- + +## orders + +| Column | Type | +| ------------ | ------------- | +| order_id | bigint | +| customer_id | bigint | +| order_status | enum | +| total_amount | decimal(18,2) | +| created_at | timestamp | +| updated_at | timestamp | + +Indexes: + +* PK(order_id) +* customer_id + +Foreign Keys: + +* customer_id -> customers.customer_id + +--- -```bash -git add . -git commit -m "Submit assignment" -git push origin main +## order_items + +| Column | Type | +| ------------- | ------------- | +| order_item_id | bigint | +| order_id | bigint | +| product_id | bigint | +| quantity | integer | +| unit_price | decimal(18,2) | +| line_amount | decimal(18,2) | +| created_at | timestamp | + +Indexes: + +* PK(order_item_id) +* order_id +* product_id + +Foreign Keys: + +* order_id -> orders.order_id +* product_id -> products.product_id + +--- + +## payment_attempts + +| Column | Type | +| ---------------------- | ------------- | +| payment_attempt_id | bigint | +| order_id | bigint | +| payment_status | enum | +| gateway_transaction_id | string | +| amount | decimal(18,2) | +| attempt_timestamp | timestamp | + +Indexes: + +* PK(payment_attempt_id) +* order_id + +Foreign Keys: + +* order_id -> orders.order_id + +--- + +# Entity Relationships + +```text +Customer + | + | 1:M + | +Orders + | + | 1:M + | +Order Items + | + | M:1 + | +Products + +Orders + | + | 1:M + | +Payment Attempts +``` + +--- + +# Important Business Invariants + +1. Customer IDs are unique. +2. Product IDs are unique. +3. Orders must reference a valid customer. +4. Order Items must reference valid orders and products. +5. Payment Attempts must reference valid orders. +6. Monetary amounts must be non-negative. +7. Sum of line items must equal order total. +8. Status transitions must follow allowed business workflow. + +--- + +# CDC Pipeline + +The solution is designed around WAL-based CDC. + +## Production Architecture + +```text +PostgreSQL + | + v +WAL + | + v +Debezium + | + v +Kafka + | + v +Spark ``` -### 5. Open a Pull Request +## Assignment Implementation + +For local execution, CDC events are simulated using Python objects. + +Each CDC event contains: + +* sequence_number +* table_name +* operation +* primary_key +* before_image +* after_image +* event_timestamp + +Supported operations: + +* INSERT +* UPDATE +* DELETE -Once you are done, open a **Pull Request** from your fork back to this repository: +--- + +# Bronze Layer + +Purpose: + +Store immutable CDC history. + +Dataset: + +```text +data/bronze/cdc_events +``` -1. Go to your forked repository on GitHub. -2. Click **Contribute** > **Open pull request**. -3. Add a brief description of your approach and any decisions you made. -4. Submit the pull request. +Properties: + +* Append-only +* Stores every change +* Supports replay +* Supports historical reconstruction + +--- + +# Silver Layer + +Purpose: + +Create latest operational state. + +Datasets: + +```text +data/silver/customers +data/silver/products +data/silver/orders +data/silver/order_items +data/silver/payment_attempts +``` + +Properties: + +* Latest record per business key +* CDC metadata removed +* Optimized for warehouse loading + +--- + +# Warehouse Layer + +## Dimensions + +### dim_customer + +SCD Type 2 dimension. + +Tracks: + +* customer status changes +* customer profile changes + +Columns: + +* effective_from +* effective_to +* is_current + +--- + +### dim_product + +SCD Type 2 dimension. + +Tracks: + +* product status changes +* product attribute changes + +Columns: + +* effective_from +* effective_to +* is_current + +--- + +## Facts + +### fact_orders + +Order-level facts. + +### fact_order_items + +Line-level facts. + +### fact_payment_attempts + +Payment transaction facts. + +--- + +# Historical Recovery and Time Travel + +Historical recovery is supported through Bronze CDC history. + +Process: + +```text +Bronze CDC + | + v +Replay Until Timestamp + | + v +Rebuild Silver + | + v +Rebuild Warehouse +``` + +Capabilities: + +* Point-in-time reconstruction +* Replay after failures +* Warehouse restoration +* Auditability + +--- + +# Schema Change Detection + +Schema contracts are maintained in: + +```text +config/schema_contracts.json +config/enum_contracts.json +``` + +The pipeline validates source schemas before ingestion. + +Detected breaking changes: + +* Dropped column +* Renamed column +* Data type change +* Enum value changes +* Nullable changes +* Primary key changes +* Foreign key changes + +Behavior: + +```text +Schema Change Detected + | + v +Pipeline Fails + | + v +CDC Stops + | + v +Warehouse Not Updated +``` + +This prevents silent corruption. + +--- + +# Validation Parity + +## System Validations + +### Primary Key Validation + +* customer_id +* product_id +* order_id +* order_item_id +* payment_attempt_id + +### Foreign Key Validation + +* orders -> customers +* order_items -> orders +* order_items -> products +* payment_attempts -> orders + +### Not Null Validation + +Critical columns are validated. + +--- + +## Business Validations + +### Non-Negative Amounts + +Validated: + +* total_amount +* unit_price +* line_amount +* payment amount + +### Order Total Validation + +```text +SUM(order_items.line_amount) += +orders.total_amount +``` + +### Status Transition Validation + +Allowed: + +```text +CREATED -> PAID +CREATED -> CANCELLED +PAID -> SHIPPED +SHIPPED -> DELIVERED +``` + +Rejected: + +```text +DELIVERED -> CREATED +DELIVERED -> PAID +CANCELLED -> PAID +``` + +Validation failures stop the pipeline. + +--- + +# Catalog Exposure + +Catalog Metadata: + +```text +catalog/catalog.json +``` + +Published Datasets: + +## Bronze + +* bronze.cdc_events + +## Silver + +* silver.customers +* silver.products +* silver.orders +* silver.order_items +* silver.payment_attempts + +## Warehouse + +* warehouse.dim_customer +* warehouse.dim_product +* warehouse.fact_orders +* warehouse.fact_order_items +* warehouse.fact_payment_attempts + +Metadata Published: + +* dataset name +* owner +* layer +* refresh frequency +* description +* storage path + +--- + +# Airflow Workflow + +```text +validate_schema + | + v +ingest_cdc + | + v +build_silver + | + v +build_warehouse + | + v +run_data_quality_checks + | + v +publish_catalog +``` + +Schedule: + +Every 5 minutes. + +--- + +# Project Structure + +```text +submission/ + +├── airflow_dags/ +│ └── ecommerce_cdc_pipeline.py + +├── pipeline/ +│ ├── bronze.py +│ ├── silver.py +│ ├── warehouse.py +│ ├── cdc.py +│ ├── schema_contracts.py +│ └── catalog.py + +├── scripts/ +│ ├── validate_schema.py +│ ├── ingest_cdc.py +│ ├── build_silver.py +│ ├── build_warehouse.py +│ ├── run_data_quality_checks.py +│ ├── status_transition_validation.py +│ ├── publish_catalog.py +│ └── rebuild_warehouse.py + +├── source/ +│ ├── create_tables.py +│ ├── seed_data.py +│ ├── models.py +│ └── postgres.py + +├── catalog/ +│ └── catalog.json + +├── config/ +│ ├── schema_contracts.json +│ └── enum_contracts.json + +└── README.md +``` --- -## Notes +# Assumptions -- Make sure your solution runs and all instructions for running it are included. -- If you have any questions, reach out to the hiring team. +1. Local execution uses Spark Parquet datasets instead of a live PostgreSQL instance. +2. WAL-based CDC is simulated for assignment scope. +3. Production deployment would use PostgreSQL WAL, Debezium, and Kafka. +4. Airflow orchestrates end-to-end pipeline execution. +5. Warehouse dimensions use SCD Type 2 history tracking. diff --git a/data_evaluation_guide.md b/data_evaluation_guide.md deleted file mode 100644 index a919221..0000000 --- a/data_evaluation_guide.md +++ /dev/null @@ -1,389 +0,0 @@ -# Evaluation Guide — CDC Lakehouse Reliability Assignment - -## Purpose of This Guide - -This guide helps reviewers evaluate whether the candidate demonstrated senior-level data engineering judgment for a CDC-driven lakehouse workflow. - -The assignment is intentionally not about building a wide analytics project. It is about whether the candidate can reason about: - -- source modeling quality -- CDC correctness -- replay and recovery -- schema evolution safety -- validation parity -- historical reconstruction -- warehouse snapshot correctness -- dataset discoverability - -Use this rubric to assess both the implementation and the pull request description. - ---- - -## Reviewer Mindset - -Look for: - -- correctness before tooling preference -- explicit data contracts and invariants -- safe failure on schema incompatibility -- durable change retention -- clear distinction between lake history and warehouse snapshot -- thoughtful validation design -- easy-to-follow platform flow - -Do not over-index on any specific tool choice if the reasoning is sound. - ---- - -## Scoring Rubric - -You may score each category on a 1–4 scale: - -- **1 — Weak** -- **2 — Mixed / Partial** -- **3 — Strong** -- **4 — Exceptional** - -A strong submission will usually score mostly 3s, with one or two 4s in areas such as CDC design, schema safety, or validation rigor. - ---- - -## 1) Source Schema Design - -### What Reviewers Should Look For - -- 3 to 10 tables with realistic structure -- strong and weak entities -- sensible keys and relationships -- appropriate use of data types -- indexes that match likely access/change patterns -- clear constraints and invariants - -### Strong Signals - -- source schema resembles a realistic transactional domain -- strong vs weak entities are clearly identified and justified -- primary and foreign keys are sensible -- currency, dates, enums/statuses, and nullable fields are modeled intentionally -- indexes are not arbitrary; they support access patterns or CDC workflow assumptions -- important business rules are documented - -### Weak Signals - -- flat or unrealistic schema with little relational thinking -- tables exist mainly to satisfy the count requirement -- no clear strong/weak entity distinction -- missing keys, weak relationships, or under-modeled constraints -- indexes absent or added without rationale - -### Reviewer Questions - -- Does this look like a real transactional source system? -- Are relationships and entities modeled intentionally? -- Are constraints doing useful work, or merely storing fields? - ---- - -## 2) CDC Strategy and Correctness - -### What Reviewers Should Look For - -- clear definition of how changes are captured -- inserts, updates, and deletes all handled -- durable replay/restart story -- duplicate/retry awareness -- lake and warehouse fed from a coherent CDC design - -### Strong Signals - -- candidate explains exactly what a “change” means in their design -- restart/replay behavior is explicit -- deletes are not ignored -- duplicate events or reprocessing are handled safely -- checkpoints, offsets, or extraction boundaries are defined -- design makes near real-time warehouse availability believable - -### Weak Signals - -- CDC is described vaguely as periodic refresh -- no delete handling -- no replay story -- duplicate events could corrupt history or snapshot -- near real-time claim unsupported by design - -### Reviewer Questions - -- If ingestion restarts mid-run, what happens? -- Can this design avoid losing or duplicating changes? -- Is the warehouse derived in a way that is consistent with the lake history? - ---- - -## 3) Lake Modeling - -### What Reviewers Should Look For - -- every change captured in the lake -- append-oriented or immutable design -- enough metadata to reconstruct change history -- durable historical record - -### Strong Signals - -- operation type, timestamps, keys, and ordering/version metadata are preserved -- lake model supports full audit of source changes -- lake is clearly distinct from curated warehouse outputs -- append-only semantics are visible in both design and tests - -### Weak Signals - -- lake is effectively just another snapshot table -- updates overwrite history -- missing metadata makes reconstruction difficult -- lake and warehouse roles are blurred - -### Reviewer Questions - -- Can every source change be recovered from the lake? -- Is the lake truly historical, or just another transformed layer? -- What information would be needed for replay or audit, and is it present? - ---- - -## 4) Warehouse Modeling and Time Travel - -### What Reviewers Should Look For - -- warehouse exposes latest snapshot -- design supports moving backward in time / restore -- historical reconstruction story is credible -- downstream model is understandable - -### Strong Signals - -- latest-state tables are clearly defined -- temporal, versioned, or SCD-style logic is used thoughtfully where helpful -- candidate explains how point-in-time restore works operationally -- warehouse logic preserves correctness under late or repeated changes - -### Weak Signals - -- warehouse only has current snapshot with no restore story -- time travel is hand-waved without data model support -- current-state logic could drift from CDC history -- update ordering assumptions are unclear - -### Reviewer Questions - -- Can I reconstruct prior state from this design? -- Is the warehouse really usable for restore / time-based recovery? -- Does the warehouse reflect latest state consistently? - ---- - -## 5) Schema Change Safety - -### What Reviewers Should Look For - -- explicit detection of incompatible source changes -- ingestion stop behavior -- clear warning/failure signaling -- safe handling of non-backward-compatible schemas - -### Strong Signals - -- candidate identifies concrete compatibility rules -- breaking change detection is automated or clearly validated -- ingestion fails closed rather than silently drifting -- warnings or alerts are surfaced in a reviewable way -- schema contract is documented clearly - -### Weak Signals - -- schema evolution is ignored -- pipeline keeps running after breaking changes with undefined behavior -- no clear definition of what counts as incompatible -- failure signaling is vague or absent - -### Reviewer Questions - -- Would this pipeline stop safely on a dropped or renamed column? -- Is schema drift visible quickly? -- Does the candidate think in terms of contracts, not just happy-path ingestion? - ---- - -## 6) Validation Parity - -### What Reviewers Should Look For - -- source validations carried into warehouse checks/models -- both system and business rules considered -- failures surfaced clearly -- parity documented, not implied - -### Strong Signals - -- not-null, uniqueness, referential, and domain rules are checked -- business rules are restated downstream intentionally -- validation logic is automated through tests/assertions/checks -- failures are made actionable - -### Weak Signals - -- warehouse assumes source is always valid -- only superficial row-count checks exist -- business validations omitted entirely -- no traceability from source rule to warehouse assertion - -### Reviewer Questions - -- Are downstream consumers protected from invalid source states? -- Did the candidate mirror the important source assumptions? -- Could validation drift over time, or is it explicit? - ---- - -## 7) Catalog Exposure and Discoverability - -### What Reviewers Should Look For - -- both lake and warehouse are exposed in a discoverable way -- metadata is useful -- ownership and intended use are clear -- schema descriptions or data contract information exist - -### Strong Signals - -- lake and warehouse datasets are named and described clearly -- metadata includes schema, purpose, and update cadence -- consumer access path is documented -- candidate distinguishes operational data exposure from curated analytics exposure - -### Weak Signals - -- catalog requirement acknowledged but not addressed meaningfully -- no metadata beyond file/table names -- unclear which datasets consumers should use -- lake and warehouse access expectations are muddled - -### Reviewer Questions - -- Could a downstream user discover the right dataset easily? -- Is there enough metadata to understand what each dataset is for? -- Is access/documentation aligned with the modeled layers? - ---- - -## 8) Layering and Code Quality - -### What Reviewers Should Look For - -- clear separation of schema, ingestion, transformation, validation, and metadata concerns -- code/SQL/pipeline logic that is easy to follow -- maintainable repository structure -- minimal unnecessary complexity - -### Strong Signals - -- repository layout mirrors the data platform flow -- transformations and validations are separated clearly -- ingestion logic is understandable -- tests and configs live in obvious places -- naming makes the pipeline easy to reason about - -### Weak Signals - -- all logic mixed together in a few scripts -- no separation between raw, modeled, and validated data -- hard to see where invariants are enforced -- structure appears accidental rather than deliberate - -### Reviewer Questions - -- Can another engineer extend this without re-learning everything? -- Are responsibilities clearly separated? -- Is the flow from source to catalog easy to follow? - ---- - -## 9) Testing and Validation Depth - -### What Reviewers Should Look For - -- tests for CDC correctness -- tests for schema change failure -- tests for latest snapshot correctness -- tests or assertions for historical reconstruction -- evidence of Red, Blue, Green discipline - -### Strong Signals - -- inserts, updates, deletes, duplicates, and replay are all tested -- incompatible schema changes are tested explicitly -- warehouse correctness is asserted against change history -- validations are executable, not merely described -- candidate explains important missing tests if time constrained - -### Weak Signals - -- only happy-path pipeline run tested -- no schema change tests despite assignment emphasis -- no delete/history validation -- little confidence in correctness under failure or replay - -### Reviewer Questions - -- Do the tests prove the important platform behaviors? -- Is there confidence in correctness after restart or schema drift? -- Are the most failure-prone paths explicitly checked? - ---- - -## Suggested Overall Rating Bands - -### Exceptional -The candidate clearly understands CDC reliability, schema contracts, and temporal data design. -Lake, warehouse, validation, and schema safety reinforce one another well. - -### Strong -The solution is coherent, pragmatic, and well-structured. -There may be simplifications, but the correctness story is convincing. - -### Mixed -There are solid ideas, but one or more critical areas are underdeveloped, such as schema safety, replay, or validation parity. - -### Weak -The submission behaves more like a simple ETL refresh than a reliable CDC-based data platform. -History, schema change safety, or correctness guarantees are weak. - ---- - -## Common Failure Patterns - -Reviewers should watch for these: - -- lake stores only current state instead of every change -- warehouse snapshot logic is not clearly derived from CDC history -- deletes ignored -- replay/restart behavior undefined -- incompatible schema changes not detected or not blocking -- warehouse validations weaker than source assumptions -- no credible restore/time-travel story -- catalog requirement addressed only superficially - ---- - -## What to Value Most - -When in doubt, prioritize: - -1. CDC correctness and replayability -2. safe handling of incompatible schema changes -3. full change retention in the lake -4. correct latest snapshot and restore logic in the warehouse -5. validation parity with source rules -6. maintainable layering -7. breadth of tooling last - -A modest solution that is deeply correct should outrank a broader but fragile one. diff --git a/submission/sample-candidate/__pycache__/conftest.cpython-314-pytest-9.0.2.pyc b/submission/sample-candidate/__pycache__/conftest.cpython-314-pytest-9.0.2.pyc deleted file mode 100644 index 7309004..0000000 Binary files a/submission/sample-candidate/__pycache__/conftest.cpython-314-pytest-9.0.2.pyc and /dev/null differ diff --git a/submission/sample-candidate/airflow_dags/ecommerce_cdc_pipeline.py b/submission/sample-candidate/airflow_dags/ecommerce_cdc_pipeline.py new file mode 100644 index 0000000..1bfc880 --- /dev/null +++ b/submission/sample-candidate/airflow_dags/ecommerce_cdc_pipeline.py @@ -0,0 +1,185 @@ +from datetime import datetime +from airflow import DAG +from airflow.operators.python import PythonOperator + + +# ===================================================== +# TASK WRAPPERS +# ===================================================== + +def validate_schema(): + + from scripts.validate_schema import main + + main() + + +def ingest_cdc(): + + from scripts.ingest_cdc import main + + main() + + +def build_silver(): + + from scripts.build_silver import main + + main() + + +def build_warehouse(): + + from scripts.build_warehouse import main + + main() + + +def run_data_quality(): + + from scripts.run_data_quality_checks import main + + main() + + +def publish_catalog(): + + from scripts.publish_catalog import main + + main() + + +# ===================================================== +# DAG CONFIGURATION +# ===================================================== + +default_args = { + + "owner": "data-engineering", + + "depends_on_past": False, + + "email_on_failure": True, + + "email_on_retry": False, + + "retries": 1 +} + + +# ===================================================== +# DAG +# ===================================================== + +with DAG( + + dag_id="ecommerce_cdc_pipeline", + + description=( + "CDC pipeline from PostgreSQL " + "to Bronze, Silver and Warehouse" + ), + + default_args=default_args, + + start_date=datetime( + 2026, + 1, + 1 + ), + + schedule="*/5 * * * *", + + catchup=False, + + tags=[ + "cdc", + "spark", + "warehouse" + ] + +) as dag: + + # ========================================== + # 1. SCHEMA VALIDATION + # ========================================== + + validate_schema_task = PythonOperator( + + task_id="validate_schema", + + python_callable=validate_schema + ) + + # ========================================== + # 2. CDC INGESTION + # ========================================== + + ingest_cdc_task = PythonOperator( + + task_id="ingest_cdc", + + python_callable=ingest_cdc + ) + + # ========================================== + # 3. BUILD SILVER + # ========================================== + + build_silver_task = PythonOperator( + + task_id="build_silver", + + python_callable=build_silver + ) + + # ========================================== + # 4. BUILD WAREHOUSE + # ========================================== + + build_warehouse_task = PythonOperator( + + task_id="build_warehouse", + + python_callable=build_warehouse + ) + + # ========================================== + # 5. DATA QUALITY CHECKS + # ========================================== + + data_quality_task = PythonOperator( + + task_id="run_data_quality_checks", + + python_callable=run_data_quality + ) + + # ========================================== + # 6. PUBLISH CATALOG + # ========================================== + + publish_catalog_task = PythonOperator( + + task_id="publish_catalog", + + python_callable=publish_catalog + ) + + # ========================================== + # DEPENDENCIES + # ========================================== + + ( + validate_schema_task + >> + ingest_cdc_task + >> + build_silver_task + >> + build_warehouse_task + >> + data_quality_task + >> + publish_catalog_task + ) \ No newline at end of file diff --git a/submission/sample-candidate/catalog/catalog.json b/submission/sample-candidate/catalog/catalog.json index 83dfa55..de91d0e 100644 --- a/submission/sample-candidate/catalog/catalog.json +++ b/submission/sample-candidate/catalog/catalog.json @@ -1,77 +1,85 @@ { "datasets": [ + { - "name": "lake_cdc_events", - "layer": "lake", - "description": "Append-only log of every CDC event captured from the payments source system. Preserves full change history for replay and point-in-time recovery.", - "owner": "data-platform", - "consumers": ["data-platform", "analytics", "audit"], - "update_cadence": "real-time", - "schema": { - "sequence": "INTEGER — monotonic capture offset", - "operation": "VARCHAR — insert | update | delete", - "table_name": "VARCHAR — source table name", - "primary_key": "VARCHAR — source row PK value", - "data": "VARCHAR (JSON) — full row snapshot at capture time", - "captured_at": "TIMESTAMP — UTC capture time" - } + "name": "bronze.cdc_events", + "layer": "bronze", + "path": "data/bronze/cdc_events", + "owner": "data-engineering", + "refresh_frequency": "5 minutes", + "description": "Immutable CDC event history from PostgreSQL source system" }, + { - "name": "wh_customers", + "name": "silver.customers", + "layer": "silver", + "path": "data/silver/customers", + "owner": "data-engineering", + "refresh_frequency": "5 minutes", + "description": "Latest customer snapshot derived from CDC events" + }, + + { + "name": "silver.products", + "layer": "silver", + "path": "data/silver/products", + "owner": "data-engineering", + "refresh_frequency": "5 minutes", + "description": "Latest product snapshot derived from CDC events" + }, + + { + "name": "silver.orders", + "layer": "silver", + "path": "data/silver/orders", + "owner": "data-engineering", + "refresh_frequency": "5 minutes", + "description": "Latest order snapshot derived from CDC events" + }, + + { + "name": "warehouse.dim_customer", + "layer": "warehouse", + "path": "data/warehouse/dim_customer", + "owner": "analytics", + "refresh_frequency": "5 minutes", + "description": "Customer SCD Type 2 dimension" + }, + + { + "name": "warehouse.dim_product", + "layer": "warehouse", + "path": "data/warehouse/dim_product", + "owner": "analytics", + "refresh_frequency": "5 minutes", + "description": "Product SCD Type 2 dimension" + }, + + { + "name": "warehouse.fact_orders", "layer": "warehouse", - "description": "Current-state customer snapshot. Reflects the latest known state of each customer row. Soft-deleted rows are flagged with _deleted=true.", - "owner": "data-platform", - "consumers": ["analytics", "product", "finance"], - "update_cadence": "near-real-time", - "schema": { - "customer_id": "VARCHAR — primary key", - "name": "VARCHAR", - "email": "VARCHAR", - "status": "VARCHAR — active | suspended | closed", - "created_at": "TIMESTAMP", - "updated_at": "TIMESTAMP", - "_cdc_seq": "INTEGER — sequence of last CDC event", - "_deleted": "BOOLEAN — true if source row was deleted" - } + "path": "data/warehouse/fact_orders", + "owner": "analytics", + "refresh_frequency": "5 minutes", + "description": "Current order fact table" }, + { - "name": "wh_wallets", + "name": "warehouse.fact_order_items", "layer": "warehouse", - "description": "Current-state wallet snapshot. Balance reflects the latest value written via CDC. Negative balances are a data quality violation.", - "owner": "data-platform", - "consumers": ["analytics", "finance"], - "update_cadence": "near-real-time", - "schema": { - "wallet_id": "VARCHAR — primary key", - "customer_id": "VARCHAR — FK to wh_customers", - "balance": "DECIMAL(18,2)", - "currency": "VARCHAR", - "status": "VARCHAR — active | frozen | closed", - "created_at": "TIMESTAMP", - "updated_at": "TIMESTAMP", - "_cdc_seq": "INTEGER", - "_deleted": "BOOLEAN" - } + "path": "data/warehouse/fact_order_items", + "owner": "analytics", + "refresh_frequency": "5 minutes", + "description": "Order item fact table" }, + { - "name": "wh_transactions", + "name": "warehouse.fact_payment_attempts", "layer": "warehouse", - "description": "Current-state transaction snapshot. Each row is the latest state of a transaction from the source. Amount must always be positive.", - "owner": "data-platform", - "consumers": ["analytics", "finance", "audit"], - "update_cadence": "near-real-time", - "schema": { - "transaction_id": "VARCHAR — primary key", - "wallet_id": "VARCHAR — FK to wh_wallets", - "amount": "DECIMAL(18,2) — always > 0", - "direction": "VARCHAR — credit | debit", - "status": "VARCHAR — pending | settled | failed | reversed", - "reference": "VARCHAR — nullable external reference", - "created_at": "TIMESTAMP", - "settled_at": "TIMESTAMP — nullable until settled", - "_cdc_seq": "INTEGER", - "_deleted": "BOOLEAN" - } + "path": "data/warehouse/fact_payment_attempts", + "owner": "analytics", + "refresh_frequency": "5 minutes", + "description": "Payment attempts fact table" } ] -} +} \ No newline at end of file diff --git a/submission/sample-candidate/config/enum_contracts.json b/submission/sample-candidate/config/enum_contracts.json new file mode 100644 index 0000000..83e45a7 --- /dev/null +++ b/submission/sample-candidate/config/enum_contracts.json @@ -0,0 +1,26 @@ +{ + "customer_status": [ + "ACTIVE", + "INACTIVE", + "SUSPENDED" + ], + + "product_status": [ + "ACTIVE", + "DISCONTINUED" + ], + + "order_status": [ + "CREATED", + "PAID", + "SHIPPED", + "DELIVERED", + "CANCELLED" + ], + + "payment_status": [ + "SUCCESS", + "FAILED", + "PENDING" + ] + } \ No newline at end of file diff --git a/submission/sample-candidate/config/schema_contracts.json b/submission/sample-candidate/config/schema_contracts.json new file mode 100644 index 0000000..8a97798 --- /dev/null +++ b/submission/sample-candidate/config/schema_contracts.json @@ -0,0 +1,51 @@ +{ + "customers": { + "customer_id": "bigint", + "email": "string", + "first_name": "string", + "last_name": "string", + "phone": "string", + "customer_status": "string", + "created_at": "timestamp", + "updated_at": "timestamp" + }, + + "products": { + "product_id": "bigint", + "sku": "string", + "product_name": "string", + "category": "string", + "unit_price": "decimal(18,2)", + "product_status": "string", + "created_at": "timestamp", + "updated_at": "timestamp" + }, + + "orders": { + "order_id": "bigint", + "customer_id": "bigint", + "order_status": "string", + "total_amount": "decimal(18,2)", + "created_at": "timestamp", + "updated_at": "timestamp" + }, + + "order_items": { + "order_item_id": "bigint", + "order_id": "bigint", + "product_id": "bigint", + "quantity": "integer", + "unit_price": "decimal(18,2)", + "line_amount": "decimal(18,2)", + "created_at": "timestamp" + }, + + "payment_attempts": { + "payment_attempt_id": "bigint", + "order_id": "bigint", + "payment_status": "string", + "gateway_transaction_id": "string", + "amount": "decimal(18,2)", + "attempt_timestamp": "timestamp" + } + } \ No newline at end of file diff --git a/submission/sample-candidate/conftest.py b/submission/sample-candidate/conftest.py deleted file mode 100644 index ce3c67b..0000000 --- a/submission/sample-candidate/conftest.py +++ /dev/null @@ -1,6 +0,0 @@ -"""Root conftest — adds submission/ to sys.path so tests can import source/pipeline.""" - -import os -import sys - -sys.path.insert(0, os.path.dirname(__file__)) diff --git a/submission/sample-candidate/pipeline/__pycache__/__init__.cpython-314.pyc b/submission/sample-candidate/pipeline/__pycache__/__init__.cpython-314.pyc deleted file mode 100644 index 1523f32..0000000 Binary files a/submission/sample-candidate/pipeline/__pycache__/__init__.cpython-314.pyc and /dev/null differ diff --git a/submission/sample-candidate/pipeline/__pycache__/cdc.cpython-314.pyc b/submission/sample-candidate/pipeline/__pycache__/cdc.cpython-314.pyc deleted file mode 100644 index a17fa0b..0000000 Binary files a/submission/sample-candidate/pipeline/__pycache__/cdc.cpython-314.pyc and /dev/null differ diff --git a/submission/sample-candidate/pipeline/__pycache__/lake.cpython-314.pyc b/submission/sample-candidate/pipeline/__pycache__/lake.cpython-314.pyc deleted file mode 100644 index 143260f..0000000 Binary files a/submission/sample-candidate/pipeline/__pycache__/lake.cpython-314.pyc and /dev/null differ diff --git a/submission/sample-candidate/pipeline/__pycache__/warehouse.cpython-314.pyc b/submission/sample-candidate/pipeline/__pycache__/warehouse.cpython-314.pyc deleted file mode 100644 index da28d22..0000000 Binary files a/submission/sample-candidate/pipeline/__pycache__/warehouse.cpython-314.pyc and /dev/null differ diff --git a/submission/sample-candidate/pipeline/bronze.py b/submission/sample-candidate/pipeline/bronze.py new file mode 100644 index 0000000..eaeff1e --- /dev/null +++ b/submission/sample-candidate/pipeline/bronze.py @@ -0,0 +1,89 @@ +from pyspark.sql import SparkSession +from pyspark.sql.types import ( + StructType, + StructField, + LongType, + StringType, + TimestampType +) + + +class BronzeLayer: + + def __init__( + self, + spark: SparkSession, + bronze_path: str + ): + + self.spark = spark + + self.bronze_path = bronze_path + + @staticmethod + def bronze_schema(): + + return StructType([ + + StructField( + "sequence_number", + LongType(), + False + ), + + StructField( + "table_name", + StringType(), + False + ), + + StructField( + "operation", + StringType(), + False + ), + + StructField( + "primary_key", + StringType(), + False + ), + + StructField( + "before_image", + StringType(), + True + ), + + StructField( + "after_image", + StringType(), + True + ), + + StructField( + "event_timestamp", + TimestampType(), + False + ) + ]) + + def write_events( + self, + events_df + ): + + ( + events_df + .write + .mode("append") + .parquet(self.bronze_path) + ) + + def read_events(self): + + return ( + self.spark + .read + .parquet(self.bronze_path) + ) \ No newline at end of file diff --git a/submission/sample-candidate/pipeline/catalog.py b/submission/sample-candidate/pipeline/catalog.py new file mode 100644 index 0000000..e716bc2 --- /dev/null +++ b/submission/sample-candidate/pipeline/catalog.py @@ -0,0 +1,26 @@ +import json + + +class CatalogPublisher: + + def __init__( + self, + catalog_file + ): + + self.catalog_file = catalog_file + + def load_catalog(self): + + with open( + self.catalog_file, + "r" + ) as f: + + return json.load(f) + + def list_datasets(self): + + catalog = self.load_catalog() + + return catalog["datasets"] \ No newline at end of file diff --git a/submission/sample-candidate/pipeline/cdc.py b/submission/sample-candidate/pipeline/cdc.py index c8c2dcb..d022a59 100644 --- a/submission/sample-candidate/pipeline/cdc.py +++ b/submission/sample-candidate/pipeline/cdc.py @@ -1,89 +1,92 @@ -""" -CDC capture layer. +import json +from datetime import datetime +from pyspark.sql import SparkSession +from pyspark.sql import Row -Simulates WAL-based change capture: every insert/update/delete on the source -produces a CDCRecord with a monotonically increasing sequence number. -Replay safety: callers can checkpoint the last processed sequence and call -records_since(offset) to replay only unprocessed changes after a restart. -""" +class CDCIngestion: -from __future__ import annotations - -from dataclasses import dataclass, field -from datetime import datetime, timezone -from typing import Any - -VALID_OPERATIONS = frozenset({"insert", "update", "delete"}) - - -@dataclass -class CDCRecord: - operation: str - table: str - primary_key: str - data: dict[str, Any] - captured_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) - sequence: int = 0 - - def __post_init__(self) -> None: - if self.operation not in VALID_OPERATIONS: - raise ValueError( - f"Invalid CDC operation {self.operation!r}. " - f"Must be one of: {sorted(VALID_OPERATIONS)}" - ) - - -class CDCCapture: - """ - In-process CDC log. - - Production analogue: a Debezium/Kafka connector reading Postgres WAL. - Each record carries a sequence number equivalent to a Kafka offset or - Postgres LSN for checkpoint-based replay. """ + Production: - def __init__(self) -> None: - self._log: list[CDCRecord] = [] - self._seq: int = 0 + PostgreSQL WAL + -> + Debezium + -> + Kafka + -> + Spark - # ── public write API ───────────────────────────────────────────────────── + Assignment: - def insert(self, table: str, pk: str, data: dict[str, Any]) -> CDCRecord: - return self._record("insert", table, pk, data) - - def update(self, table: str, pk: str, data: dict[str, Any]) -> CDCRecord: - return self._record("update", table, pk, data) - - def delete(self, table: str, pk: str, data: dict[str, Any]) -> CDCRecord: - return self._record("delete", table, pk, data) - - # ── public read / replay API ───────────────────────────────────────────── - - def records_since(self, offset: int = 0) -> list[CDCRecord]: - """Return all records with sequence > offset (checkpoint replay).""" - return [r for r in self._log if r.sequence > offset] - - @property - def latest_sequence(self) -> int: - return self._seq - - @property - def log(self) -> list[CDCRecord]: - return list(self._log) + CDC events are provided + as JSON records. + """ - # ── internal ───────────────────────────────────────────────────────────── + def __init__( + self, + spark: SparkSession + ): + + self.spark = spark + + def create_cdc_dataframe( + self, + events: list + ): + + rows = [] + + for event in events: + + rows.append( + + Row( + sequence_number=event["sequence_number"], + table_name=event["table_name"], + operation=event["operation"], + primary_key=str( + event["primary_key"] + ), + before_image=json.dumps( + event.get("before_image") + ) + if event.get("before_image") + else None, + after_image=json.dumps( + event.get("after_image") + ) + if event.get("after_image") + else None, + event_timestamp=event[ + "event_timestamp" + ] + ) + ) - def _record( - self, operation: str, table: str, pk: str, data: dict[str, Any] - ) -> CDCRecord: - self._seq += 1 - rec = CDCRecord( - operation=operation, - table=table, - primary_key=pk, - data=data, - sequence=self._seq, + return self.spark.createDataFrame( + rows ) - self._log.append(rec) - return rec + + def latest_sequence( + self, + bronze_df + ): + + result = bronze_df.agg( + { + "sequence_number": "max" + } + ).collect()[0][0] + + return result if result else 0 + + CDC_SCHEMA = StructType([ + StructField("sequence_number", LongType(), False), + StructField("table_name", StringType(), False), + StructField("operation", StringType(), False), + StructField("primary_key", StringType(), False), + StructField("before_image", StringType(), True), + StructField("after_image", StringType(), True), + StructField("event_timestamp", TimestampType(), False) + ]) \ No newline at end of file diff --git a/submission/sample-candidate/pipeline/lake.py b/submission/sample-candidate/pipeline/lake.py deleted file mode 100644 index 7cb15a7..0000000 --- a/submission/sample-candidate/pipeline/lake.py +++ /dev/null @@ -1,69 +0,0 @@ -""" -Lake layer — append-only storage for every CDC event. - -Every change is written exactly once. The lake is the source of truth for -point-in-time replay and historical reconstruction. - -Production analogue: Parquet/Delta files on S3 or GCS, partitioned by -table_name and captured_at date. No row is ever modified or deleted. -""" - -from __future__ import annotations - -import json -from datetime import datetime - -import duckdb - -from pipeline.cdc import CDCRecord - - -def create_lake_table(conn: duckdb.DuckDBPyConnection) -> None: - conn.execute(""" - CREATE TABLE IF NOT EXISTS lake_cdc_events ( - sequence INTEGER NOT NULL, - operation VARCHAR NOT NULL, - table_name VARCHAR NOT NULL, - primary_key VARCHAR NOT NULL, - data VARCHAR NOT NULL, - captured_at TIMESTAMP NOT NULL - ) - """) - - -def append_to_lake(conn: duckdb.DuckDBPyConnection, records: list[CDCRecord]) -> int: - """ - Append CDC records to the lake. - - Returns the number of records written. - Idempotency note: in production, deduplicate by sequence before appending. - """ - if not records: - return 0 - - rows = [ - ( - r.sequence, - r.operation, - r.table, - r.primary_key, - _serialize(r.data), - r.captured_at, - ) - for r in records - ] - conn.executemany( - "INSERT INTO lake_cdc_events VALUES (?, ?, ?, ?, ?, ?)", - rows, - ) - return len(rows) - - -def _serialize(data: dict) -> str: - return json.dumps(data, default=_json_default) - - -def _json_default(obj: object) -> str: - if isinstance(obj, datetime): - return obj.isoformat() - raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable") diff --git a/submission/sample-candidate/pipeline/schema_contracts.py b/submission/sample-candidate/pipeline/schema_contracts.py new file mode 100644 index 0000000..5527cf4 --- /dev/null +++ b/submission/sample-candidate/pipeline/schema_contracts.py @@ -0,0 +1,160 @@ +from dataclasses import dataclass +from typing import Dict, List + + +@dataclass +class SchemaViolation: + + table_name: str + + violation_type: str + + message: str + + +class SchemaContractException(Exception): + pass + + +class SchemaValidator: + + """ + Compares actual source schema + against expected schema contract. + """ + + def __init__( + self, + expected_contracts: Dict + ): + + self.expected_contracts = ( + expected_contracts + ) + + def validate( + self, + actual_contracts: Dict + ): + + violations = [] + + for table_name in ( + self.expected_contracts.keys() + ): + + if table_name not in actual_contracts: + + violations.append( + SchemaViolation( + table_name, + "TABLE_MISSING", + f"{table_name} missing" + ) + ) + + continue + + expected_columns = ( + self.expected_contracts[ + table_name + ] + ) + + actual_columns = ( + actual_contracts[ + table_name + ] + ) + + violations.extend( + self._compare_columns( + table_name, + expected_columns, + actual_columns + ) + ) + + if violations: + + raise SchemaContractException( + self._build_error_message( + violations + ) + ) + + return True + + def _compare_columns( + self, + table_name, + expected_columns, + actual_columns + ): + + violations = [] + + for column_name in ( + expected_columns.keys() + ): + + if column_name not in actual_columns: + + violations.append( + SchemaViolation( + table_name, + "COLUMN_MISSING", + f"{column_name} removed" + ) + ) + + continue + + expected_type = ( + expected_columns[ + column_name + ] + ) + + actual_type = ( + actual_columns[ + column_name + ] + ) + + if expected_type != actual_type: + + violations.append( + SchemaViolation( + table_name, + "TYPE_CHANGE", + ( + f"{column_name} " + f"changed from " + f"{expected_type}" + f" to " + f"{actual_type}" + ) + ) + ) + + return violations + + @staticmethod + def _build_error_message( + violations: List[SchemaViolation] + ): + + messages = [] + + for violation in violations: + + messages.append( + ( + f"[{violation.table_name}] " + f"{violation.violation_type}: " + f"{violation.message}" + ) + ) + + return "\n".join(messages) \ No newline at end of file diff --git a/submission/sample-candidate/pipeline/silver.py b/submission/sample-candidate/pipeline/silver.py new file mode 100644 index 0000000..27c04d6 --- /dev/null +++ b/submission/sample-candidate/pipeline/silver.py @@ -0,0 +1,55 @@ +import json + +from pyspark.sql.functions import ( + col, + udf +) + +from pyspark.sql.types import ( + MapType, + StringType +) + + +class SilverLayer: + + """ + Converts raw CDC records + into curated datasets. + """ + + @staticmethod + def parse_json_column(): + + def parse_json(value): + + if value is None: + return None + + return json.loads(value) + + return udf( + parse_json, + MapType( + StringType(), + StringType() + ) + ) + + def flatten_after_image( + self, + bronze_df + ): + + parse_udf = ( + self.parse_json_column() + ) + + df = bronze_df.withColumn( + "record", + parse_udf( + col("after_image") + ) + ) + + return df \ No newline at end of file diff --git a/submission/sample-candidate/pipeline/warehouse.py b/submission/sample-candidate/pipeline/warehouse.py index ab32401..3f96b8c 100644 --- a/submission/sample-candidate/pipeline/warehouse.py +++ b/submission/sample-candidate/pipeline/warehouse.py @@ -1,124 +1,233 @@ -""" -Warehouse layer — current-state snapshot built from CDC events. - -Each warehouse table mirrors the source table with two extra columns: - _cdc_seq : sequence of the last CDC event that touched this row - _deleted : soft-delete flag set when a DELETE event is received - -Production analogue: BigQuery/Snowflake tables refreshed by a streaming -merge job keyed on primary key. SCD2 history tables would sit alongside -these current-state tables for time-travel queries. -""" - -from __future__ import annotations - -import duckdb - -from pipeline.cdc import CDCRecord - -# Source table → warehouse table -_TABLE_MAP: dict[str, str] = { - "customers": "wh_customers", - "wallets": "wh_wallets", - "transactions": "wh_transactions", -} - -# Source table → primary key column name -_PK_MAP: dict[str, str] = { - "customers": "customer_id", - "wallets": "wallet_id", - "transactions": "transaction_id", -} - - -def create_warehouse_tables(conn: duckdb.DuckDBPyConnection) -> None: - conn.execute(""" - CREATE TABLE IF NOT EXISTS wh_customers ( - customer_id VARCHAR PRIMARY KEY, - name VARCHAR, - email VARCHAR, - status VARCHAR, - created_at TIMESTAMP, - updated_at TIMESTAMP, - _cdc_seq INTEGER NOT NULL, - _deleted BOOLEAN NOT NULL DEFAULT false +from pyspark.sql import DataFrame +from pyspark.sql import functions as F + + +class SCD2Processor: + + def __init__(self, spark): + self.spark = spark + + def merge_dimension( + self, + incoming_df: DataFrame, + existing_df: DataFrame, + business_key: str, + compare_columns: list + ) -> DataFrame: + + current_df = ( + existing_df + .filter(F.col("is_current") == True) + ) + + join_condition = [ + current_df[business_key] + == incoming_df[business_key] + ] + + joined_df = ( + incoming_df.alias("new") + .join( + current_df.alias("old"), + join_condition, + "left" + ) + ) + + change_condition = None + + for col_name in compare_columns: + + condition = ( + F.col(f"new.{col_name}") + != + F.col(f"old.{col_name}") + ) + + if change_condition is None: + change_condition = condition + else: + change_condition = ( + change_condition | condition + ) + + changed_rows = joined_df.filter( + change_condition ) - """) - - conn.execute(""" - CREATE TABLE IF NOT EXISTS wh_wallets ( - wallet_id VARCHAR PRIMARY KEY, - customer_id VARCHAR, - balance DECIMAL(18, 2), - currency VARCHAR, - status VARCHAR, - created_at TIMESTAMP, - updated_at TIMESTAMP, - _cdc_seq INTEGER NOT NULL, - _deleted BOOLEAN NOT NULL DEFAULT false + + expired_rows = ( + current_df.alias("old") + .join( + changed_rows.select( + business_key + ).alias("chg"), + business_key + ) + .withColumn( + "effective_to", + F.current_timestamp() + ) + .withColumn( + "is_current", + F.lit(False) + ) ) - """) - - conn.execute(""" - CREATE TABLE IF NOT EXISTS wh_transactions ( - transaction_id VARCHAR PRIMARY KEY, - wallet_id VARCHAR, - amount DECIMAL(18, 2), - direction VARCHAR, - status VARCHAR, - reference VARCHAR, - created_at TIMESTAMP, - settled_at TIMESTAMP, - _cdc_seq INTEGER NOT NULL, - _deleted BOOLEAN NOT NULL DEFAULT false + + new_versions = ( + changed_rows + .select( + "new.*" + ) + .withColumn( + "effective_from", + F.current_timestamp() + ) + .withColumn( + "effective_to", + F.lit(None) + ) + .withColumn( + "is_current", + F.lit(True) + ) ) - """) - - -def apply_cdc_records( - conn: duckdb.DuckDBPyConnection, records: list[CDCRecord] -) -> None: - """ - Apply CDC records to the warehouse current-state tables in sequence order. - - - insert / update → upsert (insert new row or overwrite existing) - - delete → set _deleted = true - """ - for record in sorted(records, key=lambda r: r.sequence): - wh_table = _TABLE_MAP.get(record.table) - pk_col = _PK_MAP.get(record.table) - if not wh_table or not pk_col: - continue - - pk_val = record.primary_key - - if record.operation == "delete": - conn.execute( - f"UPDATE {wh_table} SET _deleted = true, _cdc_seq = ?" - f" WHERE {pk_col} = ?", - [record.sequence, pk_val], + + unchanged_rows = ( + existing_df.alias("e") + .join( + changed_rows.select( + business_key + ), + business_key, + "leftanti" ) - continue - - data = {**record.data, "_cdc_seq": record.sequence, "_deleted": False} - cols = list(data.keys()) - vals = list(data.values()) - placeholders = ", ".join(["?"] * len(vals)) - - existing = conn.execute( - f"SELECT COUNT(*) FROM {wh_table} WHERE {pk_col} = ?", - [pk_val], - ).fetchone()[0] - - if existing: - set_clause = ", ".join([f"{c} = ?" for c in cols]) - conn.execute( - f"UPDATE {wh_table} SET {set_clause} WHERE {pk_col} = ?", - vals + [pk_val], + ) + + final_df = ( + unchanged_rows + .unionByName(expired_rows) + .unionByName(new_versions) + ) + + return final_df + + +class CustomerDimensionBuilder: + + def __init__(self, spark): + self.spark = spark + + def build( + self, + silver_customer_df, + existing_dimension_df + ): + + processor = SCD2Processor( + self.spark + ) + + return processor.merge_dimension( + incoming_df=silver_customer_df, + existing_df=existing_dimension_df, + business_key="customer_id", + compare_columns=[ + "email", + "first_name", + "last_name", + "phone", + "customer_status" + ] + ) + +class ProductDimensionBuilder: + + def __init__(self, spark): + self.spark = spark + + def build( + self, + silver_product_df, + existing_dimension_df + ): + + processor = SCD2Processor( + self.spark + ) + + return processor.merge_dimension( + incoming_df=silver_product_df, + existing_df=existing_dimension_df, + business_key="product_id", + compare_columns=[ + "sku", + "product_name", + "category", + "unit_price", + "product_status" + ] + ) + +class FactOrderBuilder: + + @staticmethod + def build( + silver_orders_df + ): + + return silver_orders_df + +class FactOrderItemBuilder: + + @staticmethod + def build( + silver_order_items_df + ): + + return silver_order_items_df + +class FactPaymentBuilder: + + @staticmethod + def build( + silver_payment_df + ): + + return silver_payment_df + +class WarehouseWriter: + + def __init__( + self, + warehouse_base_path + ): + + self.base_path = warehouse_base_path + + def write_dimension( + self, + df, + dimension_name + ): + + ( + df.write + .mode("overwrite") + .parquet( + f"{self.base_path}/{dimension_name}" ) - else: - col_list = ", ".join(cols) - conn.execute( - f"INSERT INTO {wh_table} ({col_list}) VALUES ({placeholders})", - vals, + ) + + def write_fact( + self, + df, + fact_name + ): + + ( + df.write + .mode("overwrite") + .parquet( + f"{self.base_path}/{fact_name}" ) + ) \ No newline at end of file diff --git a/submission/sample-candidate/requirements.txt b/submission/sample-candidate/requirements.txt deleted file mode 100644 index ab2841c..0000000 --- a/submission/sample-candidate/requirements.txt +++ /dev/null @@ -1,2 +0,0 @@ -duckdb>=0.10.0 -pytest>=7.4 diff --git a/submission/sample-candidate/scripts/build_silver.py b/submission/sample-candidate/scripts/build_silver.py new file mode 100644 index 0000000..dc32b35 --- /dev/null +++ b/submission/sample-candidate/scripts/build_silver.py @@ -0,0 +1,452 @@ +import json +import logging + +from pyspark.sql import SparkSession +from pyspark.sql import functions as F +from pyspark.sql.window import Window + + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s" +) + +logger = logging.getLogger(__name__) + + +BRONZE_PATH = "data/bronze/cdc_events" + +SILVER_CUSTOMERS_PATH = "data/silver/customers" +SILVER_PRODUCTS_PATH = "data/silver/products" +SILVER_ORDERS_PATH = "data/silver/orders" +SILVER_ORDER_ITEMS_PATH = "data/silver/order_items" +SILVER_PAYMENTS_PATH = "data/silver/payment_attempts" + + +def create_spark_session(): + + return ( + SparkSession.builder + .appName("Build Silver Layer") + .getOrCreate() + ) + + +def extract_json_field(column_name, field_name): + + return F.get_json_object( + F.col(column_name), + f"$.{field_name}" + ) + + +def build_latest_snapshot( + bronze_df, + table_name +): + + table_df = ( + + bronze_df + + .filter( + F.col("table_name") == table_name + ) + + .filter( + F.col("operation") != "DELETE" + ) + ) + + window_spec = ( + + Window + + .partitionBy("primary_key") + + .orderBy( + F.col( + "sequence_number" + ).desc() + ) + ) + + latest_df = ( + + table_df + + .withColumn( + "rn", + F.row_number().over( + window_spec + ) + ) + + .filter( + F.col("rn") == 1 + ) + + .drop("rn") + ) + + return latest_df + + +def build_customers(bronze_df): + + latest_df = build_latest_snapshot( + bronze_df, + "customers" + ) + + return ( + + latest_df + + .select( + + extract_json_field( + "after_image", + "customer_id" + ).cast("bigint").alias( + "customer_id" + ), + + extract_json_field( + "after_image", + "email" + ).alias( + "email" + ), + + extract_json_field( + "after_image", + "first_name" + ).alias( + "first_name" + ), + + extract_json_field( + "after_image", + "last_name" + ).alias( + "last_name" + ), + + extract_json_field( + "after_image", + "phone" + ).alias( + "phone" + ), + + extract_json_field( + "after_image", + "customer_status" + ).alias( + "customer_status" + ) + ) + ) + + +def build_products(bronze_df): + + latest_df = build_latest_snapshot( + bronze_df, + "products" + ) + + return ( + + latest_df + + .select( + + extract_json_field( + "after_image", + "product_id" + ).cast("bigint").alias( + "product_id" + ), + + extract_json_field( + "after_image", + "sku" + ).alias( + "sku" + ), + + extract_json_field( + "after_image", + "product_name" + ).alias( + "product_name" + ), + + extract_json_field( + "after_image", + "category" + ).alias( + "category" + ), + + extract_json_field( + "after_image", + "unit_price" + ).cast( + "decimal(18,2)" + ).alias( + "unit_price" + ), + + extract_json_field( + "after_image", + "product_status" + ).alias( + "product_status" + ) + ) + ) + + +def build_orders(bronze_df): + + latest_df = build_latest_snapshot( + bronze_df, + "orders" + ) + + return ( + + latest_df + + .select( + + extract_json_field( + "after_image", + "order_id" + ).cast("bigint").alias( + "order_id" + ), + + extract_json_field( + "after_image", + "customer_id" + ).cast("bigint").alias( + "customer_id" + ), + + extract_json_field( + "after_image", + "order_status" + ).alias( + "order_status" + ), + + extract_json_field( + "after_image", + "total_amount" + ).cast( + "decimal(18,2)" + ).alias( + "total_amount" + ) + ) + ) + + +def build_order_items(bronze_df): + + latest_df = build_latest_snapshot( + bronze_df, + "order_items" + ) + + return ( + + latest_df + + .select( + + extract_json_field( + "after_image", + "order_item_id" + ).cast("bigint").alias( + "order_item_id" + ), + + extract_json_field( + "after_image", + "order_id" + ).cast("bigint").alias( + "order_id" + ), + + extract_json_field( + "after_image", + "product_id" + ).cast("bigint").alias( + "product_id" + ), + + extract_json_field( + "after_image", + "quantity" + ).cast("int").alias( + "quantity" + ), + + extract_json_field( + "after_image", + "unit_price" + ).cast( + "decimal(18,2)" + ).alias( + "unit_price" + ), + + extract_json_field( + "after_image", + "line_amount" + ).cast( + "decimal(18,2)" + ).alias( + "line_amount" + ) + ) + ) + + +def build_payment_attempts(bronze_df): + + latest_df = build_latest_snapshot( + bronze_df, + "payment_attempts" + ) + + return ( + + latest_df + + .select( + + extract_json_field( + "after_image", + "payment_attempt_id" + ).cast("bigint").alias( + "payment_attempt_id" + ), + + extract_json_field( + "after_image", + "order_id" + ).cast("bigint").alias( + "order_id" + ), + + extract_json_field( + "after_image", + "payment_status" + ).alias( + "payment_status" + ), + + extract_json_field( + "after_image", + "gateway_transaction_id" + ).alias( + "gateway_transaction_id" + ), + + extract_json_field( + "after_image", + "amount" + ).cast( + "decimal(18,2)" + ).alias( + "amount" + ) + ) + ) + + +def main(): + + spark = create_spark_session() + + try: + + logger.info( + "Reading Bronze CDC events..." + ) + + bronze_df = ( + + spark.read.parquet( + BRONZE_PATH + ) + ) + + customers_df = build_customers( + bronze_df + ) + + products_df = build_products( + bronze_df + ) + + orders_df = build_orders( + bronze_df + ) + + order_items_df = build_order_items( + bronze_df + ) + + payments_df = build_payment_attempts( + bronze_df + ) + + customers_df.write.mode( + "overwrite" + ).parquet( + SILVER_CUSTOMERS_PATH + ) + + products_df.write.mode( + "overwrite" + ).parquet( + SILVER_PRODUCTS_PATH + ) + + orders_df.write.mode( + "overwrite" + ).parquet( + SILVER_ORDERS_PATH + ) + + order_items_df.write.mode( + "overwrite" + ).parquet( + SILVER_ORDER_ITEMS_PATH + ) + + payments_df.write.mode( + "overwrite" + ).parquet( + SILVER_PAYMENTS_PATH + ) + + logger.info( + "Silver layer built successfully" + ) + + finally: + + spark.stop() + + +if __name__ == "__main__": + + main() \ No newline at end of file diff --git a/submission/sample-candidate/scripts/build_warehouse.py b/submission/sample-candidate/scripts/build_warehouse.py new file mode 100644 index 0000000..9ebdeb2 --- /dev/null +++ b/submission/sample-candidate/scripts/build_warehouse.py @@ -0,0 +1,349 @@ +import logging +from pathlib import Path + +from pyspark.sql import SparkSession +from pyspark.sql import functions as F + +from pipeline.warehouse import ( + CustomerDimensionBuilder, + ProductDimensionBuilder, + FactOrderBuilder, + FactOrderItemBuilder, + FactPaymentBuilder, + WarehouseWriter +) + + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s" +) + +logger = logging.getLogger(__name__) + + +# ===================================================== +# SILVER PATHS +# ===================================================== + +SILVER_CUSTOMERS_PATH = ( + "data/silver/customers" +) + +SILVER_PRODUCTS_PATH = ( + "data/silver/products" +) + +SILVER_ORDERS_PATH = ( + "data/silver/orders" +) + +SILVER_ORDER_ITEMS_PATH = ( + "data/silver/order_items" +) + +SILVER_PAYMENTS_PATH = ( + "data/silver/payment_attempts" +) + + +# ===================================================== +# WAREHOUSE PATH +# ===================================================== + +WAREHOUSE_PATH = ( + "data/warehouse" +) + + +# ===================================================== +# SPARK +# ===================================================== + +def create_spark_session(): + + return ( + + SparkSession + + .builder + + .appName( + "Build Warehouse" + ) + + .getOrCreate() + ) + + +# ===================================================== +# EMPTY DIMENSION HELPERS +# ===================================================== + +def create_empty_customer_dimension( + spark +): + + return spark.createDataFrame( + [], + """ + customer_id BIGINT, + email STRING, + first_name STRING, + last_name STRING, + phone STRING, + customer_status STRING, + effective_from TIMESTAMP, + effective_to TIMESTAMP, + is_current BOOLEAN + """ + ) + + +def create_empty_product_dimension( + spark +): + + return spark.createDataFrame( + [], + """ + product_id BIGINT, + sku STRING, + product_name STRING, + category STRING, + unit_price DECIMAL(18,2), + product_status STRING, + effective_from TIMESTAMP, + effective_to TIMESTAMP, + is_current BOOLEAN + """ + ) + + +# ===================================================== +# LOAD EXISTING DIMENSION +# ===================================================== + +def load_existing_dimension( + spark, + path, + empty_df +): + + if Path(path).exists(): + + return spark.read.parquet(path) + + return empty_df + + +# ===================================================== +# MAIN +# ===================================================== + +def main(): + + spark = create_spark_session() + + try: + + logger.info( + "Starting warehouse build" + ) + + # ========================================== + # READ SILVER TABLES + # ========================================== + + customers_df = ( + spark.read.parquet( + SILVER_CUSTOMERS_PATH + ) + ) + + products_df = ( + spark.read.parquet( + SILVER_PRODUCTS_PATH + ) + ) + + orders_df = ( + spark.read.parquet( + SILVER_ORDERS_PATH + ) + ) + + order_items_df = ( + spark.read.parquet( + SILVER_ORDER_ITEMS_PATH + ) + ) + + payments_df = ( + spark.read.parquet( + SILVER_PAYMENTS_PATH + ) + ) + + logger.info( + "Silver tables loaded" + ) + + # ========================================== + # LOAD EXISTING DIMENSIONS + # ========================================== + + existing_customer_dim = ( + load_existing_dimension( + spark, + f"{WAREHOUSE_PATH}/dim_customer", + create_empty_customer_dimension( + spark + ) + ) + ) + + existing_product_dim = ( + load_existing_dimension( + spark, + f"{WAREHOUSE_PATH}/dim_product", + create_empty_product_dimension( + spark + ) + ) + ) + + # ========================================== + # CUSTOMER DIMENSION + # ========================================== + + customer_builder = ( + CustomerDimensionBuilder( + spark + ) + ) + + dim_customer = ( + customer_builder.build( + customers_df, + existing_customer_dim + ) + ) + + logger.info( + "Customer dimension built" + ) + + # ========================================== + # PRODUCT DIMENSION + # ========================================== + + product_builder = ( + ProductDimensionBuilder( + spark + ) + ) + + dim_product = ( + product_builder.build( + products_df, + existing_product_dim + ) + ) + + logger.info( + "Product dimension built" + ) + + # ========================================== + # FACT TABLES + # ========================================== + + fact_orders = ( + + FactOrderBuilder + + .build( + orders_df + ) + ) + + fact_order_items = ( + + FactOrderItemBuilder + + .build( + order_items_df + ) + ) + + fact_payments = ( + + FactPaymentBuilder + + .build( + payments_df + ) + ) + + logger.info( + "Fact tables built" + ) + + # ========================================== + # WRITE WAREHOUSE + # ========================================== + + writer = WarehouseWriter( + WAREHOUSE_PATH + ) + + writer.write_dimension( + dim_customer, + "dim_customer" + ) + + writer.write_dimension( + dim_product, + "dim_product" + ) + + writer.write_fact( + fact_orders, + "fact_orders" + ) + + writer.write_fact( + fact_order_items, + "fact_order_items" + ) + + writer.write_fact( + fact_payments, + "fact_payment_attempts" + ) + + logger.info( + "Warehouse tables written" + ) + + logger.info( + "Warehouse build successful" + ) + + except Exception as e: + + logger.exception( + "Warehouse build failed" + ) + + raise + + finally: + + spark.stop() + + +if __name__ == "__main__": + + main() \ No newline at end of file diff --git a/submission/sample-candidate/scripts/check_schema_contracts.py b/submission/sample-candidate/scripts/check_schema_contracts.py deleted file mode 100644 index 1698b21..0000000 --- a/submission/sample-candidate/scripts/check_schema_contracts.py +++ /dev/null @@ -1,60 +0,0 @@ -#!/usr/bin/env python3 -""" -check_schema_contracts.py - -Validates that the source tables expose all columns defined in the schema -contract. A missing column means downstream CDC logic or warehouse models -will break — this script fails the build before that happens. - -Exit 0 — all contracts pass. -Exit 1 — one or more violations found. -""" - -import os -import sys - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -import duckdb - -from source.models import SCHEMA_CONTRACT, create_source_tables - - -def check_contracts(conn: duckdb.DuckDBPyConnection) -> list[str]: - """Return a list of violation messages. Empty list = all passed.""" - violations: list[str] = [] - - for table, expected_cols in SCHEMA_CONTRACT.items(): - try: - rows = conn.execute(f"DESCRIBE {table}").fetchall() - except Exception as exc: - violations.append(f"{table}: could not describe table — {exc}") - continue - - actual_cols = {row[0] for row in rows} - - for col in expected_cols: - if col not in actual_cols: - violations.append(f"{table}.{col}: column missing from source table") - - return violations - - -def main() -> int: - conn = duckdb.connect(":memory:") - create_source_tables(conn) - - violations = check_contracts(conn) - - if violations: - print("Schema contract violations:") - for v in violations: - print(f" ✗ {v}") - return 1 - - print(f"All schema contracts passed ({len(SCHEMA_CONTRACT)} tables checked).") - return 0 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/submission/sample-candidate/scripts/ingest_cdc.py b/submission/sample-candidate/scripts/ingest_cdc.py new file mode 100644 index 0000000..e1bc11f --- /dev/null +++ b/submission/sample-candidate/scripts/ingest_cdc.py @@ -0,0 +1,203 @@ +import logging +from datetime import datetime + +from pyspark.sql import SparkSession + +from pipeline.cdc import CDCIngestion +from pipeline.bronze import BronzeLayer + + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s" +) + +logger = logging.getLogger(__name__) + + +BRONZE_PATH = "data/bronze/cdc_events" + + +def create_spark_session(): + + return ( + SparkSession.builder + .appName("CDC Ingestion") + .getOrCreate() + ) + + +def fetch_cdc_events(): + """ + Assignment Scope + + Simulates WAL CDC events. + + Production: + PostgreSQL WAL + -> + Debezium + -> + Kafka + -> + Spark + """ + + return [ + + { + "sequence_number": 1, + + "table_name": "customers", + + "operation": "INSERT", + + "primary_key": 1, + + "before_image": None, + + "after_image": { + + "customer_id": 1, + + "email": "john@example.com", + + "first_name": "John", + + "last_name": "Doe", + + "phone": "9999999999", + + "customer_status": "ACTIVE" + }, + + "event_timestamp": + datetime.now() + }, + + { + "sequence_number": 2, + + "table_name": "products", + + "operation": "INSERT", + + "primary_key": 100, + + "before_image": None, + + "after_image": { + + "product_id": 100, + + "sku": "SKU100", + + "product_name": "Laptop", + + "category": "Electronics", + + "unit_price": 50000, + + "product_status": "ACTIVE" + }, + + "event_timestamp": + datetime.now() + } + ] + + +def main(): + + spark = create_spark_session() + + try: + + logger.info( + "=" * 50 + ) + + logger.info( + "Starting CDC ingestion" + ) + + events = fetch_cdc_events() + + logger.info( + f"Fetched {len(events)} CDC events" + ) + + cdc_ingestion = CDCIngestion( + spark + ) + + cdc_df = ( + cdc_ingestion + .create_cdc_dataframe( + events + ) + ) + + bronze_layer = BronzeLayer( + spark=spark, + bronze_path=BRONZE_PATH + ) + + bronze_layer.write_events( + cdc_df + ) + + logger.info( + "CDC events written to Bronze" + ) + + try: + + bronze_df = ( + bronze_layer.read_events() + ) + + max_sequence = ( + cdc_ingestion + .latest_sequence( + bronze_df + ) + ) + + logger.info( + f"Latest sequence: " + f"{max_sequence}" + ) + + except Exception: + + logger.warning( + "Unable to determine watermark" + ) + + logger.info( + "CDC ingestion successful" + ) + + logger.info( + "=" * 50 + ) + + except Exception as e: + + logger.error( + "CDC ingestion failed" + ) + + logger.exception(e) + + raise + + finally: + + spark.stop() + + +if __name__ == "__main__": + + main() \ No newline at end of file diff --git a/submission/sample-candidate/scripts/publish_catalog.py b/submission/sample-candidate/scripts/publish_catalog.py new file mode 100644 index 0000000..1ae451b --- /dev/null +++ b/submission/sample-candidate/scripts/publish_catalog.py @@ -0,0 +1,150 @@ +import json +import logging +from pathlib import Path + + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s" +) + +logger = logging.getLogger(__name__) + + +CATALOG_FILE = "catalog/catalog.json" + + +class CatalogPublisher: + + def __init__( + self, + catalog_file + ): + + self.catalog_file = catalog_file + + def load_catalog(self): + + with open( + self.catalog_file, + "r" + ) as file: + + return json.load(file) + + def validate_catalog(self): + + catalog = self.load_catalog() + + datasets = catalog.get( + "datasets", + [] + ) + + if len(datasets) == 0: + + raise ValueError( + "Catalog contains no datasets" + ) + + for dataset in datasets: + + required_fields = [ + + "name", + "layer", + "path", + "owner", + "description", + "refresh_frequency" + ] + + for field in required_fields: + + if field not in dataset: + + raise ValueError( + f"Missing field " + f"{field} " + f"in dataset " + f"{dataset}" + ) + + return True + + def publish(self): + + catalog = self.load_catalog() + + logger.info( + "=" * 50 + ) + + logger.info( + "Publishing catalog metadata" + ) + + for dataset in catalog["datasets"]: + + logger.info( + f"Dataset: " + f"{dataset['name']}" + ) + + logger.info( + f"Layer: " + f"{dataset['layer']}" + ) + + logger.info( + f"Path: " + f"{dataset['path']}" + ) + + logger.info( + f"Owner: " + f"{dataset['owner']}" + ) + + logger.info( + f"Description: " + f"{dataset['description']}" + ) + + logger.info( + "-" * 50 + ) + + logger.info( + "Catalog successfully published" + ) + + logger.info( + "=" * 50 + ) + + +def main(): + + try: + + publisher = CatalogPublisher( + CATALOG_FILE + ) + + publisher.validate_catalog() + + publisher.publish() + + except Exception as e: + + logger.exception( + "Catalog publication failed" + ) + + raise + + +if __name__ == "__main__": + + main() \ No newline at end of file diff --git a/submission/sample-candidate/scripts/rebuild_warehouse.py b/submission/sample-candidate/scripts/rebuild_warehouse.py new file mode 100644 index 0000000..92dddd7 --- /dev/null +++ b/submission/sample-candidate/scripts/rebuild_warehouse.py @@ -0,0 +1,69 @@ +from pipeline.warehouse import ( + CustomerDimensionBuilder, + ProductDimensionBuilder, + FactOrderBuilder, + FactOrderItemBuilder, + FactPaymentBuilder +) + +from pipeline.silver import SilverLayer + + +class WarehouseRebuilder: + + def __init__( + self, + spark + ): + + self.spark = spark + + def rebuild_until_timestamp( + self, + bronze_df, + recovery_timestamp + ): + + replay_df = ( + + bronze_df + + .filter( + bronze_df.event_timestamp + <= recovery_timestamp + ) + ) + + silver = SilverLayer() + + customer_df = silver.build_customer_table( + replay_df + ) + + product_df = silver.build_product_table( + replay_df + ) + + orders_df = silver.build_orders_table( + replay_df + ) + + order_items_df = ( + silver.build_order_items_table( + replay_df + ) + ) + + payments_df = ( + silver.build_payment_attempt_table( + replay_df + ) + ) + + return { + "customers": customer_df, + "products": product_df, + "orders": orders_df, + "order_items": order_items_df, + "payments": payments_df + } \ No newline at end of file diff --git a/submission/sample-candidate/scripts/run_data_quality_checks.py b/submission/sample-candidate/scripts/run_data_quality_checks.py index 31e4cba..5e1ebb9 100644 --- a/submission/sample-candidate/scripts/run_data_quality_checks.py +++ b/submission/sample-candidate/scripts/run_data_quality_checks.py @@ -1,212 +1,192 @@ -#!/usr/bin/env python3 -""" -run_data_quality_checks.py +from pyspark.sql import SparkSession +from pyspark.sql import DataFrame +from pyspark.sql import functions as F -Runs system and business data quality validations against the warehouse. -Seeds an in-memory database with representative data, applies CDC events, -then asserts correctness invariants. -System checks : PK uniqueness, not-null, referential integrity -Business checks : non-negative balances, positive amounts, valid status enums +class DataQualityException(Exception): + pass -Exit 0 — all checks pass. -Exit 1 — one or more failures found. -""" -import os -import sys +class DataQualityChecks: -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + def create_spark_session(): + return ( + SparkSession.builder + .appName("Run Data Quality Checks") + .getOrCreate() + ) -from datetime import datetime, timezone + @staticmethod + def validate_pk_uniqueness( + df: DataFrame, + pk_column: str + ): -import duckdb + duplicates = ( -from pipeline.cdc import CDCCapture -from pipeline.lake import append_to_lake, create_lake_table -from pipeline.warehouse import apply_cdc_records, create_warehouse_tables + df.groupBy(pk_column) + .count() -def _now() -> datetime: - return datetime.now(timezone.utc) + .filter( + F.col("count") > 1 + ) + ) + if duplicates.count() > 0: -def seed(conn: duckdb.DuckDBPyConnection, capture: CDCCapture) -> None: - """Populate lake + warehouse with representative test data.""" - ts = _now() + raise DataQualityException( + f"Duplicate PK found in {pk_column}" + ) - capture.insert( - "customers", - "c1", - { - "customer_id": "c1", - "name": "Alice", - "email": "alice-test-user", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, - ) - capture.insert( - "customers", - "c2", - { - "customer_id": "c2", - "name": "Bob", - "email": "bob-test-user", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, + @staticmethod + def validate_not_null( + df: DataFrame + # column_name: str + ): + + for column_name in df.columns(): + failures = ( + + df.filter( + F.col(column_name).isNull() + ) + ) + + if failures.count() > 0: + + raise DataQualityException( + f"Null values found in {df}.{column_name}" + ) + + @staticmethod + def validate_fk( + child_df: DataFrame, + parent_df: DataFrame, + fk_column: str + ): + + invalid_records = ( + + child_df.alias("child") + + .join( + parent_df.alias("parent"), + fk_column, + "leftanti" + ) + ) + + if invalid_records.count() > 0: + + raise DataQualityException( + f"FK validation failed for {fk_column}" + ) + + @staticmethod + def validate_non_negative( + df: DataFrame, + amount_column: str + ): + + failures = ( + + df.filter( + F.col(amount_column) < 0 + ) + ) + + if failures.count() > 0: + + raise DataQualityException( + f"Negative values found in {amount_column}" + ) + + @staticmethod + def validate_order_totals( + orders_df, + order_items_df + ): + + item_totals = ( + + order_items_df + + .groupBy("order_id") + + .agg( + F.sum( + "line_amount" + ).alias( + "calculated_total" + ) + ) + ) + + failures = ( + + orders_df.alias("o") + + .join( + item_totals.alias("i"), + "order_id" + ) + + .filter( + F.col("o.total_amount") + != + F.col("i.calculated_total") + ) + ) + + if failures.count() > 0: + + raise DataQualityException( + "Order totals do not match line items" + ) + +def main(): + + spark = DataQualityChecks.create_spark_session() + + dim_customer = spark.read.parquet( + "data/warehouse/dim_customer" ) - capture.insert( - "wallets", - "w1", - { - "wallet_id": "w1", - "customer_id": "c1", - "balance": 100.00, - "currency": "USD", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, + dim_product = spark.read.parquet( + "data/warehouse/dim_product" ) - capture.insert( - "wallets", - "w2", - { - "wallet_id": "w2", - "customer_id": "c2", - "balance": 50.00, - "currency": "USD", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, + + fact_orders = spark.read.parquet( + "data/warehouse/fact_orders" ) - capture.insert( - "transactions", - "t1", - { - "transaction_id": "t1", - "wallet_id": "w1", - "amount": 25.00, - "direction": "credit", - "status": "settled", - "reference": "ref-001", - "created_at": ts, - "settled_at": ts, - }, + fact_order_items = spark.read.parquet( + "data/warehouse/fact_order_items" ) - capture.insert( - "transactions", - "t2", - { - "transaction_id": "t2", - "wallet_id": "w2", - "amount": 10.00, - "direction": "debit", - "status": "settled", - "reference": "ref-002", - "created_at": ts, - "settled_at": ts, - }, + + fact_payments = spark.read.parquet( + "data/warehouse/fact_payment_attempts" ) - create_lake_table(conn) - create_warehouse_tables(conn) - records = capture.records_since(0) - append_to_lake(conn, records) - apply_cdc_records(conn, records) - - -def run_checks(conn: duckdb.DuckDBPyConnection) -> list[str]: - failures: list[str] = [] - - # ── system checks ───────────────────────────────────────────────────────── - - pk_map = { - "wh_customers": "customer_id", - "wh_wallets": "wallet_id", - "wh_transactions": "transaction_id", - } - for table, pk in pk_map.items(): - total = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] - distinct = conn.execute(f"SELECT COUNT(DISTINCT {pk}) FROM {table}").fetchone()[ - 0 - ] - if total != distinct: - failures.append( - f"{table}: PK not unique — {total} rows, {distinct} distinct {pk}" - ) + #Add the Data Quality checks for the required warehouse tables + - not_null_checks = [ - ("wh_customers", "name"), - ("wh_customers", "email"), - ("wh_wallets", "customer_id"), - ("wh_wallets", "currency"), - ("wh_transactions", "wallet_id"), - ("wh_transactions", "amount"), - ("wh_transactions", "direction"), - ] - for table, col in not_null_checks: - nulls = conn.execute( - f"SELECT COUNT(*) FROM {table} WHERE {col} IS NULL" - ).fetchone()[0] - if nulls: - failures.append(f"{table}.{col}: {nulls} NULL value(s) found") - - # ── business checks ─────────────────────────────────────────────────────── - - neg_bal = conn.execute( - "SELECT COUNT(*) FROM wh_wallets WHERE balance < 0" - ).fetchone()[0] - if neg_bal: - failures.append(f"wh_wallets: {neg_bal} row(s) with negative balance") - - non_pos = conn.execute( - "SELECT COUNT(*) FROM wh_transactions WHERE amount <= 0" - ).fetchone()[0] - if non_pos: - failures.append(f"wh_transactions: {non_pos} row(s) with non-positive amount") - - bad_dir = conn.execute( - "SELECT COUNT(*) FROM wh_transactions" - " WHERE direction NOT IN ('credit', 'debit')" - ).fetchone()[0] - if bad_dir: - failures.append(f"wh_transactions: {bad_dir} row(s) with invalid direction") - - # ── lake completeness ───────────────────────────────────────────────────── - - lake_count = conn.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] - if lake_count == 0: - failures.append("lake_cdc_events: no records found — lake appears empty") - - return failures - - -def main() -> int: - conn = duckdb.connect(":memory:") - capture = CDCCapture() - seed(conn, capture) - - failures = run_checks(conn) - - if failures: - print("Data quality failures:") - for f in failures: - print(f" ✗ {f}") - return 1 - - print( - f"All data quality checks passed ({len(run_checks.__code__.co_consts)} rules checked)." + DataQualityChecks.validate_pk_uniqueness( + dim_customer, + "customer_id" ) - return 0 + DataQualityChecks.validate_fk( + fact_orders, + dim_customer, + "customer_id" + ) + + DataQualityChecks.validate_order_totals( + fact_orders, + fact_order_items + ) if __name__ == "__main__": - sys.exit(main()) + main() \ No newline at end of file diff --git a/submission/sample-candidate/scripts/status_transition_validation.py b/submission/sample-candidate/scripts/status_transition_validation.py new file mode 100644 index 0000000..d58c508 --- /dev/null +++ b/submission/sample-candidate/scripts/status_transition_validation.py @@ -0,0 +1,168 @@ +from pyspark.sql import DataFrame +from pyspark.sql import functions as F +import json + + +class StatusTransitionException(Exception): + pass + + +class StatusTransitionValidator: + """ + Validates order status transitions from CDC events. + + Examples: + + VALID: + CREATED -> PAID + CREATED -> CANCELLED + PAID -> SHIPPED + SHIPPED -> DELIVERED + + INVALID: + DELIVERED -> CREATED + DELIVERED -> PAID + CANCELLED -> PAID + """ + + ALLOWED_TRANSITIONS = { + + "CREATED": [ + "PAID", + "CANCELLED" + ], + + "PAID": [ + "SHIPPED" + ], + + "SHIPPED": [ + "DELIVERED" + ], + + "DELIVERED": [], + + "CANCELLED": [] + } + + @staticmethod + def _extract_status(json_string): + + if json_string is None: + return None + + try: + payload = json.loads(json_string) + + return payload.get("order_status") + + except Exception: + return None + + @classmethod + def validate( + cls, + cdc_df: DataFrame + ): + """ + Validates order status transitions from CDC events. + + Expected CDC schema: + + sequence_number + table_name + operation + primary_key + before_image + after_image + event_timestamp + """ + + extract_before_status = F.udf( + lambda x: cls._extract_status(x) + ) + + extract_after_status = F.udf( + lambda x: cls._extract_status(x) + ) + + order_updates = ( + + cdc_df + + .filter( + F.col("table_name") == "orders" + ) + + .filter( + F.col("operation") == "UPDATE" + ) + + .withColumn( + "old_status", + extract_before_status( + F.col("before_image") + ) + ) + + .withColumn( + "new_status", + extract_after_status( + F.col("after_image") + ) + ) + ) + + invalid_transitions = [] + + rows = order_updates.select( + "primary_key", + "old_status", + "new_status" + ).collect() + + for row in rows: + + old_status = row["old_status"] + new_status = row["new_status"] + + if old_status is None: + continue + + allowed = cls.ALLOWED_TRANSITIONS.get( + old_status, + [] + ) + + if new_status not in allowed: + + invalid_transitions.append( + ( + row["primary_key"], + old_status, + new_status + ) + ) + + if invalid_transitions: + + error_message = "\n".join( + [ + ( + f"Order {order_id}: " + f"{old_status} -> {new_status}" + ) + for ( + order_id, + old_status, + new_status + ) in invalid_transitions + ] + ) + + raise StatusTransitionException( + f"Invalid status transitions detected:\n" + f"{error_message}" + ) + + return True \ No newline at end of file diff --git a/submission/sample-candidate/scripts/validate_catalog.py b/submission/sample-candidate/scripts/validate_catalog.py deleted file mode 100644 index 6a02026..0000000 --- a/submission/sample-candidate/scripts/validate_catalog.py +++ /dev/null @@ -1,77 +0,0 @@ -#!/usr/bin/env python3 -""" -validate_catalog.py - -Verifies that catalog/catalog.json is present and contains a complete entry -for every required lake and warehouse dataset. - -Required fields per entry: name, layer, description, owner, schema, update_cadence - -Exit 0 — catalog is valid. -Exit 1 — missing file, missing datasets, or missing required fields. -""" - -import json -import os -import sys - -CATALOG_PATH = os.path.join( - os.path.dirname(os.path.dirname(os.path.abspath(__file__))), - "catalog", - "catalog.json", -) - -REQUIRED_DATASETS = [ - "lake_cdc_events", - "wh_customers", - "wh_wallets", - "wh_transactions", -] - -REQUIRED_FIELDS = ["name", "layer", "description", "owner", "schema", "update_cadence"] - - -def validate() -> list[str]: - violations: list[str] = [] - - if not os.path.exists(CATALOG_PATH): - violations.append(f"catalog.json not found at {CATALOG_PATH}") - return violations - - with open(CATALOG_PATH) as f: - try: - catalog = json.load(f) - except json.JSONDecodeError as exc: - violations.append(f"catalog.json is not valid JSON: {exc}") - return violations - - datasets = {d["name"]: d for d in catalog.get("datasets", []) if "name" in d} - - for name in REQUIRED_DATASETS: - if name not in datasets: - violations.append(f"Missing dataset entry: {name}") - continue - - entry = datasets[name] - for field in REQUIRED_FIELDS: - if field not in entry or not entry[field]: - violations.append(f"{name}: missing or empty required field '{field}'") - - return violations - - -def main() -> int: - violations = validate() - - if violations: - print("Catalog validation failures:") - for v in violations: - print(f" ✗ {v}") - return 1 - - print(f"Catalog validation passed ({len(REQUIRED_DATASETS)} datasets verified).") - return 0 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/submission/sample-candidate/scripts/validate_schema.py b/submission/sample-candidate/scripts/validate_schema.py new file mode 100644 index 0000000..4c49388 --- /dev/null +++ b/submission/sample-candidate/scripts/validate_schema.py @@ -0,0 +1,149 @@ +import json +import logging + +from pipeline.schema_contracts import ( + SchemaValidator, + SchemaContractException +) + +from source.postgres import ( + PostgresMetadataReader +) + + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s" +) + +logger = logging.getLogger(__name__) + + +SCHEMA_CONTRACT_FILE = ( + "config/schema_contracts.json" +) + +ENUM_CONTRACT_FILE = ( + "config/enum_contracts.json" +) + + +def load_json(file_path): + + with open(file_path, "r") as f: + + return json.load(f) + + +def validate_table_schemas(): + + logger.info( + "Loading schema contracts..." + ) + + expected_contracts = load_json( + SCHEMA_CONTRACT_FILE + ) + + metadata_reader = ( + PostgresMetadataReader() + ) + + actual_contracts = ( + metadata_reader.get_table_schemas() + ) + + validator = SchemaValidator( + expected_contracts + ) + + validator.validate( + actual_contracts + ) + + logger.info( + "Schema validation successful." + ) + + +def validate_enum_domains(): + + logger.info( + "Loading enum contracts..." + ) + + expected_enums = load_json( + ENUM_CONTRACT_FILE + ) + + metadata_reader = ( + PostgresMetadataReader() + ) + + actual_enums = ( + metadata_reader.get_enum_domains() + ) + + for enum_name, expected_values in ( + expected_enums.items() + ): + + actual_values = actual_enums.get( + enum_name, + [] + ) + + if set(actual_values) != set( + expected_values + ): + + raise SchemaContractException( + f"Enum contract violation " + f"for {enum_name}. " + f"Expected={expected_values}, " + f"Actual={actual_values}" + ) + + logger.info( + "Enum validation successful." + ) + + +def main(): + + try: + + logger.info( + "=" * 50 + ) + + logger.info( + "Starting schema validation" + ) + + validate_table_schemas() + + validate_enum_domains() + + logger.info( + "Schema validation passed" + ) + + logger.info( + "=" * 50 + ) + + except Exception as e: + + logger.error( + "Schema validation failed" + ) + + logger.exception(e) + + raise + + +if __name__ == "__main__": + + main() \ No newline at end of file diff --git a/submission/sample-candidate/source/__init__.py b/submission/sample-candidate/source/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/submission/sample-candidate/source/__pycache__/__init__.cpython-314.pyc b/submission/sample-candidate/source/__pycache__/__init__.cpython-314.pyc deleted file mode 100644 index b3cc59d..0000000 Binary files a/submission/sample-candidate/source/__pycache__/__init__.cpython-314.pyc and /dev/null differ diff --git a/submission/sample-candidate/source/__pycache__/models.cpython-314.pyc b/submission/sample-candidate/source/__pycache__/models.cpython-314.pyc deleted file mode 100644 index 2cd0f91..0000000 Binary files a/submission/sample-candidate/source/__pycache__/models.cpython-314.pyc and /dev/null differ diff --git a/submission/sample-candidate/source/create_tables.py b/submission/sample-candidate/source/create_tables.py new file mode 100644 index 0000000..fc90c5e --- /dev/null +++ b/submission/sample-candidate/source/create_tables.py @@ -0,0 +1,150 @@ +from pathlib import Path + +from pyspark.sql import SparkSession + + +SOURCE_PATH = "data/source" + + +def create_spark_session(): + + return ( + SparkSession.builder + .appName("Create Source Tables") + .getOrCreate() + ) + + +def create_customers_table(spark): + + df = spark.createDataFrame( + [], + """ + customer_id BIGINT, + email STRING, + first_name STRING, + last_name STRING, + phone STRING, + customer_status STRING, + created_at TIMESTAMP, + updated_at TIMESTAMP + """ + ) + + df.write.mode("overwrite").parquet( + f"{SOURCE_PATH}/customers" + ) + + +def create_products_table(spark): + + df = spark.createDataFrame( + [], + """ + product_id BIGINT, + sku STRING, + product_name STRING, + category STRING, + unit_price DECIMAL(18,2), + product_status STRING, + created_at TIMESTAMP, + updated_at TIMESTAMP + """ + ) + + df.write.mode("overwrite").parquet( + f"{SOURCE_PATH}/products" + ) + + +def create_orders_table(spark): + + df = spark.createDataFrame( + [], + """ + order_id BIGINT, + customer_id BIGINT, + order_status STRING, + total_amount DECIMAL(18,2), + created_at TIMESTAMP, + updated_at TIMESTAMP + """ + ) + + df.write.mode("overwrite").parquet( + f"{SOURCE_PATH}/orders" + ) + + +def create_order_items_table(spark): + + df = spark.createDataFrame( + [], + """ + order_item_id BIGINT, + order_id BIGINT, + product_id BIGINT, + quantity INTEGER, + unit_price DECIMAL(18,2), + line_amount DECIMAL(18,2), + created_at TIMESTAMP + """ + ) + + df.write.mode("overwrite").parquet( + f"{SOURCE_PATH}/order_items" + ) + + +def create_payment_attempts_table(spark): + + df = spark.createDataFrame( + [], + """ + payment_attempt_id BIGINT, + order_id BIGINT, + payment_status STRING, + gateway_transaction_id STRING, + amount DECIMAL(18,2), + attempt_timestamp TIMESTAMP + """ + ) + + df.write.mode("overwrite").parquet( + f"{SOURCE_PATH}/payment_attempts" + ) + + +def main(): + + spark = create_spark_session() + + try: + + Path(SOURCE_PATH).mkdir( + parents=True, + exist_ok=True + ) + + create_customers_table(spark) + + create_products_table(spark) + + create_orders_table(spark) + + create_order_items_table(spark) + + create_payment_attempts_table(spark) + + print( + "Source tables created successfully." + ) + + finally: + + spark.stop() + + +if __name__ == "__main__": + + main() \ No newline at end of file diff --git a/submission/sample-candidate/source/models.py b/submission/sample-candidate/source/models.py index 91c9d19..8c2c5af 100644 --- a/submission/sample-candidate/source/models.py +++ b/submission/sample-candidate/source/models.py @@ -1,84 +1,190 @@ -""" -Source schema for a payments/wallet system. - -Domain: customers own wallets; wallets have transactions. - -Strong entities : customers, wallets -Weak entities : transactions (lifecycle tied to wallet) - -Invariants: -- wallet.balance >= 0 -- transaction.amount > 0 -- status fields restricted to known enum values -- settled_at must be >= created_at when present -""" - -import duckdb - -# Expected columns per table — used by schema-contract checks. -SCHEMA_CONTRACT: dict[str, list[str]] = { - "customers": ["customer_id", "name", "email", "status", "created_at", "updated_at"], - "wallets": [ - "wallet_id", - "customer_id", - "balance", - "currency", - "status", - "created_at", - "updated_at", - ], - "transactions": [ - "transaction_id", - "wallet_id", - "amount", - "direction", - "status", - "reference", - "created_at", - "settled_at", - ], -} - - -def create_source_tables(conn: duckdb.DuckDBPyConnection) -> None: - """Create source tables with constraints in the given DuckDB 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 wallets ( - wallet_id VARCHAR PRIMARY KEY, - customer_id VARCHAR NOT NULL REFERENCES customers(customer_id), - balance DECIMAL(18, 2) NOT NULL DEFAULT 0.00 - CHECK (balance >= 0), - currency VARCHAR NOT NULL, - status VARCHAR NOT NULL - CHECK (status IN ('active', 'frozen', 'closed')), - created_at TIMESTAMP NOT NULL, - updated_at TIMESTAMP NOT NULL - ) - """) - - conn.execute(""" - CREATE TABLE IF NOT EXISTS transactions ( - transaction_id VARCHAR PRIMARY KEY, - wallet_id VARCHAR NOT NULL REFERENCES wallets(wallet_id), - amount DECIMAL(18, 2) NOT NULL CHECK (amount > 0), - direction VARCHAR NOT NULL - CHECK (direction IN ('credit', 'debit')), - status VARCHAR NOT NULL - CHECK (status IN ('pending', 'settled', 'failed', 'reversed')), - reference VARCHAR, - created_at TIMESTAMP NOT NULL, - settled_at TIMESTAMP - ) - """) +from dataclasses import dataclass +from datetime import datetime +from decimal import Decimal +from enum import Enum +from typing import Optional, Dict, Any + + +# ===================================================== +# ENUMS +# ===================================================== + +class CustomerStatus(str, Enum): + ACTIVE = "ACTIVE" + INACTIVE = "INACTIVE" + SUSPENDED = "SUSPENDED" + + +class ProductStatus(str, Enum): + ACTIVE = "ACTIVE" + DISCONTINUED = "DISCONTINUED" + + +class OrderStatus(str, Enum): + CREATED = "CREATED" + PAID = "PAID" + SHIPPED = "SHIPPED" + DELIVERED = "DELIVERED" + CANCELLED = "CANCELLED" + + +class PaymentStatus(str, Enum): + SUCCESS = "SUCCESS" + FAILED = "FAILED" + PENDING = "PENDING" + + +class CDCOperation(str, Enum): + INSERT = "INSERT" + UPDATE = "UPDATE" + DELETE = "DELETE" + + +# ===================================================== +# STRONG ENTITIES +# ===================================================== + +@dataclass +class Customer: + + customer_id: int + + email: str + + first_name: str + + last_name: Optional[str] + + phone: Optional[str] + + customer_status: CustomerStatus + + created_at: datetime + + updated_at: datetime + + +@dataclass +class Product: + + product_id: int + + sku: str + + product_name: str + + category: Optional[str] + + unit_price: Decimal + + product_status: ProductStatus + + created_at: datetime + + updated_at: datetime + + +@dataclass +class Order: + + order_id: int + + customer_id: int + + order_status: OrderStatus + + total_amount: Decimal + + created_at: datetime + + updated_at: datetime + + +# ===================================================== +# WEAK ENTITIES +# ===================================================== + +@dataclass +class OrderItem: + + order_item_id: int + + order_id: int + + product_id: int + + quantity: int + + unit_price: Decimal + + line_amount: Decimal + + created_at: datetime + + +@dataclass +class PaymentAttempt: + + payment_attempt_id: int + + order_id: int + + payment_status: PaymentStatus + + gateway_transaction_id: Optional[str] + + amount: Decimal + + attempt_timestamp: datetime + + +# ===================================================== +# CDC EVENT +# ===================================================== + +@dataclass +class CDCEvent: + + sequence_number: int + + table_name: str + + operation: CDCOperation + + primary_key: Any + + before_image: Optional[Dict[str, Any]] + + after_image: Optional[Dict[str, Any]] + + event_timestamp: datetime + + +# ===================================================== +# SCHEMA CONTRACT +# ===================================================== + +@dataclass +class SchemaContract: + + table_name: str + + columns: Dict[str, str] + + +# ===================================================== +# DATA QUALITY FAILURE +# ===================================================== + +@dataclass +class ValidationFailure: + + rule_name: str + + table_name: str + + record_id: str + + failure_reason: str + + event_timestamp: datetime \ No newline at end of file diff --git a/submission/sample-candidate/source/postgres.py b/submission/sample-candidate/source/postgres.py new file mode 100644 index 0000000..c607518 --- /dev/null +++ b/submission/sample-candidate/source/postgres.py @@ -0,0 +1,121 @@ +class PostgresMetadataReader: + + def get_table_schemas(self): + + return { + + "customers": { + + "customer_id": "bigint", + + "email": "string", + + "first_name": "string", + + "last_name": "string", + + "phone": "string", + + "customer_status": "string", + + "created_at": "timestamp", + + "updated_at": "timestamp" + }, + + "products": { + + "product_id": "bigint", + + "sku": "string", + + "product_name": "string", + + "category": "string", + + "unit_price": "decimal(18,2)", + + "product_status": "string", + + "created_at": "timestamp", + + "updated_at": "timestamp" + }, + + "orders": { + + "order_id": "bigint", + + "customer_id": "bigint", + + "order_status": "string", + + "total_amount": "decimal(18,2)", + + "created_at": "timestamp", + + "updated_at": "timestamp" + }, + + "order_items": { + + "order_item_id": "bigint", + + "order_id": "bigint", + + "product_id": "bigint", + + "quantity": "integer", + + "unit_price": "decimal(18,2)", + + "line_amount": "decimal(18,2)", + + "created_at": "timestamp" + }, + + "payment_attempts": { + + "payment_attempt_id": "bigint", + + "order_id": "bigint", + + "payment_status": "string", + + "gateway_transaction_id": "string", + + "amount": "decimal(18,2)", + + "attempt_timestamp": "timestamp" + } + } + + def get_enum_domains(self): + + return { + + "customer_status": [ + "ACTIVE", + "INACTIVE", + "SUSPENDED" + ], + + "product_status": [ + "ACTIVE", + "DISCONTINUED" + ], + + "order_status": [ + "CREATED", + "PAID", + "SHIPPED", + "DELIVERED", + "CANCELLED" + ], + + "payment_status": [ + "SUCCESS", + "FAILED", + "PENDING" + ] + } \ No newline at end of file diff --git a/submission/sample-candidate/source/seed_data.py b/submission/sample-candidate/source/seed_data.py new file mode 100644 index 0000000..cce82ba --- /dev/null +++ b/submission/sample-candidate/source/seed_data.py @@ -0,0 +1,243 @@ +from decimal import Decimal +from datetime import datetime + +from pyspark.sql import SparkSession + + +SOURCE_PATH = "data/source" + + +def create_spark_session(): + + return ( + SparkSession.builder + .appName("Seed Source Data") + .getOrCreate() + ) + + +def seed_customers(spark): + + customers = [ + + ( + 1, + "john@example.com", + "John", + "Doe", + "9999999999", + "ACTIVE", + datetime.now(), + datetime.now() + ), + + ( + 2, + "alice@example.com", + "Alice", + "Smith", + None, + "ACTIVE", + datetime.now(), + datetime.now() + ) + ] + + df = spark.createDataFrame( + customers, + [ + "customer_id", + "email", + "first_name", + "last_name", + "phone", + "customer_status", + "created_at", + "updated_at" + ] + ) + + df.write.mode("overwrite").parquet( + f"{SOURCE_PATH}/customers" + ) + + +def seed_products(spark): + + products = [ + + ( + 100, + "SKU100", + "Laptop", + "Electronics", + Decimal("50000.00"), + "ACTIVE", + datetime.now(), + datetime.now() + ), + + ( + 101, + "SKU101", + "Mouse", + "Accessories", + Decimal("1500.00"), + "ACTIVE", + datetime.now(), + datetime.now() + ) + ] + + df = spark.createDataFrame( + products, + [ + "product_id", + "sku", + "product_name", + "category", + "unit_price", + "product_status", + "created_at", + "updated_at" + ] + ) + + df.write.mode("overwrite").parquet( + f"{SOURCE_PATH}/products" + ) + + +def seed_orders(spark): + + orders = [ + + ( + 1000, + 1, + "PAID", + Decimal("51500.00"), + datetime.now(), + datetime.now() + ) + ] + + df = spark.createDataFrame( + orders, + [ + "order_id", + "customer_id", + "order_status", + "total_amount", + "created_at", + "updated_at" + ] + ) + + df.write.mode("overwrite").parquet( + f"{SOURCE_PATH}/orders" + ) + + +def seed_order_items(spark): + + items = [ + + ( + 1, + 1000, + 100, + 1, + Decimal("50000.00"), + Decimal("50000.00"), + datetime.now() + ), + + ( + 2, + 1000, + 101, + 1, + Decimal("1500.00"), + Decimal("1500.00"), + datetime.now() + ) + ] + + df = spark.createDataFrame( + items, + [ + "order_item_id", + "order_id", + "product_id", + "quantity", + "unit_price", + "line_amount", + "created_at" + ] + ) + + df.write.mode("overwrite").parquet( + f"{SOURCE_PATH}/order_items" + ) + + +def seed_payment_attempts(spark): + + payments = [ + + ( + 1, + 1000, + "SUCCESS", + "TXN123456", + Decimal("51500.00"), + datetime.now() + ) + ] + + df = spark.createDataFrame( + payments, + [ + "payment_attempt_id", + "order_id", + "payment_status", + "gateway_transaction_id", + "amount", + "attempt_timestamp" + ] + ) + + df.write.mode("overwrite").parquet( + f"{SOURCE_PATH}/payment_attempts" + ) + + +def main(): + + spark = create_spark_session() + + try: + + seed_customers(spark) + + seed_products(spark) + + seed_orders(spark) + + seed_order_items(spark) + + seed_payment_attempts(spark) + + print( + "Source data seeded successfully." + ) + + finally: + + spark.stop() + + +if __name__ == "__main__": + + main() \ No newline at end of file diff --git a/submission/sample-candidate/tests/__pycache__/conftest.cpython-314-pytest-9.0.2.pyc b/submission/sample-candidate/tests/__pycache__/conftest.cpython-314-pytest-9.0.2.pyc deleted file mode 100644 index d04dfd0..0000000 Binary files a/submission/sample-candidate/tests/__pycache__/conftest.cpython-314-pytest-9.0.2.pyc and /dev/null differ diff --git a/submission/sample-candidate/tests/__pycache__/test_catalog.cpython-314-pytest-9.0.2.pyc b/submission/sample-candidate/tests/__pycache__/test_catalog.cpython-314-pytest-9.0.2.pyc deleted file mode 100644 index 5ee79d6..0000000 Binary files a/submission/sample-candidate/tests/__pycache__/test_catalog.cpython-314-pytest-9.0.2.pyc and /dev/null differ diff --git a/submission/sample-candidate/tests/__pycache__/test_cdc.cpython-314-pytest-9.0.2.pyc b/submission/sample-candidate/tests/__pycache__/test_cdc.cpython-314-pytest-9.0.2.pyc deleted file mode 100644 index a287f0a..0000000 Binary files a/submission/sample-candidate/tests/__pycache__/test_cdc.cpython-314-pytest-9.0.2.pyc and /dev/null differ diff --git a/submission/sample-candidate/tests/__pycache__/test_data_quality.cpython-314-pytest-9.0.2.pyc b/submission/sample-candidate/tests/__pycache__/test_data_quality.cpython-314-pytest-9.0.2.pyc deleted file mode 100644 index cbd99cf..0000000 Binary files a/submission/sample-candidate/tests/__pycache__/test_data_quality.cpython-314-pytest-9.0.2.pyc and /dev/null differ diff --git a/submission/sample-candidate/tests/__pycache__/test_schema_contracts.cpython-314-pytest-9.0.2.pyc b/submission/sample-candidate/tests/__pycache__/test_schema_contracts.cpython-314-pytest-9.0.2.pyc deleted file mode 100644 index 76d6304..0000000 Binary files a/submission/sample-candidate/tests/__pycache__/test_schema_contracts.cpython-314-pytest-9.0.2.pyc and /dev/null differ diff --git a/submission/sample-candidate/tests/conftest.py b/submission/sample-candidate/tests/conftest.py deleted file mode 100644 index 7a883e5..0000000 --- a/submission/sample-candidate/tests/conftest.py +++ /dev/null @@ -1,125 +0,0 @@ -"""Shared pytest fixtures for the payments CDC pipeline tests.""" - -from datetime import datetime, timezone - -import duckdb -import pytest - -from pipeline.cdc import CDCCapture -from pipeline.lake import append_to_lake, create_lake_table -from pipeline.warehouse import apply_cdc_records, create_warehouse_tables -from source.models import create_source_tables - - -def _ts() -> datetime: - return datetime.now(timezone.utc) - - -@pytest.fixture() -def conn() -> duckdb.DuckDBPyConnection: - """Fresh in-memory DuckDB with source + lake + warehouse tables.""" - c = duckdb.connect(":memory:") - create_source_tables(c) - create_lake_table(c) - create_warehouse_tables(c) - return c - - -@pytest.fixture() -def capture() -> CDCCapture: - """Empty CDC capture log.""" - return CDCCapture() - - -@pytest.fixture() -def seeded(conn: duckdb.DuckDBPyConnection, capture: CDCCapture): - """ - DuckDB connection pre-loaded with two customers, two wallets, - and two transactions — all flushed through lake and warehouse. - Returns (conn, capture). - """ - ts = _ts() - - capture.insert( - "customers", - "c1", - { - "customer_id": "c1", - "name": "Alice", - "email": "alice-test-user", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, - ) - capture.insert( - "customers", - "c2", - { - "customer_id": "c2", - "name": "Bob", - "email": "bob-test-user", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, - ) - capture.insert( - "wallets", - "w1", - { - "wallet_id": "w1", - "customer_id": "c1", - "balance": 100.00, - "currency": "USD", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, - ) - capture.insert( - "wallets", - "w2", - { - "wallet_id": "w2", - "customer_id": "c2", - "balance": 50.00, - "currency": "USD", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, - ) - capture.insert( - "transactions", - "t1", - { - "transaction_id": "t1", - "wallet_id": "w1", - "amount": 25.00, - "direction": "credit", - "status": "settled", - "reference": "ref-001", - "created_at": ts, - "settled_at": ts, - }, - ) - capture.insert( - "transactions", - "t2", - { - "transaction_id": "t2", - "wallet_id": "w2", - "amount": 10.00, - "direction": "debit", - "status": "settled", - "reference": "ref-002", - "created_at": ts, - "settled_at": ts, - }, - ) - - records = capture.records_since(0) - append_to_lake(conn, records) - apply_cdc_records(conn, records) - return conn, capture diff --git a/submission/sample-candidate/tests/test_catalog.py b/submission/sample-candidate/tests/test_catalog.py deleted file mode 100644 index 9f7d056..0000000 --- a/submission/sample-candidate/tests/test_catalog.py +++ /dev/null @@ -1,134 +0,0 @@ -""" -Tests — catalog metadata completeness and correctness. - -Covers: file presence, all required datasets, required fields, layer labels, -schema presence, and that lake/warehouse datasets are correctly distinguished. -""" - -import json -import os - -import pytest - -CATALOG_PATH = os.path.join( - os.path.dirname(os.path.dirname(__file__)), - "catalog", - "catalog.json", -) - -REQUIRED_DATASETS = [ - "lake_cdc_events", - "wh_customers", - "wh_wallets", - "wh_transactions", -] - -REQUIRED_FIELDS = ["name", "layer", "description", "owner", "schema", "update_cadence"] - - -@pytest.fixture(scope="module") -def catalog() -> dict: - with open(CATALOG_PATH) as f: - return json.load(f) - - -@pytest.fixture(scope="module") -def datasets(catalog: dict) -> dict[str, dict]: - return {d["name"]: d for d in catalog.get("datasets", []) if "name" in d} - - -# ── file presence ───────────────────────────────────────────────────────────── - - -def test_catalog_file_exists(): - assert os.path.exists(CATALOG_PATH), f"catalog.json not found at {CATALOG_PATH}" - - -def test_catalog_file_is_valid_json(): - with open(CATALOG_PATH) as f: - data = json.load(f) - assert isinstance(data, dict) - - -def test_catalog_has_datasets_key(catalog: dict): - assert "datasets" in catalog - assert isinstance(catalog["datasets"], list) - - -# ── required datasets ───────────────────────────────────────────────────────── - - -@pytest.mark.parametrize("name", REQUIRED_DATASETS) -def test_required_dataset_is_present(name: str, datasets: dict): - assert name in datasets, f"Dataset '{name}' is missing from catalog" - - -def test_catalog_has_at_least_four_datasets(datasets: dict): - assert len(datasets) >= 4 - - -# ── required fields ─────────────────────────────────────────────────────────── - - -@pytest.mark.parametrize("name", REQUIRED_DATASETS) -@pytest.mark.parametrize("field", REQUIRED_FIELDS) -def test_required_field_is_present_and_non_empty(name: str, field: str, datasets: dict): - entry = datasets.get(name, {}) - assert field in entry, f"Dataset '{name}' is missing field '{field}'" - assert entry[field], f"Dataset '{name}' field '{field}' is empty" - - -# ── layer labels ────────────────────────────────────────────────────────────── - - -def test_lake_cdc_events_is_labeled_as_lake(datasets: dict): - assert datasets["lake_cdc_events"]["layer"] == "lake" - - -@pytest.mark.parametrize( - "name", - ["wh_customers", "wh_wallets", "wh_transactions"], -) -def test_warehouse_datasets_are_labeled_as_warehouse(name: str, datasets: dict): - assert datasets[name]["layer"] == "warehouse" - - -# ── schema presence ─────────────────────────────────────────────────────────── - - -@pytest.mark.parametrize("name", REQUIRED_DATASETS) -def test_schema_field_is_a_non_empty_dict(name: str, datasets: dict): - schema = datasets[name].get("schema", {}) - assert isinstance(schema, dict) - assert len(schema) > 0, f"Dataset '{name}' has an empty schema" - - -def test_lake_schema_includes_operation_column(datasets: dict): - assert "operation" in datasets["lake_cdc_events"]["schema"] - - -def test_lake_schema_includes_sequence_column(datasets: dict): - assert "sequence" in datasets["lake_cdc_events"]["schema"] - - -def test_wh_customers_schema_includes_pk(datasets: dict): - assert "customer_id" in datasets["wh_customers"]["schema"] - - -def test_wh_transactions_schema_includes_amount(datasets: dict): - assert "amount" in datasets["wh_transactions"]["schema"] - - -# ── update cadence ──────────────────────────────────────────────────────────── - - -def test_lake_update_cadence_is_real_time(datasets: dict): - assert datasets["lake_cdc_events"]["update_cadence"] == "real-time" - - -@pytest.mark.parametrize( - "name", - ["wh_customers", "wh_wallets", "wh_transactions"], -) -def test_warehouse_update_cadence_is_near_real_time(name: str, datasets: dict): - assert datasets[name]["update_cadence"] == "near-real-time" diff --git a/submission/sample-candidate/tests/test_cdc.py b/submission/sample-candidate/tests/test_cdc.py deleted file mode 100644 index 704902e..0000000 --- a/submission/sample-candidate/tests/test_cdc.py +++ /dev/null @@ -1,135 +0,0 @@ -""" -Tests — CDC capture correctness. - -Covers: insert/update/delete capture, invalid operation rejection, -sequence monotonicity, checkpoint-based replay, duplicate safety. -""" - -from datetime import datetime, timezone - -import pytest - -from pipeline.cdc import CDCCapture, CDCRecord - - -def _ts() -> datetime: - return datetime.now(timezone.utc) - - -# ── insert ──────────────────────────────────────────────────────────────────── - - -def test_insert_creates_record_with_correct_operation(): - cap = CDCCapture() - rec = cap.insert("customers", "c1", {"customer_id": "c1", "name": "Alice"}) - assert rec.operation == "insert" - assert rec.table == "customers" - assert rec.primary_key == "c1" - assert rec.data["name"] == "Alice" - - -def test_insert_assigns_sequence_starting_at_one(): - cap = CDCCapture() - rec = cap.insert("customers", "c1", {}) - assert rec.sequence == 1 - - -# ── update ──────────────────────────────────────────────────────────────────── - - -def test_update_creates_record_with_update_operation(): - cap = CDCCapture() - cap.insert("customers", "c1", {"status": "active"}) - rec = cap.update("customers", "c1", {"status": "suspended"}) - assert rec.operation == "update" - assert rec.data["status"] == "suspended" - - -# ── delete ──────────────────────────────────────────────────────────────────── - - -def test_delete_creates_record_with_delete_operation(): - cap = CDCCapture() - cap.insert("customers", "c1", {"customer_id": "c1"}) - rec = cap.delete("customers", "c1", {"customer_id": "c1"}) - assert rec.operation == "delete" - assert rec.primary_key == "c1" - - -# ── invalid operation ───────────────────────────────────────────────────────── - - -def test_invalid_operation_raises_value_error(): - with pytest.raises(ValueError, match="Invalid CDC operation"): - CDCRecord(operation="upsert", table="customers", primary_key="c1", data={}) - - -# ── sequence monotonicity ───────────────────────────────────────────────────── - - -def test_sequences_are_strictly_increasing(): - cap = CDCCapture() - r1 = cap.insert("customers", "c1", {}) - r2 = cap.insert("customers", "c2", {}) - r3 = cap.update("customers", "c1", {}) - assert r1.sequence < r2.sequence < r3.sequence - - -def test_latest_sequence_reflects_last_record(): - cap = CDCCapture() - cap.insert("customers", "c1", {}) - cap.insert("customers", "c2", {}) - last = cap.update("customers", "c1", {}) - assert cap.latest_sequence == last.sequence - - -# ── checkpoint-based replay ─────────────────────────────────────────────────── - - -def test_records_since_returns_only_records_after_offset(): - cap = CDCCapture() - cap.insert("customers", "c1", {}) - cap.insert("customers", "c2", {}) - checkpoint = cap.latest_sequence - r3 = cap.insert("customers", "c3", {}) - - replayed = cap.records_since(checkpoint) - - assert len(replayed) == 1 - assert replayed[0].sequence == r3.sequence - - -def test_records_since_zero_returns_all_records(): - cap = CDCCapture() - cap.insert("customers", "c1", {}) - cap.insert("customers", "c2", {}) - cap.delete("customers", "c1", {}) - assert len(cap.records_since(0)) == 3 - - -def test_records_since_latest_sequence_returns_empty(): - cap = CDCCapture() - cap.insert("customers", "c1", {}) - assert cap.records_since(cap.latest_sequence) == [] - - -# ── duplicate / replay safety ───────────────────────────────────────────────── - - -def test_same_pk_can_appear_multiple_times_in_log(): - """CDC appends every event — deduplication happens downstream.""" - cap = CDCCapture() - cap.insert("customers", "c1", {"status": "active"}) - cap.update("customers", "c1", {"status": "suspended"}) - cap.update("customers", "c1", {"status": "closed"}) - - records = cap.records_since(0) - pk_records = [r for r in records if r.primary_key == "c1"] - assert len(pk_records) == 3 - - -def test_captured_at_is_utc_datetime(): - cap = CDCCapture() - rec = cap.insert("customers", "c1", {}) - assert isinstance(rec.captured_at, datetime) - assert rec.captured_at.tzinfo is not None diff --git a/submission/sample-candidate/tests/test_data_quality.py b/submission/sample-candidate/tests/test_data_quality.py deleted file mode 100644 index 4dbf7f9..0000000 --- a/submission/sample-candidate/tests/test_data_quality.py +++ /dev/null @@ -1,252 +0,0 @@ -""" -Tests — data quality and warehouse correctness. - -Covers: insert propagation, update correctness, soft-delete, replay safety, -PK uniqueness, not-null checks, business rule assertions, lake completeness, -and lake immutability (no deletes or updates to lake rows). -""" - -from datetime import datetime, timezone - -import pytest - -from pipeline.lake import append_to_lake -from pipeline.warehouse import apply_cdc_records - - -def _ts() -> datetime: - return datetime.now(timezone.utc) - - -# ── insert propagation ──────────────────────────────────────────────────────── - - -def test_insert_appears_in_warehouse(seeded): - conn, _ = seeded - row = conn.execute( - "SELECT name FROM wh_customers WHERE customer_id = 'c1'" - ).fetchone() - assert row is not None - assert row[0] == "Alice" - - -def test_insert_appears_in_lake(seeded): - conn, _ = seeded - count = conn.execute( - "SELECT COUNT(*) FROM lake_cdc_events WHERE table_name = 'customers'" - " AND operation = 'insert'" - ).fetchone()[0] - assert count == 2 - - -def test_all_tables_populated_after_seed(seeded): - conn, _ = seeded - assert conn.execute("SELECT COUNT(*) FROM wh_customers").fetchone()[0] == 2 - assert conn.execute("SELECT COUNT(*) FROM wh_wallets").fetchone()[0] == 2 - assert conn.execute("SELECT COUNT(*) FROM wh_transactions").fetchone()[0] == 2 - - -# ── update correctness ──────────────────────────────────────────────────────── - - -def test_update_overwrites_warehouse_row(seeded): - conn, capture = seeded - ts = _ts() - capture.update( - "customers", - "c1", - { - "customer_id": "c1", - "name": "Alice Smith", - "email": "alice-test-user", - "status": "suspended", - "created_at": ts, - "updated_at": ts, - }, - ) - new_records = capture.records_since(capture.latest_sequence - 1) - apply_cdc_records(conn, new_records) - - row = conn.execute( - "SELECT name, status FROM wh_customers WHERE customer_id = 'c1'" - ).fetchone() - assert row[0] == "Alice Smith" - assert row[1] == "suspended" - - -def test_update_does_not_create_duplicate_warehouse_row(seeded): - conn, capture = seeded - ts = _ts() - capture.update( - "wallets", - "w1", - { - "wallet_id": "w1", - "customer_id": "c1", - "balance": 200.00, - "currency": "USD", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, - ) - apply_cdc_records(conn, capture.records_since(capture.latest_sequence - 1)) - - count = conn.execute( - "SELECT COUNT(*) FROM wh_wallets WHERE wallet_id = 'w1'" - ).fetchone()[0] - assert count == 1 - - -# ── soft delete ─────────────────────────────────────────────────────────────── - - -def test_delete_marks_warehouse_row_as_deleted(seeded): - conn, capture = seeded - capture.delete("customers", "c2", {"customer_id": "c2"}) - apply_cdc_records(conn, capture.records_since(capture.latest_sequence - 1)) - - row = conn.execute( - "SELECT _deleted FROM wh_customers WHERE customer_id = 'c2'" - ).fetchone() - assert row is not None - assert row[0] is True - - -def test_delete_does_not_remove_row_from_warehouse(seeded): - conn, capture = seeded - capture.delete("wallets", "w2", {"wallet_id": "w2"}) - apply_cdc_records(conn, capture.records_since(capture.latest_sequence - 1)) - - count = conn.execute( - "SELECT COUNT(*) FROM wh_wallets WHERE wallet_id = 'w2'" - ).fetchone()[0] - assert count == 1 - - -# ── lake immutability ───────────────────────────────────────────────────────── - - -def test_lake_row_count_only_increases(seeded): - conn, capture = seeded - before = conn.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] - - ts = _ts() - capture.update( - "customers", - "c1", - { - "customer_id": "c1", - "name": "Alice Updated", - "email": "alice-test-user", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, - ) - new_records = capture.records_since(capture.latest_sequence - 1) - append_to_lake(conn, new_records) - - after = conn.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] - assert after > before - - -def test_lake_retains_all_operations_for_same_pk(seeded): - conn, capture = seeded - ts = _ts() - capture.update( - "customers", - "c1", - { - "customer_id": "c1", - "name": "Alice v2", - "email": "alice-test-user", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, - ) - capture.delete("customers", "c1", {"customer_id": "c1"}) - append_to_lake(conn, capture.records_since(capture.latest_sequence - 2)) - - events = conn.execute( - "SELECT operation FROM lake_cdc_events" - " WHERE table_name = 'customers' AND primary_key = 'c1'" - " ORDER BY sequence" - ).fetchall() - ops = [e[0] for e in events] - assert "insert" in ops - assert "update" in ops - assert "delete" in ops - - -# ── PK uniqueness ───────────────────────────────────────────────────────────── - - -def test_warehouse_customers_pk_is_unique(seeded): - conn, _ = seeded - total = conn.execute("SELECT COUNT(*) FROM wh_customers").fetchone()[0] - distinct = conn.execute( - "SELECT COUNT(DISTINCT customer_id) FROM wh_customers" - ).fetchone()[0] - assert total == distinct - - -def test_warehouse_wallets_pk_is_unique(seeded): - conn, _ = seeded - total = conn.execute("SELECT COUNT(*) FROM wh_wallets").fetchone()[0] - distinct = conn.execute( - "SELECT COUNT(DISTINCT wallet_id) FROM wh_wallets" - ).fetchone()[0] - assert total == distinct - - -# ── business rules ──────────────────────────────────────────────────────────── - - -def test_transaction_amounts_are_positive(seeded): - conn, _ = seeded - bad = conn.execute( - "SELECT COUNT(*) FROM wh_transactions WHERE amount <= 0" - ).fetchone()[0] - assert bad == 0 - - -def test_wallet_balances_are_non_negative(seeded): - conn, _ = seeded - bad = conn.execute("SELECT COUNT(*) FROM wh_wallets WHERE balance < 0").fetchone()[ - 0 - ] - assert bad == 0 - - -def test_transaction_directions_are_valid(seeded): - conn, _ = seeded - bad = conn.execute( - "SELECT COUNT(*) FROM wh_transactions" - " WHERE direction NOT IN ('credit', 'debit')" - ).fetchone()[0] - assert bad == 0 - - -# ── replay safety ───────────────────────────────────────────────────────────── - - -def test_replaying_same_records_does_not_create_duplicates(seeded): - conn, capture = seeded - # Replay all records from the beginning - all_records = capture.records_since(0) - apply_cdc_records(conn, all_records) - - count = conn.execute("SELECT COUNT(*) FROM wh_customers").fetchone()[0] - assert count == 2 # Must not double to 4 - - -@pytest.mark.parametrize("offset", [0, 1, 2]) -def test_partial_replay_from_checkpoint_is_idempotent(seeded, offset: int): - conn, capture = seeded - partial = capture.records_since(offset) - apply_cdc_records(conn, partial) - - count = conn.execute("SELECT COUNT(*) FROM wh_customers").fetchone()[0] - assert count == 2 diff --git a/submission/sample-candidate/tests/test_schema_contracts.py b/submission/sample-candidate/tests/test_schema_contracts.py deleted file mode 100644 index 69d30f7..0000000 --- a/submission/sample-candidate/tests/test_schema_contracts.py +++ /dev/null @@ -1,170 +0,0 @@ -""" -Tests — schema contract detection and safe-stop behavior. - -Covers: passing contracts, missing column detection, incompatible schema change -detection (dropped column, added unexpected table), and that ingestion can be -stopped when a contract violation is found. -""" - -import duckdb -import pytest - -from source.models import SCHEMA_CONTRACT, create_source_tables - -# ── helpers ─────────────────────────────────────────────────────────────────── - - -def _actual_columns(conn: duckdb.DuckDBPyConnection, table: str) -> set[str]: - return {row[0] for row in conn.execute(f"DESCRIBE {table}").fetchall()} - - -def _check_contracts(conn: duckdb.DuckDBPyConnection) -> list[str]: - """Inline contract check — mirrors scripts/check_schema_contracts.py.""" - violations: list[str] = [] - for table, expected_cols in SCHEMA_CONTRACT.items(): - try: - actual = _actual_columns(conn, table) - except Exception as exc: - violations.append(f"{table}: {exc}") - continue - for col in expected_cols: - if col not in actual: - violations.append(f"{table}.{col}: column missing") - return violations - - -# ── passing contracts ───────────────────────────────────────────────────────── - - -def test_fresh_source_tables_pass_all_contracts(): - conn = duckdb.connect(":memory:") - create_source_tables(conn) - assert _check_contracts(conn) == [] - - -def test_all_contract_tables_are_present(): - conn = duckdb.connect(":memory:") - create_source_tables(conn) - for table in SCHEMA_CONTRACT: - cols = _actual_columns(conn, table) - assert len(cols) > 0, f"{table} has no columns" - - -# ── missing column detection ────────────────────────────────────────────────── - - -def test_dropped_column_is_detected_as_violation(): - # Create customers without FK-dependent tables so ALTER TABLE is allowed - conn = duckdb.connect(":memory:") - conn.execute(""" - CREATE TABLE customers ( - customer_id VARCHAR PRIMARY KEY, - name VARCHAR NOT NULL, - status VARCHAR NOT NULL, - created_at TIMESTAMP NOT NULL, - updated_at TIMESTAMP NOT NULL - ) - """) - # 'email' was never added — contract check should catch it - violations = _check_contracts(conn) - assert any("email" in v for v in violations) - - -def test_multiple_dropped_columns_all_reported(): - # Create wallets without FK constraint so ALTER TABLE is allowed - conn = duckdb.connect(":memory:") - conn.execute(""" - CREATE TABLE customers (customer_id VARCHAR PRIMARY KEY, - name VARCHAR NOT NULL, email VARCHAR NOT NULL, - status VARCHAR NOT NULL, created_at TIMESTAMP NOT NULL, - updated_at TIMESTAMP NOT NULL) - """) - conn.execute(""" - CREATE TABLE wallets ( - wallet_id VARCHAR PRIMARY KEY, - customer_id VARCHAR NOT NULL, - currency VARCHAR NOT NULL, - status VARCHAR NOT NULL, - created_at TIMESTAMP NOT NULL, - updated_at TIMESTAMP NOT NULL - ) - """) - # 'balance' never added — contract check should report both balance and balance - violations = _check_contracts(conn) - assert any("balance" in v for v in violations) - - -def test_violations_include_table_and_column_name(): - conn = duckdb.connect(":memory:") - create_source_tables(conn) - conn.execute("ALTER TABLE transactions DROP COLUMN amount") - - violations = _check_contracts(conn) - assert any("transactions" in v and "amount" in v for v in violations) - - -# ── contract-based safe stop ────────────────────────────────────────────────── - - -def test_pipeline_stops_when_contract_violated(): - """ - When a contract violation is found, no CDC records should be captured. - This models the stop-the-line behavior required by the assignment. - """ - from pipeline.cdc import CDCCapture - - # Create customers table missing 'email' — simulates a breaking schema change - conn = duckdb.connect(":memory:") - conn.execute(""" - CREATE TABLE customers ( - customer_id VARCHAR PRIMARY KEY, - name VARCHAR NOT NULL, - status VARCHAR NOT NULL, - created_at TIMESTAMP NOT NULL, - updated_at TIMESTAMP NOT NULL - ) - """) - - violations = _check_contracts(conn) - capture = CDCCapture() - - if violations: - # Pipeline aborts — no records captured - pass - else: - capture.insert("customers", "c1", {"customer_id": "c1", "name": "Alice"}) - - assert ( - len(capture.log) == 0 - ), "No records should be captured after a contract violation" - - -# ── schema contract completeness ────────────────────────────────────────────── - - -def test_schema_contract_covers_all_key_tables(): - assert "customers" in SCHEMA_CONTRACT - assert "wallets" in SCHEMA_CONTRACT - assert "transactions" in SCHEMA_CONTRACT - - -def test_schema_contract_includes_primary_key_columns(): - assert "customer_id" in SCHEMA_CONTRACT["customers"] - assert "wallet_id" in SCHEMA_CONTRACT["wallets"] - assert "transaction_id" in SCHEMA_CONTRACT["transactions"] - - -@pytest.mark.parametrize( - "table,col", - [ - ("customers", "status"), - ("wallets", "balance"), - ("wallets", "currency"), - ("transactions", "amount"), - ("transactions", "direction"), - ], -) -def test_business_critical_columns_are_in_contract(table: str, col: str): - assert ( - col in SCHEMA_CONTRACT[table] - ), f"Business-critical column {table}.{col} is not in SCHEMA_CONTRACT"