Lab Type

Hands-On Engineering

Difficulty

Advanced Hands-On

Reward

600 XP · 🛠️ Pipeline Builder

Core Stack

PySpark · Delta Lake · SQL

Target Deliverable

Bronze → Silver → Gold Pipeline

⚡ Hands-On Engineering Lab

DEV-011

Build a Production-Ready Sales Pipeline

You receive raw retail CSV files plagued with realistic real-world flaws: duplicate transactions, corrupted price records, missing customer keys, and untrimmed strings. Your mission is to build, execute, and validate a production-grade Medallion Lakehouse pipeline from scratch.

Mission & Context

Inspect Starter Data

Build Pipeline Code

Run Validation Tests

Capture & Submit Evidence

Mission & Business Context

Understand the operational problem, financial stakes, and your architectural goals.

The Engineering Scenario

The retail operations team at OmniCart Retail imports hundreds of store sales CSV files daily into cloud storage. Currently, analysts query these files directly, which has led to severe business escalations: duplicate transactions inflated regional revenue figures by 8%, while corrupted records with negative quantities silently skewed inventory audits.

As the incoming Data Engineer, you must replace this chaotic ad-hoc process with an automated, idempotent PySpark Medallion Lakehouse pipeline (Bronze → Silver → Gold). Bad records must be intercepted and quarantined without failing the batch, deduplication must be enforced deterministically, and Gold analytical marts must provide rock-solid reconciled metrics.

Zero Inferred Schemas

InferSchema in production creates silent casting bugs. Define explicit StructTypes.

Quarantine vs. Drop

Never drop invalid records silently. Route bad rows to a quarantine table with rejection reasons.

Strict Deduplication

Enforce unique primary keys (txn_id) while preserving the latest valid update.

Financial Reconciliation

Gold net revenue must reconcile 100% against Silver clean transaction sums.

Starter Materials & Datasets

Inspect the real input datasets and boilerplate starter script. Copy and use them in your local or Databricks environment.

📄 customers.csv

customer_id,first_name,last_name,email,phone,city,state,signup_date
C101,Aarav,Sharma,aarav.sharma@example.com,+91-9876543210,Mumbai,MH,2023-01-15
C102,Diya,Patel,diya.patel@example.com,9876543211,Ahmedabad,GJ,2023-02-20
C103,Rohan,Gupta,rohan.gupta@example.com,+91-9876543212,Bengaluru,KA,2023-03-10
C104,Ananya,Iyer,,9876543213,Chennai,TN,2023-04-05
C105,Kabir,Mehta,kabir.mehta@example.com,+91-9876543214,Delhi,DL,2023-05-12
C102,Diya,Patel,diya.patel@example.com,9876543211,Ahmedabad,GJ,2023-02-20
C106,Pooja,Verma,pooja.verma@example.com,INVALID_PHONE,Pune,MH,2023-06-18
C107,Vikram,Singh,vikram.singh@domain_missing,9876543216,Jaipur,RJ,2023-07-22
C108,   Neha   ,Deshmukh,neha.d@example.com,+91-9876543217,Nagpur,MH,2023-08-30
C109,Aditya,Reddy,aditya.reddy@example.com,9876543218,Hyderabad,TS,2023-09-14

⚠️ Data Flaw Note: Notice duplicate C102, missing email for C104, invalid phone for C106, untrimmed whitespace on C108.

Hands-On Implementation Steps

Execute each step in your PySpark environment. Do not skip data-quality or schema checks.

Source Inspection & Schema Modeling

Define explicit PySpark StructTypes for customers, products, and sales. Do NOT use inferSchema=True.

▸Explicitly map monetary fields (price, cost, discount) to DecimalType(10,2).

▸Map timestamps to TimestampType() and quantities to IntegerType().

▸Ensure nullable=False on essential primary keys (customer_id, product_id, txn_id).

Bronze Layer Ingestion & Audit Metadata

Read raw CSVs using the defined schemas and append ingestion metadata columns.

▸Append _ingested_at = current_timestamp() and _source_file = input_file_name().

▸Save raw DataFrames to Delta tables: bronze_customers, bronze_products, bronze_sales.

▸Ensure the Bronze layer is append-only and retains the raw state intact.

Silver Cleansing & Quarantine Routing

Implement robust cleansing rules and separate good records from corrupt records.

▸Rule 1: quantity must be strictly > 0. Route negative or zero quantities to quarantine.

▸Rule 2: unit_price must be > 0.00.

▸Rule 3: Deduplicate sales by txn_id keeping the latest record by txn_timestamp.

▸Rule 4: Trim all string columns (first_name, last_name, category) to remove leading/trailing whitespace.

▸Rule 5: Compute net_amount = (quantity * unit_price) - discount.

▸Write clean records to silver_sales and rejected rows to quarantine_sales with reject_reason.

Silver Enrichment & Conformed Lookups

Join silver_sales with silver_customers and silver_products.

▸Perform LEFT JOIN with products to enrich category and product_name.

▸Perform LEFT JOIN with customers to enrich customer city and state.

▸If customer_id is missing or non-existent (e.g. C999), map customer_key to default -1 (Unknown Customer).

Gold Layer Analytics Mart Construction

Compute executive business metrics and write to optimized Gold Delta tables.

▸Gold Table 1: gold_daily_store_sales (txn_date, store_id, total_revenue, total_units, txn_count, avg_order_value).

▸Gold Table 2: gold_product_category_sales (category, units_sold, gross_revenue, net_revenue).

▸Gold Table 3: gold_customer_metrics (customer_id, total_spend, total_transactions, last_purchase_date).

Expected Output & Schema Specifications

Verify your resulting Delta tables match the expected schema and row counts.

Expected Table Volumes & Outcomes

Raw Sales Rows

13 Rows

Including duplicates & flaws

Silver Clean Sales

9 Rows

Deduplicated & validated

Quarantined Rows

4 Rows

Bad quantity, future date, nulls

Gold Store Daily Records

6 Store-Date Rows

Aggregated metrics

Gold Table Schema (gold_daily_store_sales)

Column Name	Data Type	Description
txn_date	DATE	Calendar date of transactions (derived from txn_timestamp)
store_id	STRING	Unique store identifier
total_net_sales	DECIMAL(12,2)	Sum of net_amount after discount deductions
total_units_sold	INTEGER	Sum of item quantities
txn_count	INTEGER	Count of unique clean transactions
avg_order_value	DECIMAL(10,2)	total_net_sales / txn_count

Validation Checks & Test Queries

Execute these test assertions against your completed pipeline to verify data integrity.

Run these queries in PySpark / Spark SQL. All tests must pass before you submit your evidence:

SQL Test 1 — Primary Key Uniqueness (Must return 0 rows)

SELECT txn_id, COUNT(*) AS cnt 
FROM silver_sales 
GROUP BY txn_id 
HAVING COUNT(*) > 1;

SQL Test 2 — Zero Negative Amounts in Silver (Must return 0 rows)

SELECT * FROM silver_sales 
WHERE quantity <= 0 OR unit_price <= 0 OR net_amount < 0;

SQL Test 3 — Financial Reconciliation (Variance must be exactly 0.00)

SELECT 
    (SELECT ROUND(SUM(net_amount), 2) FROM silver_sales) AS silver_net,
    (SELECT ROUND(SUM(total_net_sales), 2) FROM gold_daily_store_sales) AS gold_net,
    ROUND((SELECT SUM(net_amount) FROM silver_sales) - (SELECT SUM(total_net_sales) FROM gold_daily_store_sales), 2) AS variance;

Evidence To Submit

Checklist of artifacts and outputs you need to document to complete the lab.

To receive grading and claim your badge, compile the following proof artifacts:

☑

1. Pipeline Source Code

Complete, well-commented PySpark script or exported Jupyter/Databricks notebook (.py / .ipynb).

☑

2. Quarantine Audit Table Screenshot

Query output showing rows routed to quarantine_sales along with their reject_reason values.

☑

3. Silver Clean Sales Record Count

Screenshot or log showing exactly 9 clean deduplicated rows in silver_sales.

☑

4. Gold Mart Summary Output

Formatted query output table of gold_daily_store_sales and gold_product_category_sales.

☑

5. Financial Reconciliation Result

Output of SQL Test 3 proving variance = 0.00 between Silver and Gold layers.

☑

6. Design Rationale Summary (1-2 paragraphs)

Brief explanation of how you handled late/future dates and why inferSchema is risky in production.

Common Mistakes To Avoid

Review these frequent anti-patterns discovered in production data engineering reviews.

⚠️ Using inferSchema=True in Production

inferSchema requires an extra scan over the dataset and can change column data types unexpectedly if an upstream system introduces a string in a numeric field.

⚠️ Dropping Bad Records Without Auditing

Filtering out invalid records with a simple df.filter(quantity > 0) without writing them to a quarantine sink causes silent data loss and makes debugging upstream vendor feeds impossible.

⚠️ Naive Deduplication on Partial Columns

Running dropDuplicates(['txn_id']) without sorting by txn_timestamp might retain an older stale transaction version instead of the latest update.

⚠️ Floating Point Rounding Errors on Monetary Fields

Using FloatType or DoubleType for prices and discounts creates IEEE 754 precision drift (e.g. $49.99000000000001). Always use DecimalType(10,2).

Stretch / Bonus Objectives

Want to take this lab to the next level? Tackle these optional advanced engineering challenges.

⭐ Parameterized Execution

Refactor the pipeline into a modular CLI script accepting --input-path, --output-path, and --execution-date arguments via argparse.

⭐ Automated Great Expectations

Implement automated schema validation assertions using Great Expectations or Soda Core before writing to Gold.

⭐ Delta Lake Time Travel

Demonstrate Delta table time-travel by running RESTORE TABLE silver_sales TO VERSION AS OF 0 and verifying table history.

Lab Certification

Complete DEV-011 to Earn 600 XP & Pipeline Builder Badge

Build the pipeline, pass all 3 validation SQL queries, compile your evidence checklist, and submit your work for review.

🛠️

600 XP

Pipeline Builder