CDC-based data platform for an e-commerce transactional system - #8
Open
abir-qureshi99 wants to merge 2 commits into
Open
CDC-based data platform for an e-commerce transactional system#8abir-qureshi99 wants to merge 2 commits into
abir-qureshi99 wants to merge 2 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR replaces the prior DuckDB/pytest-based “payments/wallet” CDC exercise with a PySpark-based CDC platform for an e-commerce transactional domain, adding Bronze/Silver/Warehouse layers, contract validation, Airflow orchestration, and updated documentation/catalog metadata.
Changes:
- Reworks the domain model from customers/wallets/transactions to customers/products/orders/order_items/payment_attempts, with config-driven schema + enum contracts.
- Adds PySpark-based ingestion and modeling: Bronze CDC storage, Silver snapshots, and Warehouse SCD2 dimensions + fact tables.
- Introduces orchestration and operational scripts (Airflow DAG, validation, build steps), while removing the previous test suite and DuckDB-based pipeline.
Reviewed changes
Copilot reviewed 35 out of 48 changed files in this pull request and generated 10 comments.
Show a summary per file
| File | Description |
|---|---|
| submission/sample-candidate/tests/test_schema_contracts.py | Removes schema-contract tests (DuckDB-era). |
| submission/sample-candidate/tests/test_data_quality.py | Removes warehouse/lake correctness tests (DuckDB-era). |
| submission/sample-candidate/tests/test_cdc.py | Removes CDC capture tests (DuckDB-era). |
| submission/sample-candidate/tests/test_catalog.py | Removes catalog metadata tests (DuckDB-era). |
| submission/sample-candidate/tests/conftest.py | Removes DuckDB/CDC pytest fixtures. |
| submission/sample-candidate/source/seed_data.py | Adds Spark-based seed data writer for source parquet datasets. |
| submission/sample-candidate/source/postgres.py | Adds simulated Postgres metadata reader for schema + enum domains. |
| submission/sample-candidate/source/models.py | Replaces DuckDB DDL + SCHEMA_CONTRACT with dataclasses/enums for e-commerce domain objects. |
| submission/sample-candidate/source/create_tables.py | Adds Spark-based “empty table” parquet writers for source datasets. |
| submission/sample-candidate/scripts/validate_schema.py | Adds schema + enum contract validation script (config-driven). |
| submission/sample-candidate/scripts/validate_catalog.py | Removes prior catalog validator script. |
| submission/sample-candidate/scripts/status_transition_validation.py | Adds Spark-based validator for order status transition rules. |
| submission/sample-candidate/scripts/run_data_quality_checks.py | Replaces DuckDB-based checks with Spark-based DQ checks against warehouse parquet outputs. |
| submission/sample-candidate/scripts/rebuild_warehouse.py | Adds a rebuild/recovery helper intended for point-in-time replay. |
| submission/sample-candidate/scripts/publish_catalog.py | Adds a catalog publisher/logger with required-field validation. |
| submission/sample-candidate/scripts/ingest_cdc.py | Adds Spark-based CDC ingestion + Bronze write path with simulated events. |
| submission/sample-candidate/scripts/check_schema_contracts.py | Removes DuckDB schema-contract check script. |
| submission/sample-candidate/scripts/build_warehouse.py | Adds Spark-based warehouse build from Silver inputs (SCD2 dims + facts). |
| submission/sample-candidate/scripts/build_silver.py | Adds Spark-based Silver snapshot construction from Bronze CDC. |
| submission/sample-candidate/requirements.txt | Removes prior DuckDB/pytest dependency pinning. |
| submission/sample-candidate/pipeline/warehouse.py | Replaces DuckDB upsert warehouse logic with Spark SCD2 dimension processing and parquet writer. |
| submission/sample-candidate/pipeline/silver.py | Adds a small SilverLayer helper (JSON parsing/flattening). |
| submission/sample-candidate/pipeline/schema_contracts.py | Adds schema contract validator + exception types. |
| submission/sample-candidate/pipeline/lake.py | Removes prior DuckDB lake (append-only table) implementation. |
| submission/sample-candidate/pipeline/cdc.py | Replaces in-process CDC capture with Spark DataFrame ingestion utilities. |
| submission/sample-candidate/pipeline/catalog.py | Adds minimal catalog reader/publisher helper. |
| submission/sample-candidate/pipeline/bronze.py | Adds Bronze layer parquet writer/reader and schema definition. |
| submission/sample-candidate/conftest.py | Removes root pytest path setup. |
| submission/sample-candidate/config/schema_contracts.json | Adds schema contracts configuration for tables/columns/types. |
| submission/sample-candidate/config/enum_contracts.json | Adds enum domain contracts configuration. |
| submission/sample-candidate/catalog/catalog.json | Updates catalog entries to bronze/silver/warehouse dataset set and metadata fields. |
| submission/sample-candidate/airflow_dags/ecommerce_cdc_pipeline.py | Adds Airflow DAG orchestrating validation → ingestion → silver → warehouse → DQ → catalog. |
| README.md | Rewrites README to describe the e-commerce CDC platform architecture and usage. |
| data_evaluation_guide.md | Removes prior evaluation guide document. |
| DATA_ASSIGNMENT.md | Removes prior assignment specification document. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+84
to
+92
| 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 |
Comment on lines
+1
to
+4
| import json | ||
| from datetime import datetime | ||
| from pyspark.sql import SparkSession | ||
| from pyspark.sql import Row |
Comment on lines
+37
to
56
| 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 | ||
| ) |
Comment on lines
+95
to
+104
| 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], | ||
| ) |
Comment on lines
+85
to
+88
| .withColumn( | ||
| "effective_to", | ||
| F.lit(None) | ||
| ) |
| FactPaymentBuilder | ||
| ) | ||
|
|
||
| from pipeline.silver import SilverLayer |
Comment on lines
+37
to
+61
| 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 | ||
| ) | ||
| ) |
Comment on lines
+47
to
+50
|
|
||
| for column_name in df.columns(): | ||
| failures = ( | ||
|
|
Comment on lines
+10
to
+17
| 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() | ||
| ) |
Comment on lines
+118
to
+123
| rows = order_updates.select( | ||
| "primary_key", | ||
| "old_status", | ||
| "new_status" | ||
| ).collect() | ||
|
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.