🛒 ShopStream — End-to-End Databricks Lakehouse Project

Batch + Streaming E-Commerce Data Engineering Pipeline on Databricks

---

🎯 Project Objectives

• Build a practical lakehouse pipeline on Databricks.
• Ingest e-commerce CSV data into Delta tables.
• Apply data cleansing and validation in the Silver layer.
• Create business-ready Gold datasets.
• Demonstrate Delta Lake time travel and table history.
• Implement incremental file ingestion using Auto Loader.
• Create streaming Bronze, Silver, and Gold tables.
• Build a Databricks dashboard for sales analysis.
• Implement analytical transformations using dbt.
• Implement automated data quality testing using dbt.
• Implement SCD Type 2 customer history using dbt snapshots.
• Orchestrate the batch pipeline using Apache Airflow.
• Manage the project using Git and GitHub.

---

🏗️ Architecture

🛠️ Technology Stack

Technology	Purpose
Databricks	Lakehouse development and execution
Apache Spark / PySpark	Data transformation and streaming
SQL	Batch transformations and analytics
Delta Lake	Reliable table storage and time travel
Databricks Auto Loader	Incremental file ingestion
Unity Catalog	Catalog and table organization
Databricks SQL / Dashboards	Business analytics
dbt	Data transformation and modeling
dbt Snapshots	SCD Type 2 customer history
Apache Airflow	Batch pipeline orchestration
Git / GitHub	Version control

---

📂 Source Data

The batch pipeline works with:

• orders_2026_h1.csv
• customers.csv
• products.csv

The project uses the Unity Catalog namespace:

shopstream.core

Source data is stored in Databricks Volumes.

---

🥉 Bronze Layer

Notebook:

01_bronze_batch_ingestion.sql

Tables

shopstream.core.bronze_orders
shopstream.core.bronze_customers
shopstream.core.bronze_products

The Bronze layer uses Databricks SQL COPY INTO to ingest CSV files into Delta tables.

The pipeline also performs row-count checks after ingestion.

Verified Data

bronze_customers → 1,000 rows
bronze_orders    → 13,717 rows
bronze_products  → 197 rows

---

🥈 Silver Layer

Notebook:

02_silver_layer.sql

Tables

shopstream.core.silver_orders
shopstream.core.silver_customers
shopstream.core.silver_products

The Silver layer performs:

• Duplicate detection
• Deduplication using ROW_NUMBER()
• Quantity validation
• Data type conversion
• Revenue calculation
• Status standardization
• Empty coupon-code handling
• Removal of cancelled orders

Order-line revenue is calculated as:

line_revenue = quantity × unit_price

Product margin is calculated as:

unit_margin = unit_price - unit_cost

Verified Data

silver_customers → 1,000 rows
silver_orders    → 11,061 rows
silver_products  → 197 rows

---

🥇 Gold Layer

Notebook:

03_gold_layer.sql

Gold Tables

shopstream.core.gold_daily_revenue
shopstream.core.gold_category_performance
shopstream.core.gold_customer_ltv

Daily Revenue

Provides:

• Order date
• Number of orders
• Units sold
• Revenue

Only completed orders are included.

Category Performance

Provides:

• Category
• Orders
• Units sold
• Revenue
• Gross margin

Customer LTV

Provides:

• Customer ID
• Customer name
• Country
• Signup channel
• Lifetime orders
• Lifetime revenue

Verified Gold Data

gold_daily_revenue        → 168 rows
gold_category_performance → 8 rows
gold_customer_ltv         → 995 rows

---

🔄 Streaming Pipeline

Notebook:

04_Auto_Loader_Streaming.py

The project implements a separate streaming pipeline using Databricks Auto Loader.

Incoming Event Files
        ↓
Databricks Auto Loader
        ↓
bronze_orders_stream
        ↓
silver_orders_stream
        ↓
gold_daily_revenue_stream

The streaming implementation demonstrates:

• Auto Loader / cloudFiles
• Schema inference
• Schema location
• Checkpointing
• Structured Streaming
• Delta tables
• availableNow trigger
• Streaming deduplication
• Streaming transformations

Streaming Bronze

shopstream.core.bronze_orders_stream

Incoming files are incrementally loaded into a Delta table.

Streaming Silver

shopstream.core.silver_orders_stream

The streaming Silver transformation:

• Standardizes status values
• Validates quantity
• Keeps completed and returned statuses
• Calculates line_revenue
• Removes duplicate order_line_id values

The implemented quantity validation keeps values between 1 and 3.

Streaming Gold

shopstream.core.gold_daily_revenue_stream

Daily revenue is calculated from the streaming Silver table by grouping records by order date.

---

⏪ Delta Lake Time Travel

The project demonstrates Delta Lake table history and time travel using:

DESCRIBE HISTORY

and:

VERSION AS OF

Example:

SELECT *
FROM shopstream.core.gold_daily_revenue VERSION AS OF 0;

This demonstrates the ability to inspect and query previous versions of Delta tables.

---

🧰 dbt Transformation Layer

The project uses dbt Core with the Databricks adapter for SQL-based analytical transformations.

dbt connects to the Databricks Unity Catalog namespace:

shopstream.core

dbt Staging Models

stg_customers
stg_orders
stg_products

The staging layer provides a clean interface between the Databricks Silver tables and dbt analytical models.

dbt Mart Models

fct_daily_revenue
fct_category_performance
fct_customer_ltv

These models provide analytics-ready datasets for business reporting.

---

🧪 dbt Data Quality

The project includes 11 automated dbt data quality tests.

Tests cover important data integrity rules including:

• NOT NULL validation
• UNIQUE validation
• Accepted values
• Key field validation
• Source and model integrity

Latest validation:

PASS = 11
WARN = 0
ERROR = 0
SKIP = 0
TOTAL = 11

All 11/11 tests passed successfully.

---

🔁 SCD Type 2 — Customer History

Customer history is maintained using a dbt snapshot.

Snapshot:

customers_snapshot

The snapshot uses:

strategy = check
unique_key = customer_id

Tracked customer attributes include:

• name
• email
• city
• country
• signup_date
• signup_channel

The snapshot maintains:

dbt_scd_id
dbt_updated_at
dbt_valid_from
dbt_valid_to

SCD Type 2 Demonstration

A real customer attribute change was tested using:

Customer ID: C00001

Original city:

Singapore

Updated city:

Mumbai

The dbt snapshot successfully created two historical versions:

C00001 | Singapore | historical version
C00001 | Mumbai   | current version

The previous version received a populated dbt_valid_to, while the latest version remained active with dbt_valid_to = NULL.

This demonstrates:

• Historical record preservation
• Effective start timestamps
• Effective end timestamps
• Current active record identification
• Unique SCD version identifiers
• Change tracking using dbt snapshots

---

⏱️ Apache Airflow Orchestration

The batch pipeline is orchestrated using Apache Airflow.

DAG

shopstream_databricks_pipeline

Main Task

run_shopstream

The DAG is configured for scheduled batch execution.

The orchestration layer demonstrates:

• Scheduled batch execution
• Databricks pipeline orchestration
• DAG-based workflow management
• Task monitoring
• Pipeline execution tracking

---

📊 ShopStream Sales Dashboard

The project includes a Databricks dashboard named:

ShopStream Sales

The dashboard provides a business-facing view of the Gold-layer data.

Revenue by Category

Shows revenue performance across categories including:

• Beauty
• Books
• Electronics
• Fashion
• Fitness
• Grocery
• Home & Kitchen
• Toys

Daily Revenue

Shows the revenue trend across the available months in the dashboard.

The final analytics flow is:

Bronze
   ↓
Silver
   ↓
Gold
   ↓
ShopStream Sales Dashboard

---

📓 Project Structure

shopstream-data-engineering/
│
├── architecture/
│   └── shopstream-architecture.png
│
├── data/
│   ├── customers.csv
│   ├── products.csv
│   └── orders_2026_h1.csv
│
├── notebooks/
│   ├── 01_bronze_batch_ingestion.sql
│   ├── 02_silver_layer.sql
│   ├── 03_gold_layer.sql
│   └── 04_Auto_Loader_Streaming.py
│
├── screenshots/
│   └── shopstream-sales-dashboard.png
│
├── dbt_shopstream/
│   └── shopstream/
│       ├── dbt_project.yml
│       │
│       ├── models/
│       │   ├── staging/
│       │   │   ├── stg_customers.sql
│       │   │   ├── stg_orders.sql
│       │   │   ├── stg_products.sql
│       │   │   ├── sources.yml
│       │   │   └── schema.yml
│       │   │
│       │   └── marts/
│       │       ├── fct_daily_revenue.sql
│       │       ├── fct_category_performance.sql
│       │       └── fct_customer_ltv.sql
│       │
│       ├── snapshots/
│       │   └── customers_snapshot.sql
│       │
│       ├── macros/
│       ├── seeds/
│       ├── tests/
│       └── analyses/
│
├── .gitignore
└── README.md

📈 Project Validation Summary

Component	Result
Bronze customers	1,000
Bronze orders	13,717
Bronze products	197
Silver customers	1,000
Silver orders	11,061
Silver products	197
Gold daily revenue	168
Gold categories	8
Gold customer LTV	995
dbt data quality tests	11 / 11 passed
SCD Type 2	Successfully demonstrated
Airflow DAG	Successfully configured
Git repository	Clean and synced

---

👨‍💻 Author

Anshu Gupta

🔗 Connect with Me

• GitHub: https://github.com/anshu02042002
• LinkedIn: https://www.linkedin.com/in/anshu-gupta-de