From c4d932a80bc2b2d906d01d279596db72f0dd4ce5 Mon Sep 17 00:00:00 2001 From: singhsimer09-max Date: Sun, 14 Jun 2026 22:23:03 +0530 Subject: [PATCH] Submit CDC lakehouse reliability assignment --- submission/simerjit/APPROACH.md | 66 +++ submission/simerjit/PR_DESCRIPTION.md | 74 +++ submission/simerjit/README.md | 58 +++ submission/simerjit/catalog.yml | 68 +++ submission/simerjit/job_config.yml | 53 ++ .../notebooks/00_setup_and_source_model.py | 66 +++ .../notebooks/01_generate_cdc_events.py | 178 +++++++ .../notebooks/02_bronze_ingest_delta.py | 69 +++ .../notebooks/03_apply_silver_merges.py | 482 ++++++++++++++++++ .../04_restore_with_delta_time_travel.py | 78 +++ .../simerjit/notebooks/05_quality_checks.py | 102 ++++ .../notebooks/06_cdc_correctness_tests.py | 128 +++++ .../simerjit/notebooks/08_catalog_metadata.py | 124 +++++ .../notebooks/09_final_summary_or_pr_notes.py | 193 +++++++ submission/simerjit/schema_contracts.yml | 64 +++ submission/simerjit/sql/source_schema.sql | 77 +++ 16 files changed, 1880 insertions(+) create mode 100644 submission/simerjit/APPROACH.md create mode 100644 submission/simerjit/PR_DESCRIPTION.md create mode 100644 submission/simerjit/README.md create mode 100644 submission/simerjit/catalog.yml create mode 100644 submission/simerjit/job_config.yml create mode 100644 submission/simerjit/notebooks/00_setup_and_source_model.py create mode 100644 submission/simerjit/notebooks/01_generate_cdc_events.py create mode 100644 submission/simerjit/notebooks/02_bronze_ingest_delta.py create mode 100644 submission/simerjit/notebooks/03_apply_silver_merges.py create mode 100644 submission/simerjit/notebooks/04_restore_with_delta_time_travel.py create mode 100644 submission/simerjit/notebooks/05_quality_checks.py create mode 100644 submission/simerjit/notebooks/06_cdc_correctness_tests.py create mode 100644 submission/simerjit/notebooks/08_catalog_metadata.py create mode 100644 submission/simerjit/notebooks/09_final_summary_or_pr_notes.py create mode 100644 submission/simerjit/schema_contracts.yml create mode 100644 submission/simerjit/sql/source_schema.sql diff --git a/submission/simerjit/APPROACH.md b/submission/simerjit/APPROACH.md new file mode 100644 index 0000000..5d49f9f --- /dev/null +++ b/submission/simerjit/APPROACH.md @@ -0,0 +1,66 @@ +# Approach + +## Understanding + +The assignment asks for a CDC lakehouse reliability pipeline. The solution should model a realistic transactional source, capture inserts/updates/deletes, preserve raw change history in a lake layer, publish current-state warehouse tables, stop on unsafe schema changes, support restore/replay, expose catalog metadata, and include validations or tests. + +## Proposed Solution + +I implemented the assignment with Databricks, PySpark, and Delta Lake using a wallet and payments domain. + +The source model includes: + +- `customers` +- `wallet_accounts` +- `merchants` +- `payment_transactions` +- `payment_attempts` + +CDC events are simulated as JSON files in a Databricks Volume. Each event contains: + +- `source_table` +- `cdc_operation` +- `cdc_sequence` +- `source_commit_ts` +- `schema_version` +- `payload_json` + +## Lakehouse Design + +The Bronze lake table is: + +- `robustrade_dev.bronze.cdc_events` + +It is append-only and keeps every CDC event. + +The Silver warehouse tables are: + +- `robustrade_dev.silver.customers_current` +- `robustrade_dev.silver.wallet_accounts_current` +- `robustrade_dev.silver.merchants_current` +- `robustrade_dev.silver.payment_transactions_current` +- `robustrade_dev.silver.payment_attempts_current` + +Silver tables are maintained with Delta `MERGE` and keep the latest non-deleted row per primary key. + +## Reliability Plan + +The implementation includes: + +- CDC inserts, updates, and deletes. +- Auto Loader ingestion into Bronze Delta. +- Delta `MERGE` for Silver current-state tables. +- Schema contract checks before Silver writes. +- CDC correctness checks for update/delete/idempotency behavior. +- Validation parity checks for business rules. +- Catalog metadata. +- Delta time travel restore demo. + +## Recovery Strategy + +Bronze Delta is append-only and replayable. Silver tables can be rebuilt from Bronze, and accidental Silver changes can be recovered with Delta Lake history using `DESCRIBE HISTORY` and `RESTORE TABLE`. + +## Responsible AI Usage + +AI assistance was used to help structure the notebooks, debug Databricks errors, and refine explanations. I personally ran the notebooks in Databricks, reviewed the outputs, corrected errors, and validated the final pipeline behavior. + diff --git a/submission/simerjit/PR_DESCRIPTION.md b/submission/simerjit/PR_DESCRIPTION.md new file mode 100644 index 0000000..e6bc405 --- /dev/null +++ b/submission/simerjit/PR_DESCRIPTION.md @@ -0,0 +1,74 @@ +# PR: CDC Lakehouse Reliability Assignment + +## Summary + +This PR implements a CDC lakehouse reliability pipeline using Databricks, PySpark, and Delta Lake. It simulates a wallet/payments source system, writes CDC events to a Databricks Volume, ingests them into an append-only Bronze Delta table, applies schema-safe Delta merges into Silver current-state tables, runs validation checks, documents catalog metadata, and demonstrates Delta time travel restore. + +## Source Schema Design + +The source model has five tables: + +- `customers` +- `wallet_accounts` +- `merchants` +- `payment_transactions` +- `payment_attempts` + +`customers`, `wallet_accounts`, `merchants`, and `payment_transactions` are strong entities. `payment_attempts` is a weak entity because each attempt depends on a parent payment transaction. + +The source design includes primary keys, foreign keys, indexes, timestamps, nullable optional fields, currency fields, status domains, non-negative wallet balance checks, and positive payment amount checks. + +## CDC Strategy + +CDC is simulated as JSON files in a Databricks Volume. Each event includes `source_table`, `cdc_operation`, `cdc_sequence`, `source_commit_ts`, `schema_version`, and `payload_json`. + +The sample stream contains inserts, an update, and a delete. In production, this simulated stream would be replaced by database logs, Debezium, Kafka, Fivetran, or another CDC connector. + +## Lake And Warehouse Modeling + +The lake layer is `robustrade_dev.bronze.cdc_events`. It is append-only and retains full CDC history. + +The warehouse layer contains Silver current-state tables. These tables are built with Delta `MERGE`, ordered by `source_commit_ts` and `cdc_sequence`, and retain only the latest non-deleted row per key. + +## Schema Change Safety + +Before writing to Silver, each CDC payload is parsed using an explicit schema and checked against a contract. Missing required fields, data type changes, or required fields becoming nullable stop the pipeline before downstream mutation. + +## Validation Parity + +Warehouse checks mirror source business rules: + +- primary key uniqueness +- referential integrity +- non-negative wallet balances +- positive payment amounts +- accepted transaction statuses +- duplicate CDC sequence detection + +## Catalog Exposure + +Catalog metadata is provided in `catalog.yml` and the `08_catalog_metadata` notebook. It documents dataset name, layer, platform, owner, description, consumers, primary key, and update cadence. + +## Historical Recovery + +Silver tables can be rebuilt from Bronze. The restore notebook also demonstrates Delta Lake time travel with `DESCRIBE HISTORY` and `RESTORE TABLE`. + +## Responsible AI Usage + +AI assistance was used to help structure the notebooks, debug Databricks errors, and refine explanations. I personally ran the notebooks in Databricks, reviewed the outputs, corrected errors, and validated the final pipeline behavior. + +## Testing And Validation + +Completed validations: + +- 7 CDC events generated. +- 7 events loaded into Bronze Delta. +- Schema safe-stop test failed correctly for an unsafe type change. +- Silver merges created current-state tables. +- `payment_transactions_current` showed transaction `t_001` as `SETTLED`. +- CDC correctness checks passed. +- Rerunning the Silver merge did not create duplicate rows. +- Quality checks passed. +- Catalog metadata coverage passed. +- Delta restore returned a corrupted transaction status back to `SETTLED`. + diff --git a/submission/simerjit/README.md b/submission/simerjit/README.md new file mode 100644 index 0000000..ea5c2d5 --- /dev/null +++ b/submission/simerjit/README.md @@ -0,0 +1,58 @@ +# CDC Lakehouse Reliability Assignment + +Databricks, PySpark, and Delta Lake solution for the Robustrade data engineering assignment. + +## Stack + +- Databricks +- PySpark +- Delta Lake +- Unity Catalog +- Databricks Volumes +- YAML metadata/configuration + +## Run Order + +Run the notebooks in this order: + +1. `00_setup_and_source_model` +2. `01_generate_cdc_events` +3. `02_bronze_ingest_delta` +4. `03_apply_silver_merges` +5. `06_cdc_correctness_tests` +6. `05_quality_checks` +7. `08_catalog_metadata` +8. `04_restore_with_delta_time_travel` +9. `09_final_summary_or_pr_notes` + +## What The Project Demonstrates + +- Source schema design for a wallet/payments domain. +- Simulated CDC event contract. +- Insert, update, and delete CDC events. +- Bronze Delta append-only history. +- Silver Delta current-state tables. +- Schema-safe stopping before incompatible writes. +- CDC correctness and idempotency checks. +- Validation parity for warehouse tables. +- Catalog metadata. +- Delta time travel restore. + +## Important Tables + +- `robustrade_dev.bronze.cdc_events` +- `robustrade_dev.silver.customers_current` +- `robustrade_dev.silver.wallet_accounts_current` +- `robustrade_dev.silver.merchants_current` +- `robustrade_dev.silver.payment_transactions_current` +- `robustrade_dev.silver.payment_attempts_current` + +## Supporting Files + +- `APPROACH.md`: initial understanding and implementation approach. +- `PR_DESCRIPTION.md`: PR-style summary mapped to assignment expectations. +- `catalog.yml`: dataset catalog metadata. +- `schema_contracts.yml`: expected source payload contracts. +- `job_config.yml`: notebook task order and dependencies. +- `sql/source_schema.sql`: source model DDL. + diff --git a/submission/simerjit/catalog.yml b/submission/simerjit/catalog.yml new file mode 100644 index 0000000..6a70536 --- /dev/null +++ b/submission/simerjit/catalog.yml @@ -0,0 +1,68 @@ +version: 1 +owner: data-platform +domain: wallet-payments + +datasets: + - name: robustrade_dev.bronze.cdc_events + layer: bronze + platform: Databricks Delta Lake + description: Append-only CDC event log retaining all source inserts, updates, and deletes. + primary_key: source_table, cdc_sequence + update_cadence: per pipeline run + consumers: + - data engineering + - incident response + - backfill jobs + + - name: robustrade_dev.silver.customers_current + layer: silver + platform: Databricks Delta Lake + description: Latest non-deleted customer state. + primary_key: customer_id + update_cadence: per pipeline run + consumers: + - analytics + - risk + - customer operations + + - name: robustrade_dev.silver.wallet_accounts_current + layer: silver + platform: Databricks Delta Lake + description: Latest wallet balance and status. + primary_key: wallet_id + update_cadence: per pipeline run + consumers: + - analytics + - finance operations + + - name: robustrade_dev.silver.merchants_current + layer: silver + platform: Databricks Delta Lake + description: Latest merchant reference state. + primary_key: merchant_id + update_cadence: per pipeline run + consumers: + - analytics + - risk + + - name: robustrade_dev.silver.payment_transactions_current + layer: silver + platform: Databricks Delta Lake + description: Latest payment transaction state. + primary_key: transaction_id + update_cadence: per pipeline run + consumers: + - analytics + - finance + - reconciliation + + - name: robustrade_dev.silver.payment_attempts_current + layer: silver + platform: Databricks Delta Lake + description: Latest payment processor attempt state. + primary_key: attempt_id + update_cadence: per pipeline run + consumers: + - payments operations + - incident response + diff --git a/submission/simerjit/job_config.yml b/submission/simerjit/job_config.yml new file mode 100644 index 0000000..7f39893 --- /dev/null +++ b/submission/simerjit/job_config.yml @@ -0,0 +1,53 @@ +job_name: cdc_lakehouse_reliability +catalog: robustrade_dev +volume_path: /Volumes/robustrade_dev/audit/landing + +tasks: + - task_key: setup + notebook: 00_setup_and_source_model + description: Create catalog and schemas, document source model. + + - task_key: generate_cdc_events + notebook: 01_generate_cdc_events + description: Generate simulated CDC JSON events. + + - task_key: bronze_ingest + notebook: 02_bronze_ingest_delta + depends_on: + - generate_cdc_events + description: Ingest CDC JSON into Bronze Delta table. + + - task_key: silver_merge + notebook: 03_apply_silver_merges + depends_on: + - bronze_ingest + description: Validate schemas and merge CDC into Silver current-state tables. + + - task_key: cdc_correctness_tests + notebook: 06_cdc_correctness_tests + depends_on: + - silver_merge + description: Verify CDC update, delete, idempotency, and duplicate handling. + + - task_key: quality_checks + notebook: 05_quality_checks + depends_on: + - silver_merge + description: Run warehouse validation parity checks. + + - task_key: catalog_metadata + notebook: 08_catalog_metadata + depends_on: + - silver_merge + description: Document lake and warehouse datasets. + + - task_key: restore_demo + notebook: 04_restore_with_delta_time_travel + depends_on: + - silver_merge + description: Demonstrate Delta history and restore. + + - task_key: final_summary + notebook: 09_final_summary_or_pr_notes + description: Summarize assignment solution and validation results. + diff --git a/submission/simerjit/notebooks/00_setup_and_source_model.py b/submission/simerjit/notebooks/00_setup_and_source_model.py new file mode 100644 index 0000000..16152b1 --- /dev/null +++ b/submission/simerjit/notebooks/00_setup_and_source_model.py @@ -0,0 +1,66 @@ +# Databricks notebook source +# MAGIC %sql +# MAGIC CREATE CATALOG IF NOT EXISTS robustrade_dev; +# MAGIC USE CATALOG robustrade_dev; +# MAGIC +# MAGIC CREATE SCHEMA IF NOT EXISTS bronze; +# MAGIC CREATE SCHEMA IF NOT EXISTS silver; +# MAGIC CREATE SCHEMA IF NOT EXISTS audit; + +# COMMAND ---------- + +# MAGIC %md +# MAGIC Catalog Check + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC SHOW CATALOGS; + +# COMMAND ---------- + +# MAGIC %md +# MAGIC Schema Check + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC SHOW SCHEMAS IN robustrade_dev; + +# COMMAND ---------- + +# MAGIC %md +# MAGIC # Source Model +# MAGIC +# MAGIC This project models a wallet and payments source system with five tables: +# MAGIC +# MAGIC - customers +# MAGIC - wallet_accounts +# MAGIC - merchants +# MAGIC - payment_transactions +# MAGIC - payment_attempts +# MAGIC +# MAGIC Strong entities: +# MAGIC - customers +# MAGIC - wallet_accounts +# MAGIC - merchants +# MAGIC - payment_transactions +# MAGIC +# MAGIC Weak entity: +# MAGIC - payment_attempts +# MAGIC +# MAGIC payment_attempts is weak because each attempt depends on a parent payment transaction. +# MAGIC +# MAGIC Relationships: +# MAGIC - customers -> wallet_accounts +# MAGIC - customers -> payment_transactions +# MAGIC - wallet_accounts -> payment_transactions +# MAGIC - merchants -> payment_transactions +# MAGIC - payment_transactions -> payment_attempts +# MAGIC +# MAGIC Important invariants: +# MAGIC - wallet balances must be non-negative +# MAGIC - payment amounts must be positive +# MAGIC - statuses must be from allowed values +# MAGIC - payments must reference valid customers, wallets, and merchants +# MAGIC - attempts must reference valid payment transactions \ No newline at end of file diff --git a/submission/simerjit/notebooks/01_generate_cdc_events.py b/submission/simerjit/notebooks/01_generate_cdc_events.py new file mode 100644 index 0000000..2a6c8f0 --- /dev/null +++ b/submission/simerjit/notebooks/01_generate_cdc_events.py @@ -0,0 +1,178 @@ +# Databricks notebook source +dbutils.widgets.text("catalog", "robustrade_dev") +dbutils.widgets.text("volume_path", "/Volumes/robustrade_dev/audit/landing") + +catalog = dbutils.widgets.get("catalog") +volume_path = dbutils.widgets.get("volume_path").rstrip("/") + +spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.bronze") +spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.silver") +spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.audit") + +# COMMAND ---------- + +spark.sql(f"CREATE VOLUME IF NOT EXISTS {catalog}.audit.landing") + +# COMMAND ---------- + +import json + +from pyspark.sql import Row +from pyspark.sql import functions as F + +events = [ + Row( + source_table="customers", + cdc_operation="INSERT", + cdc_sequence=1, + source_commit_ts="2026-01-01T10:00:00Z", + schema_version=1, + payload_json=json.dumps( + { + "customer_id": "c_001", + "email": "ada@example.com", + "full_name": "Ada Lovelace", + "phone_number": "+15550100", + "kyc_level": "BASIC", + "customer_status": "ACTIVE", + "updated_at": "2026-01-01T10:00:00Z", + } + ), + ), + Row( + source_table="wallet_accounts", + cdc_operation="INSERT", + cdc_sequence=2, + source_commit_ts="2026-01-01T10:02:00Z", + schema_version=1, + payload_json=json.dumps( + { + "wallet_id": "w_001", + "customer_id": "c_001", + "currency": "USD", + "balance_minor": 500000, + "wallet_status": "OPEN", + "updated_at": "2026-01-01T10:02:00Z", + } + ), + ), + Row( + source_table="merchants", + cdc_operation="INSERT", + cdc_sequence=3, + source_commit_ts="2026-01-01T10:03:00Z", + schema_version=1, + payload_json=json.dumps( + { + "merchant_id": "m_001", + "merchant_name": "Analytical Engines Ltd", + "merchant_category": "SOFTWARE", + "merchant_status": "ACTIVE", + "created_at": "2026-01-01T10:03:00Z", + "updated_at": "2026-01-01T10:03:00Z", + } + ), + ), + Row( + source_table="payment_transactions", + cdc_operation="INSERT", + cdc_sequence=4, + source_commit_ts="2026-01-01T10:05:00Z", + schema_version=1, + payload_json=json.dumps( + { + "transaction_id": "t_001", + "wallet_id": "w_001", + "customer_id": "c_001", + "merchant_id": "m_001", + "amount_minor": 1999, + "currency": "USD", + "transaction_status": "AUTHORIZED", + "failure_code": None, + "event_time": "2026-01-01T10:04:59Z", + "updated_at": "2026-01-01T10:05:00Z", + } + ), + ), + Row( + source_table="payment_transactions", + cdc_operation="UPDATE", + cdc_sequence=5, + source_commit_ts="2026-01-01T10:06:00Z", + schema_version=1, + payload_json=json.dumps( + { + "transaction_id": "t_001", + "wallet_id": "w_001", + "customer_id": "c_001", + "merchant_id": "m_001", + "amount_minor": 1999, + "currency": "USD", + "transaction_status": "SETTLED", + "failure_code": None, + "event_time": "2026-01-01T10:04:59Z", + "updated_at": "2026-01-01T10:06:00Z", + } + ), + ), + Row( + source_table="payment_attempts", + cdc_operation="INSERT", + cdc_sequence=6, + source_commit_ts="2026-01-01T10:06:30Z", + schema_version=1, + payload_json=json.dumps( + { + "attempt_id": "a_001", + "transaction_id": "t_001", + "processor_name": "stripe", + "attempt_status": "APPROVED", + "failure_code": None, + "attempted_at": "2026-01-01T10:05:01Z", + "updated_at": "2026-01-01T10:06:30Z", + } + ), + ), + Row( + source_table="customers", + cdc_operation="DELETE", + cdc_sequence=7, + source_commit_ts="2026-01-01T10:08:00Z", + schema_version=1, + payload_json=json.dumps( + { + "customer_id": "c_deleted", + "email": "removed@example.com", + "full_name": "Removed Customer", + "phone_number": None, + "kyc_level": "BASIC", + "customer_status": "CLOSED", + "updated_at": "2026-01-01T10:08:00Z", + } + ), + ), +] + +df = spark.createDataFrame(events).withColumn("event_date", F.to_date("source_commit_ts")) + +landing_path = f"{volume_path}/cdc_events" + +( + df.coalesce(1) + .write.mode("overwrite") + .partitionBy("event_date") + .json(landing_path) +) + +print(f"Wrote CDC JSON events to {landing_path}") + +# COMMAND ---------- + +display(dbutils.fs.ls(f"{volume_path}/cdc_events")) + +# COMMAND ---------- + +display( + spark.read.json(f"{volume_path}/cdc_events") + .orderBy("cdc_sequence") +) \ No newline at end of file diff --git a/submission/simerjit/notebooks/02_bronze_ingest_delta.py b/submission/simerjit/notebooks/02_bronze_ingest_delta.py new file mode 100644 index 0000000..93fb6b7 --- /dev/null +++ b/submission/simerjit/notebooks/02_bronze_ingest_delta.py @@ -0,0 +1,69 @@ +# Databricks notebook source +dbutils.widgets.text("catalog", "robustrade_dev") +dbutils.widgets.text("volume_path", "/Volumes/robustrade_dev/audit/landing") + +catalog = dbutils.widgets.get("catalog") +volume_path = dbutils.widgets.get("volume_path").rstrip("/") + +bronze_table = f"{catalog}.bronze.cdc_events" +source_path = f"{volume_path}/cdc_events" +checkpoint_path = f"{volume_path}/_checkpoints/bronze_cdc_events" +schema_location = f"{volume_path}/_schemas/bronze_cdc_events" + +spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.bronze") + +# COMMAND ---------- + +from pyspark.sql import functions as F + +raw = ( + spark.readStream.format("cloudFiles") + .option("cloudFiles.format", "json") + .option("cloudFiles.inferColumnTypes", "true") + .option("cloudFiles.schemaLocation", schema_location) + .load(source_path) +) + +bronze = ( + raw.withColumn("source_commit_ts", F.to_timestamp("source_commit_ts")) + .withColumn("ingested_at", F.current_timestamp()) + .withColumn("ingest_batch_id", F.expr("uuid()")) + .withColumn("_source_file", F.col("_metadata.file_path")) +) + +query = ( + bronze.writeStream.format("delta") + .option("checkpointLocation", checkpoint_path) + .option("mergeSchema", "false") + .trigger(availableNow=True) + .toTable(bronze_table) +) + +query.awaitTermination() + +# COMMAND ---------- + +spark.sql( + f""" + ALTER TABLE {bronze_table} SET TBLPROPERTIES ( + 'delta.appendOnly' = 'true', + 'quality' = 'bronze', + 'description' = 'Immutable CDC event log for wallet and payment sources' + ) + """ +) + +# COMMAND ---------- + +display(spark.table(bronze_table).orderBy("cdc_sequence")) + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC SELECT COUNT(*) AS bronze_row_count +# MAGIC FROM robustrade_dev.bronze.cdc_events; + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC DESCRIBE DETAIL robustrade_dev.bronze.cdc_events; \ No newline at end of file diff --git a/submission/simerjit/notebooks/03_apply_silver_merges.py b/submission/simerjit/notebooks/03_apply_silver_merges.py new file mode 100644 index 0000000..1787486 --- /dev/null +++ b/submission/simerjit/notebooks/03_apply_silver_merges.py @@ -0,0 +1,482 @@ +# Databricks notebook source +dbutils.widgets.text("catalog", "robustrade_dev") +dbutils.widgets.text("volume_path", "/Volumes/robustrade_dev/audit/landing") + +catalog = dbutils.widgets.get("catalog") +volume_path = dbutils.widgets.get("volume_path") + +from pyspark.sql import functions as F +from pyspark.sql import types as T + +spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.silver") + +bronze_table = f"{catalog}.bronze.cdc_events" +bronze = spark.table(bronze_table) + +# COMMAND ---------- + +from dataclasses import dataclass +from typing import Iterable, List, Mapping + + +@dataclass(frozen=True) +class ColumnSpec: + name: str + data_type: str + nullable: bool + + +@dataclass(frozen=True) +class ContractViolation: + table: str + message: str + + +def normalize_type(data_type: str) -> str: + aliases = { + "bigint": "long", + "integer": "int", + "varchar": "string", + "datetime": "timestamp", + } + value = data_type.lower().strip() + return aliases.get(value, value) + + +def validate_schema_contract( + table: str, + required_columns: Mapping[str, str], + nullable_columns: Mapping[str, str], + actual_columns: Iterable[ColumnSpec], +) -> List[ContractViolation]: + actual = {column.name: column for column in actual_columns} + violations = [] + + for name, expected_type in required_columns.items(): + if name not in actual: + violations.append(ContractViolation(table, f"required column missing: {name}")) + continue + + column = actual[name] + if normalize_type(column.data_type) != normalize_type(expected_type): + violations.append( + ContractViolation( + table, + f"type mismatch for {name}: expected {expected_type}, got {column.data_type}", + ) + ) + if column.nullable: + violations.append(ContractViolation(table, f"required column became nullable: {name}")) + + for name, expected_type in nullable_columns.items(): + if name not in actual: + violations.append(ContractViolation(table, f"nullable column missing: {name}")) + continue + + column = actual[name] + if normalize_type(column.data_type) != normalize_type(expected_type): + violations.append( + ContractViolation( + table, + f"type mismatch for {name}: expected {expected_type}, got {column.data_type}", + ) + ) + + return violations + + +def stop_on_violations(violations: Iterable[ContractViolation]) -> None: + messages = [f"{violation.table}: {violation.message}" for violation in violations] + if messages: + raise ValueError("Unsafe source schema change detected:\n" + "\n".join(messages)) + +# COMMAND ---------- + +payload_schemas = { + "customers": T.StructType( + [ + T.StructField("customer_id", T.StringType(), False), + T.StructField("email", T.StringType(), False), + T.StructField("full_name", T.StringType(), False), + T.StructField("phone_number", T.StringType(), True), + T.StructField("kyc_level", T.StringType(), True), + T.StructField("customer_status", T.StringType(), False), + T.StructField("updated_at", T.TimestampType(), False), + ] + ), + "wallet_accounts": T.StructType( + [ + T.StructField("wallet_id", T.StringType(), False), + T.StructField("customer_id", T.StringType(), False), + T.StructField("currency", T.StringType(), False), + T.StructField("balance_minor", T.LongType(), False), + T.StructField("wallet_status", T.StringType(), False), + T.StructField("updated_at", T.TimestampType(), False), + ] + ), + "merchants": T.StructType( + [ + T.StructField("merchant_id", T.StringType(), False), + T.StructField("merchant_name", T.StringType(), False), + T.StructField("merchant_category", T.StringType(), False), + T.StructField("merchant_status", T.StringType(), False), + T.StructField("created_at", T.TimestampType(), False), + T.StructField("updated_at", T.TimestampType(), False), + ] + ), + "payment_transactions": T.StructType( + [ + T.StructField("transaction_id", T.StringType(), False), + T.StructField("wallet_id", T.StringType(), False), + T.StructField("customer_id", T.StringType(), False), + T.StructField("merchant_id", T.StringType(), True), + T.StructField("amount_minor", T.LongType(), False), + T.StructField("currency", T.StringType(), False), + T.StructField("transaction_status", T.StringType(), False), + T.StructField("failure_code", T.StringType(), True), + T.StructField("event_time", T.TimestampType(), False), + T.StructField("updated_at", T.TimestampType(), False), + ] + ), + "payment_attempts": T.StructType( + [ + T.StructField("attempt_id", T.StringType(), False), + T.StructField("transaction_id", T.StringType(), False), + T.StructField("processor_name", T.StringType(), False), + T.StructField("attempt_status", T.StringType(), False), + T.StructField("failure_code", T.StringType(), True), + T.StructField("attempted_at", T.TimestampType(), False), + T.StructField("updated_at", T.TimestampType(), False), + ] + ), +} + +# COMMAND ---------- + +contracts = { + "customers": { + "required": { + "customer_id": "string", + "email": "string", + "full_name": "string", + "customer_status": "string", + "updated_at": "timestamp", + }, + "nullable": {"phone_number": "string", "kyc_level": "string"}, + }, + "wallet_accounts": { + "required": { + "wallet_id": "string", + "customer_id": "string", + "currency": "string", + "balance_minor": "long", + "wallet_status": "string", + "updated_at": "timestamp", + }, + "nullable": {}, + }, + "merchants": { + "required": { + "merchant_id": "string", + "merchant_name": "string", + "merchant_category": "string", + "merchant_status": "string", + "created_at": "timestamp", + "updated_at": "timestamp", + }, + "nullable": {}, + }, + "payment_transactions": { + "required": { + "transaction_id": "string", + "wallet_id": "string", + "customer_id": "string", + "amount_minor": "long", + "currency": "string", + "transaction_status": "string", + "event_time": "timestamp", + "updated_at": "timestamp", + }, + "nullable": {"merchant_id": "string", "failure_code": "string"}, + }, + "payment_attempts": { + "required": { + "attempt_id": "string", + "transaction_id": "string", + "processor_name": "string", + "attempt_status": "string", + "attempted_at": "timestamp", + "updated_at": "timestamp", + }, + "nullable": {"failure_code": "string"}, + }, +} + +# COMMAND ---------- + +for table_name, schema in payload_schemas.items(): + actual_columns = [ + ColumnSpec(field.name, field.dataType.simpleString(), field.nullable) + for field in schema.fields + ] + + violations = validate_schema_contract( + table_name, + contracts[table_name]["required"], + contracts[table_name]["nullable"], + actual_columns, + ) + + stop_on_violations(violations) + +print("All schema contracts passed.") + +# COMMAND ---------- + +# bad_columns = [ +# ColumnSpec("amount_minor", "string", False) +# ] + +# bad_violations = validate_schema_contract( +# "payment_transactions", +# {"amount_minor": "long"}, +# {}, +# bad_columns, +# ) + +# stop_on_violations(bad_violations) + +# COMMAND ---------- + +from dataclasses import dataclass +from typing import List + + +@dataclass(frozen=True) +class TablePlan: + source_table: str + primary_key: str + payload_columns: List[str] + + +plans = [ + TablePlan( + source_table="customers", + primary_key="customer_id", + payload_columns=[ + "customer_id", + "email", + "full_name", + "phone_number", + "kyc_level", + "customer_status", + "updated_at", + ], + ), + TablePlan( + source_table="wallet_accounts", + primary_key="wallet_id", + payload_columns=[ + "wallet_id", + "customer_id", + "currency", + "balance_minor", + "wallet_status", + "updated_at", + ], + ), + TablePlan( + source_table="merchants", + primary_key="merchant_id", + payload_columns=[ + "merchant_id", + "merchant_name", + "merchant_category", + "merchant_status", + "created_at", + "updated_at", + ], + ), + TablePlan( + source_table="payment_transactions", + primary_key="transaction_id", + payload_columns=[ + "transaction_id", + "wallet_id", + "customer_id", + "merchant_id", + "amount_minor", + "currency", + "transaction_status", + "failure_code", + "event_time", + "updated_at", + ], + ), + TablePlan( + source_table="payment_attempts", + primary_key="attempt_id", + payload_columns=[ + "attempt_id", + "transaction_id", + "processor_name", + "attempt_status", + "failure_code", + "attempted_at", + "updated_at", + ], + ), +] + +# COMMAND ---------- + +from delta.tables import DeltaTable +from pyspark.sql.window import Window + + +def merge_condition(target_alias: str, source_alias: str, primary_key: str) -> str: + return f"{target_alias}.{primary_key} = {source_alias}.{primary_key}" + + +def build_set_map(columns, source_alias: str = "s"): + values = {column: f"{source_alias}.{column}" for column in columns} + values["last_cdc_operation"] = f"{source_alias}.cdc_operation" + values["last_cdc_sequence"] = f"{source_alias}.cdc_sequence" + values["last_source_commit_ts"] = f"{source_alias}.source_commit_ts" + values["silver_updated_at"] = "current_timestamp()" + return values + + +def build_insert_map(columns, source_alias: str = "s"): + return build_set_map(columns, source_alias) + + +def dedupe_latest_events(events_df, source_table: str, primary_key: str): + filtered = events_df.where(F.col("source_table") == source_table) + + flattened = filtered.select( + "source_table", + "cdc_operation", + "cdc_sequence", + "source_commit_ts", + "ingested_at", + "payload.*", + ) + + window = Window.partitionBy(primary_key).orderBy( + F.col("source_commit_ts").desc(), + F.col("cdc_sequence").desc(), + ) + + return ( + flattened.withColumn("rn", F.row_number().over(window)) + .where(F.col("rn") == 1) + .drop("rn") + .withColumn("is_deleted", F.col("cdc_operation") == F.lit("DELETE")) + ) + + +def apply_cdc_merge(spark, events_df, target_table: str, plan: TablePlan): + latest = dedupe_latest_events(events_df, plan.source_table, plan.primary_key) + delta_table = DeltaTable.forName(spark, target_table) + + ( + delta_table.alias("t") + .merge(latest.alias("s"), merge_condition("t", "s", plan.primary_key)) + .whenMatchedDelete(condition="s.is_deleted = true") + .whenMatchedUpdate( + condition="s.is_deleted = false AND s.cdc_sequence >= t.last_cdc_sequence", + set=build_set_map(plan.payload_columns), + ) + .whenNotMatchedInsert( + condition="s.is_deleted = false", + values=build_insert_map(plan.payload_columns), + ) + .execute() + ) + +# COMMAND ---------- + +for plan in plans: + schema = payload_schemas[plan.source_table] + + parsed_events = bronze.withColumn( + "payload", + F.from_json("payload_json", schema), + ) + + empty_current = ( + parsed_events.where(F.col("source_table") == plan.source_table) + .select("payload.*") + .limit(0) + .withColumn("last_cdc_operation", F.lit(None).cast("string")) + .withColumn("last_cdc_sequence", F.lit(None).cast("long")) + .withColumn("last_source_commit_ts", F.lit(None).cast("timestamp")) + .withColumn("silver_updated_at", F.current_timestamp()) + ) + + target_table = f"{catalog}.silver.{plan.source_table}_current" + + if not spark.catalog.tableExists(target_table): + ( + empty_current.write + .format("delta") + .mode("overwrite") + .saveAsTable(target_table) + ) + + spark.sql( + f""" + ALTER TABLE {target_table} SET TBLPROPERTIES ( + 'quality' = 'silver', + 'description' = 'Latest current-state table derived from CDC events' + ) + """ + ) + + apply_cdc_merge(spark, parsed_events, target_table, plan) + + print(f"Applied CDC merge into {target_table}") + +# COMMAND ---------- + +display(spark.table(f"{catalog}.silver.customers_current")) + +# COMMAND ---------- + +display(spark.table(f"{catalog}.silver.wallet_accounts_current")) + +# COMMAND ---------- + +display(spark.table(f"{catalog}.silver.merchants_current")) + +# COMMAND ---------- + +display(spark.table(f"{catalog}.silver.payment_transactions_current")) + +# COMMAND ---------- + +display(spark.table(f"{catalog}.silver.payment_attempts_current")) + +# COMMAND ---------- + +display( + spark.sql( + f""" + SELECT transaction_id, transaction_status, last_cdc_sequence + FROM {catalog}.silver.payment_transactions_current + """ + ) +) + +# COMMAND ---------- + +for table_name in [ + "customers_current", + "wallet_accounts_current", + "merchants_current", + "payment_transactions_current", + "payment_attempts_current", +]: + count = spark.table(f"{catalog}.silver.{table_name}").count() + print(table_name, count) \ No newline at end of file diff --git a/submission/simerjit/notebooks/04_restore_with_delta_time_travel.py b/submission/simerjit/notebooks/04_restore_with_delta_time_travel.py new file mode 100644 index 0000000..a893ba3 --- /dev/null +++ b/submission/simerjit/notebooks/04_restore_with_delta_time_travel.py @@ -0,0 +1,78 @@ +# Databricks notebook source +dbutils.widgets.text("catalog", "robustrade_dev") +dbutils.widgets.text("table_name", "payment_transactions_current") +dbutils.widgets.text("restore_version", "") + +catalog = dbutils.widgets.get("catalog") +table_name = dbutils.widgets.get("table_name") +restore_version = dbutils.widgets.get("restore_version") + +full_table = f"{catalog}.silver.{table_name}" + +# COMMAND ---------- + +display(spark.table(full_table)) + +# COMMAND ---------- + +history = spark.sql(f"DESCRIBE HISTORY {full_table}") +display(history) + +# COMMAND ---------- + +spark.sql( + f""" + UPDATE {full_table} + SET transaction_status = 'CORRUPTED_FOR_RESTORE_DEMO' + WHERE transaction_id = 't_001' + """ +) + +# COMMAND ---------- + +display( + spark.sql( + f""" + SELECT transaction_id, transaction_status + FROM {full_table} + WHERE transaction_id = 't_001' + """ + ) +) + +# COMMAND ---------- + +history_after_bad_update = spark.sql(f"DESCRIBE HISTORY {full_table}") +display(history_after_bad_update) + +# COMMAND ---------- + +history_rows = spark.sql(f"DESCRIBE HISTORY {full_table}").select("version").orderBy("version").collect() + +versions = [row["version"] for row in history_rows] + +restore_to_version = versions[-2] + +spark.sql(f"RESTORE TABLE {full_table} TO VERSION AS OF {restore_to_version}") + +print(f"Restored {full_table} to version {restore_to_version}") + +# COMMAND ---------- + +display( + spark.sql( + f""" + SELECT transaction_id, transaction_status + FROM {full_table} + WHERE transaction_id = 't_001' + """ + ) +) + +# COMMAND ---------- + +if restore_version: + spark.sql(f"RESTORE TABLE {full_table} TO VERSION AS OF {int(restore_version)}") + print(f"Restored {full_table} to Delta version {restore_version}") +else: + print("No restore_version widget supplied. Demo restore already completed above.") \ No newline at end of file diff --git a/submission/simerjit/notebooks/05_quality_checks.py b/submission/simerjit/notebooks/05_quality_checks.py new file mode 100644 index 0000000..a4446a1 --- /dev/null +++ b/submission/simerjit/notebooks/05_quality_checks.py @@ -0,0 +1,102 @@ +# Databricks notebook source +dbutils.widgets.text("catalog", "robustrade_dev") +catalog = dbutils.widgets.get("catalog") + +# COMMAND ---------- + +checks = [ + ( + "customers_current_pk_unique", + f""" + SELECT customer_id, COUNT(*) AS row_count + FROM {catalog}.silver.customers_current + GROUP BY customer_id + HAVING COUNT(*) > 1 + """, + ), + ( + "wallets_have_customers", + f""" + SELECT w.wallet_id + FROM {catalog}.silver.wallet_accounts_current w + LEFT JOIN {catalog}.silver.customers_current c + ON w.customer_id = c.customer_id + WHERE c.customer_id IS NULL + """, + ), + ( + "payments_have_wallets_and_customers", + f""" + SELECT p.transaction_id + FROM {catalog}.silver.payment_transactions_current p + LEFT JOIN {catalog}.silver.wallet_accounts_current w + ON p.wallet_id = w.wallet_id + LEFT JOIN {catalog}.silver.customers_current c + ON p.customer_id = c.customer_id + WHERE w.wallet_id IS NULL OR c.customer_id IS NULL + """, + ), + ( + "attempts_have_transactions", + f""" + SELECT a.attempt_id + FROM {catalog}.silver.payment_attempts_current a + LEFT JOIN {catalog}.silver.payment_transactions_current p + ON a.transaction_id = p.transaction_id + WHERE p.transaction_id IS NULL + """, + ), + ( + "non_negative_wallet_balances", + f""" + SELECT wallet_id + FROM {catalog}.silver.wallet_accounts_current + WHERE balance_minor < 0 + """, + ), + ( + "positive_payment_amounts", + f""" + SELECT transaction_id + FROM {catalog}.silver.payment_transactions_current + WHERE amount_minor <= 0 + """, + ), + ( + "valid_payment_statuses", + f""" + SELECT transaction_id + FROM {catalog}.silver.payment_transactions_current + WHERE transaction_status NOT IN ('AUTHORIZED', 'SETTLED', 'FAILED', 'REVERSED') + """, + ), + ( + "bronze_has_no_duplicate_sequences", + f""" + SELECT source_table, cdc_sequence, COUNT(*) AS row_count + FROM {catalog}.bronze.cdc_events + GROUP BY source_table, cdc_sequence + HAVING COUNT(*) > 1 + """, + ), +] + +# COMMAND ---------- + +failures = [] + +for check_name, sql in checks: + failed_rows = spark.sql(sql) + failure_count = failed_rows.count() + + if failure_count: + print(f"FAILED: {check_name}") + display(failed_rows) + failures.append(f"{check_name}: {failure_count} failing rows") + else: + print(f"PASSED: {check_name}") + +if failures: + raise ValueError("Quality checks failed:\n" + "\n".join(failures)) + +print("All quality checks passed.") diff --git a/submission/simerjit/notebooks/06_cdc_correctness_tests.py b/submission/simerjit/notebooks/06_cdc_correctness_tests.py new file mode 100644 index 0000000..bedffd3 --- /dev/null +++ b/submission/simerjit/notebooks/06_cdc_correctness_tests.py @@ -0,0 +1,128 @@ +# Databricks notebook source +dbutils.widgets.text("catalog", "robustrade_dev") + +catalog = dbutils.widgets.get("catalog") + +bronze_table = f"{catalog}.bronze.cdc_events" + +customers = f"{catalog}.silver.customers_current" +wallets = f"{catalog}.silver.wallet_accounts_current" +merchants = f"{catalog}.silver.merchants_current" +payments = f"{catalog}.silver.payment_transactions_current" +attempts = f"{catalog}.silver.payment_attempts_current" + +# COMMAND ---------- + +display( + spark.sql( + f""" + SELECT cdc_operation, COUNT(*) AS event_count + FROM {bronze_table} + GROUP BY cdc_operation + ORDER BY cdc_operation + """ + ) +) + +# COMMAND ---------- + +bronze_count = spark.table(bronze_table).count() + +assert bronze_count == 7, f"Expected 7 Bronze CDC events, got {bronze_count}" + +print("Bronze event count check passed.") + +# COMMAND ---------- + +payment_status = spark.sql( + f""" + SELECT transaction_status + FROM {payments} + WHERE transaction_id = 't_001' + """ +).collect()[0]["transaction_status"] + +assert payment_status == "SETTLED", f"Expected SETTLED, got {payment_status}" + +print("Payment update check passed.") + +# COMMAND ---------- + +last_sequence = spark.sql( + f""" + SELECT last_cdc_sequence + FROM {payments} + WHERE transaction_id = 't_001' + """ +).collect()[0]["last_cdc_sequence"] + +assert last_sequence == 5, f"Expected sequence 5, got {last_sequence}" + +print("Latest sequence check passed.") + +# COMMAND ---------- + +deleted_customer_count = spark.sql( + f""" + SELECT COUNT(*) AS row_count + FROM {customers} + WHERE customer_id = 'c_deleted' + """ +).collect()[0]["row_count"] + +assert deleted_customer_count == 0, ( + f"Expected deleted customer to be absent from Silver, got {deleted_customer_count}" +) + +print("Delete handling check passed.") + +# COMMAND ---------- + +expected_counts = { + "customers_current": 1, + "wallet_accounts_current": 1, + "merchants_current": 1, + "payment_transactions_current": 1, + "payment_attempts_current": 1, +} + +for table_name, expected_count in expected_counts.items(): + actual_count = spark.table(f"{catalog}.silver.{table_name}").count() + assert actual_count == expected_count, ( + f"{table_name}: expected {expected_count}, got {actual_count}" + ) + +print("Silver row count checks passed.") + +# COMMAND ---------- + +pk_checks = { + "customers_current": "customer_id", + "wallet_accounts_current": "wallet_id", + "merchants_current": "merchant_id", + "payment_transactions_current": "transaction_id", + "payment_attempts_current": "attempt_id", +} + +for table_name, pk in pk_checks.items(): + duplicate_count = spark.sql( + f""" + SELECT COUNT(*) AS duplicate_groups + FROM ( + SELECT {pk}, COUNT(*) AS row_count + FROM {catalog}.silver.{table_name} + GROUP BY {pk} + HAVING COUNT(*) > 1 + ) + """ + ).collect()[0]["duplicate_groups"] + + assert duplicate_count == 0, ( + f"{table_name}: expected no duplicate {pk}, got {duplicate_count}" + ) + +print("Primary key duplicate checks passed.") + +# COMMAND ---------- + +print("All CDC correctness checks passed.") \ No newline at end of file diff --git a/submission/simerjit/notebooks/08_catalog_metadata.py b/submission/simerjit/notebooks/08_catalog_metadata.py new file mode 100644 index 0000000..d5b8d94 --- /dev/null +++ b/submission/simerjit/notebooks/08_catalog_metadata.py @@ -0,0 +1,124 @@ +# Databricks notebook source +# MAGIC %md +# MAGIC # Catalog Metadata +# MAGIC +# MAGIC This notebook documents the lake and warehouse datasets created by the CDC lakehouse pipeline. +# MAGIC +# MAGIC The assignment asks for catalog exposure. In production, this metadata could be published to Unity Catalog, OpenMetadata, DataHub, or another governance catalog. +# MAGIC +# MAGIC For this assignment, I maintain catalog metadata inside this notebook. It documents: +# MAGIC +# MAGIC - dataset name +# MAGIC - layer +# MAGIC - platform +# MAGIC - owner +# MAGIC - description +# MAGIC - consumers +# MAGIC - primary key +# MAGIC - update cadence + +# COMMAND ---------- + +catalog_metadata = { + "version": 1, + "owner": "data-platform", + "domain": "wallet-payments", + "datasets": [ + { + "name": "robustrade_dev.bronze.cdc_events", + "layer": "lake", + "platform": "Databricks Delta Lake", + "description": "Append-only CDC event log retaining every source insert, update, and delete.", + "consumers": ["data engineering", "incident response", "backfill jobs"], + "update_cadence": "near real time via Auto Loader", + "metadata": { + "primary_ordering": "source_commit_ts, cdc_sequence", + "retention": "full assignment history", + }, + }, + { + "name": "robustrade_dev.silver.customers_current", + "layer": "warehouse", + "platform": "Databricks Delta Lake", + "description": "Latest non-deleted customer state.", + "primary_key": "customer_id", + "consumers": ["analytics", "risk", "customer operations"], + "update_cadence": "after each CDC merge job", + }, + { + "name": "robustrade_dev.silver.wallet_accounts_current", + "layer": "warehouse", + "platform": "Databricks Delta Lake", + "description": "Latest wallet balance and status.", + "primary_key": "wallet_id", + "consumers": ["analytics", "finance operations"], + "update_cadence": "after each CDC merge job", + }, + { + "name": "robustrade_dev.silver.merchants_current", + "layer": "warehouse", + "platform": "Databricks Delta Lake", + "description": "Latest merchant reference state.", + "primary_key": "merchant_id", + "consumers": ["analytics", "risk"], + "update_cadence": "after each CDC merge job", + }, + { + "name": "robustrade_dev.silver.payment_transactions_current", + "layer": "warehouse", + "platform": "Databricks Delta Lake", + "description": "Latest payment transaction state.", + "primary_key": "transaction_id", + "consumers": ["analytics", "finance", "reconciliation"], + "update_cadence": "after each CDC merge job", + }, + { + "name": "robustrade_dev.silver.payment_attempts_current", + "layer": "warehouse", + "platform": "Databricks Delta Lake", + "description": "Weak entity containing processor attempts for each payment transaction.", + "primary_key": "attempt_id", + "consumers": ["payments operations", "incident response"], + "update_cadence": "after each CDC merge job", + }, + ], +} + +# COMMAND ---------- + +display(spark.createDataFrame(catalog_metadata["datasets"])) + +# COMMAND ---------- + +expected_datasets = { + "robustrade_dev.bronze.cdc_events", + "robustrade_dev.silver.customers_current", + "robustrade_dev.silver.wallet_accounts_current", + "robustrade_dev.silver.merchants_current", + "robustrade_dev.silver.payment_transactions_current", + "robustrade_dev.silver.payment_attempts_current", +} + +actual_datasets = {dataset["name"] for dataset in catalog_metadata["datasets"]} + +missing = expected_datasets - actual_datasets + +assert not missing, f"Missing catalog entries: {missing}" + +print("Catalog metadata coverage check passed.") + +# COMMAND ---------- + +table_comments = { + "bronze.cdc_events": "Append-only CDC event log retaining every source insert, update, and delete.", + "silver.customers_current": "Latest non-deleted customer state.", + "silver.wallet_accounts_current": "Latest wallet balance and status.", + "silver.merchants_current": "Latest merchant reference state.", + "silver.payment_transactions_current": "Latest payment transaction state.", + "silver.payment_attempts_current": "Weak entity containing processor attempts for each payment transaction.", +} + +for table_name, comment in table_comments.items(): + spark.sql(f"COMMENT ON TABLE robustrade_dev.{table_name} IS '{comment}'") + +print("Unity Catalog table comments updated.") \ No newline at end of file diff --git a/submission/simerjit/notebooks/09_final_summary_or_pr_notes.py b/submission/simerjit/notebooks/09_final_summary_or_pr_notes.py new file mode 100644 index 0000000..f2b61ea --- /dev/null +++ b/submission/simerjit/notebooks/09_final_summary_or_pr_notes.py @@ -0,0 +1,193 @@ +# Databricks notebook source +# MAGIC %md +# MAGIC # CDC Lakehouse Reliability Assignment +# MAGIC +# MAGIC This project implements a Databricks, PySpark, and Delta Lake CDC lakehouse pipeline for a wallet and payments domain. +# MAGIC +# MAGIC The pipeline simulates source CDC events, loads them into an append-only Bronze Delta table, applies schema-safe Delta merges into Silver current-state tables, validates warehouse quality, documents catalog metadata, and demonstrates Delta time travel restore. + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Architecture +# MAGIC +# MAGIC ```text +# MAGIC Source model +# MAGIC -> Simulated CDC JSON events +# MAGIC -> Databricks Volume landing path +# MAGIC -> Auto Loader +# MAGIC -> Bronze Delta table +# MAGIC -> PySpark schema contracts +# MAGIC -> Delta MERGE +# MAGIC -> Silver current-state Delta tables +# MAGIC -> Quality checks +# MAGIC -> Catalog metadata +# MAGIC -> Delta time travel restore +# MAGIC ``` + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Notebook Run Order +# MAGIC +# MAGIC 1. `00_setup_and_source_model` +# MAGIC 2. `01_generate_cdc_events` +# MAGIC 3. `02_bronze_ingest_delta` +# MAGIC 4. `03_apply_silver_merges` +# MAGIC 5. `06_cdc_correctness_tests` +# MAGIC 6. `05_quality_checks` +# MAGIC 7. `08_catalog_metadata` +# MAGIC 8. `04_restore_with_delta_time_travel` +# MAGIC 9. `09_final_summary_or_pr_notes` + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Source Schema Design +# MAGIC +# MAGIC The source system models wallet and payment processing with five tables: +# MAGIC +# MAGIC - `customers` +# MAGIC - `wallet_accounts` +# MAGIC - `merchants` +# MAGIC - `payment_transactions` +# MAGIC - `payment_attempts` +# MAGIC +# MAGIC Strong entities are `customers`, `wallet_accounts`, `merchants`, and `payment_transactions`. +# MAGIC +# MAGIC `payment_attempts` is a weak entity because it depends on a parent payment transaction. +# MAGIC +# MAGIC The source design includes primary keys, foreign keys, indexes, timestamps, nullable optional fields, currency fields, status domains, non-negative balance checks, and positive payment amount checks. + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## CDC Strategy +# MAGIC +# MAGIC CDC is simulated using JSON event files written to a Databricks Volume. +# MAGIC +# MAGIC Each CDC event contains: +# MAGIC +# MAGIC - `source_table` +# MAGIC - `cdc_operation` +# MAGIC - `cdc_sequence` +# MAGIC - `source_commit_ts` +# MAGIC - `schema_version` +# MAGIC - `payload_json` +# MAGIC +# MAGIC The sample CDC stream includes inserts, one update, and one delete. +# MAGIC +# MAGIC In production, this simulated source would be replaced by database logs, Debezium, Fivetran, Kafka, or another CDC connector. The downstream Bronze and Silver logic would use the same event contract. + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Lake And Warehouse Modeling +# MAGIC +# MAGIC The lake layer is: +# MAGIC +# MAGIC - `robustrade_dev.bronze.cdc_events` +# MAGIC +# MAGIC This table is append-only and keeps the full CDC history. +# MAGIC +# MAGIC The warehouse layer is: +# MAGIC +# MAGIC - `robustrade_dev.silver.customers_current` +# MAGIC - `robustrade_dev.silver.wallet_accounts_current` +# MAGIC - `robustrade_dev.silver.merchants_current` +# MAGIC - `robustrade_dev.silver.payment_transactions_current` +# MAGIC - `robustrade_dev.silver.payment_attempts_current` +# MAGIC +# MAGIC Silver tables use Delta `MERGE` to keep the latest non-deleted row per primary key. + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Schema Change Safety +# MAGIC +# MAGIC Before writing to Silver, the pipeline validates each CDC payload against an explicit schema contract. +# MAGIC +# MAGIC The pipeline stops if: +# MAGIC +# MAGIC - a required column is missing +# MAGIC - a data type changes +# MAGIC - a required column becomes nullable +# MAGIC +# MAGIC This prevents incompatible source schema changes from silently corrupting downstream tables. + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Validation Parity +# MAGIC +# MAGIC Source business rules are mirrored in the warehouse quality notebook. +# MAGIC +# MAGIC Checks include: +# MAGIC +# MAGIC - primary key uniqueness +# MAGIC - referential integrity +# MAGIC - non-negative wallet balances +# MAGIC - positive payment amounts +# MAGIC - accepted transaction statuses +# MAGIC - duplicate CDC sequence detection +# MAGIC +# MAGIC The quality notebook raises an error if any check fails. + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Historical Recovery +# MAGIC +# MAGIC Recovery is handled in two ways: +# MAGIC +# MAGIC 1. Silver tables can be rebuilt from the append-only Bronze CDC log. +# MAGIC 2. Delta Lake time travel can restore tables using `DESCRIBE HISTORY` and `RESTORE TABLE`. +# MAGIC +# MAGIC The restore notebook demonstrates corrupting a development Silver table, restoring it to the previous version, and confirming the transaction status returns to `SETTLED`. + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Catalog Exposure +# MAGIC +# MAGIC Catalog metadata is documented in the `08_catalog_metadata` notebook. +# MAGIC +# MAGIC It includes: +# MAGIC +# MAGIC - dataset name +# MAGIC - layer +# MAGIC - platform +# MAGIC - owner +# MAGIC - description +# MAGIC - consumers +# MAGIC - primary key +# MAGIC - update cadence +# MAGIC +# MAGIC The notebook also adds Unity Catalog table comments. + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Test And Validation Results +# MAGIC +# MAGIC The following validations were completed: +# MAGIC +# MAGIC - Part 2 generated 7 CDC events. +# MAGIC - Part 3 loaded 7 rows into `bronze.cdc_events`. +# MAGIC - Part 4 schema contracts passed and unsafe type change correctly failed. +# MAGIC - Part 5 Silver merges created current-state tables. +# MAGIC - `payment_transactions_current` shows transaction `t_001` as `SETTLED`. +# MAGIC - Part 6 CDC correctness checks passed. +# MAGIC - Rerunning the Silver merge did not create duplicate rows. +# MAGIC - Part 7 quality checks passed. +# MAGIC - Part 8 catalog metadata coverage passed. +# MAGIC - Part 9 Delta restore returned the corrupted transaction back to `SETTLED`. + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Final Status +# MAGIC +# MAGIC The project satisfies the CDC lakehouse reliability assignment requirements using Databricks, PySpark, and Delta Lake. +# MAGIC +# MAGIC It demonstrates source modeling, CDC ingestion, Bronze history retention, Silver current-state modeling, schema-safe stopping, validation parity, catalog exposure, CDC correctness testing, idempotent replay, and Delta time travel recovery. diff --git a/submission/simerjit/schema_contracts.yml b/submission/simerjit/schema_contracts.yml new file mode 100644 index 0000000..e29a7a9 --- /dev/null +++ b/submission/simerjit/schema_contracts.yml @@ -0,0 +1,64 @@ +version: 1 + +tables: + customers: + primary_key: customer_id + required: + customer_id: string + email: string + full_name: string + customer_status: string + updated_at: timestamp + nullable: + phone_number: string + kyc_level: string + + wallet_accounts: + primary_key: wallet_id + required: + wallet_id: string + customer_id: string + currency: string + balance_minor: long + wallet_status: string + updated_at: timestamp + nullable: {} + + merchants: + primary_key: merchant_id + required: + merchant_id: string + merchant_name: string + merchant_category: string + merchant_status: string + created_at: timestamp + updated_at: timestamp + nullable: {} + + payment_transactions: + primary_key: transaction_id + required: + transaction_id: string + wallet_id: string + customer_id: string + amount_minor: long + currency: string + transaction_status: string + event_time: timestamp + updated_at: timestamp + nullable: + merchant_id: string + failure_code: string + + payment_attempts: + primary_key: attempt_id + required: + attempt_id: string + transaction_id: string + processor_name: string + attempt_status: string + attempted_at: timestamp + updated_at: timestamp + nullable: + failure_code: string + diff --git a/submission/simerjit/sql/source_schema.sql b/submission/simerjit/sql/source_schema.sql new file mode 100644 index 0000000..942cad6 --- /dev/null +++ b/submission/simerjit/sql/source_schema.sql @@ -0,0 +1,77 @@ +CREATE TABLE customers ( + customer_id STRING NOT NULL, + email STRING NOT NULL, + full_name STRING NOT NULL, + phone_number STRING, + kyc_level STRING, + customer_status STRING NOT NULL, + updated_at TIMESTAMP NOT NULL, + CONSTRAINT customers_pk PRIMARY KEY (customer_id), + CONSTRAINT customers_status_chk CHECK (customer_status IN ('ACTIVE', 'SUSPENDED', 'CLOSED')), + CONSTRAINT customers_kyc_chk CHECK (kyc_level IS NULL OR kyc_level IN ('BASIC', 'ENHANCED')) +); + +CREATE TABLE wallet_accounts ( + wallet_id STRING NOT NULL, + customer_id STRING NOT NULL, + currency STRING NOT NULL, + balance_minor BIGINT NOT NULL, + wallet_status STRING NOT NULL, + updated_at TIMESTAMP NOT NULL, + CONSTRAINT wallet_accounts_pk PRIMARY KEY (wallet_id), + CONSTRAINT wallet_customer_fk FOREIGN KEY (customer_id) REFERENCES customers(customer_id), + CONSTRAINT wallet_balance_non_negative CHECK (balance_minor >= 0), + CONSTRAINT wallet_currency_chk CHECK (currency IN ('USD', 'EUR', 'GBP')), + CONSTRAINT wallet_status_chk CHECK (wallet_status IN ('OPEN', 'FROZEN', 'CLOSED')) +); + +CREATE TABLE merchants ( + merchant_id STRING NOT NULL, + merchant_name STRING NOT NULL, + merchant_category STRING NOT NULL, + merchant_status STRING NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL, + CONSTRAINT merchants_pk PRIMARY KEY (merchant_id), + CONSTRAINT merchant_status_chk CHECK (merchant_status IN ('ACTIVE', 'SUSPENDED', 'CLOSED')) +); + +CREATE TABLE payment_transactions ( + transaction_id STRING NOT NULL, + wallet_id STRING NOT NULL, + customer_id STRING NOT NULL, + merchant_id STRING, + amount_minor BIGINT NOT NULL, + currency STRING NOT NULL, + transaction_status STRING NOT NULL, + failure_code STRING, + event_time TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL, + CONSTRAINT payment_transactions_pk PRIMARY KEY (transaction_id), + CONSTRAINT payment_wallet_fk FOREIGN KEY (wallet_id) REFERENCES wallet_accounts(wallet_id), + CONSTRAINT payment_customer_fk FOREIGN KEY (customer_id) REFERENCES customers(customer_id), + CONSTRAINT payment_merchant_fk FOREIGN KEY (merchant_id) REFERENCES merchants(merchant_id), + CONSTRAINT payment_amount_positive CHECK (amount_minor > 0), + CONSTRAINT payment_currency_chk CHECK (currency IN ('USD', 'EUR', 'GBP')), + CONSTRAINT payment_status_chk CHECK (transaction_status IN ('AUTHORIZED', 'SETTLED', 'FAILED', 'REVERSED')) +); + +CREATE TABLE payment_attempts ( + attempt_id STRING NOT NULL, + transaction_id STRING NOT NULL, + processor_name STRING NOT NULL, + attempt_status STRING NOT NULL, + failure_code STRING, + attempted_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL, + CONSTRAINT payment_attempts_pk PRIMARY KEY (attempt_id), + CONSTRAINT attempt_transaction_fk FOREIGN KEY (transaction_id) REFERENCES payment_transactions(transaction_id), + CONSTRAINT attempt_status_chk CHECK (attempt_status IN ('REQUESTED', 'APPROVED', 'DECLINED', 'ERROR')) +); + +CREATE INDEX customers_email_idx ON customers(email); +CREATE INDEX wallet_accounts_customer_idx ON wallet_accounts(customer_id); +CREATE INDEX payment_transactions_wallet_time_idx ON payment_transactions(wallet_id, event_time); +CREATE INDEX payment_transactions_customer_time_idx ON payment_transactions(customer_id, event_time); +CREATE INDEX payment_attempts_transaction_idx ON payment_attempts(transaction_id); +