E-Commerce Streaming Pipeline

A cloud-native ELT data engineering project that ingests e-commerce event data, processes it at scale using PySpark on Databricks, and loads it into a Snowflake Schema data warehouse following the Medallion Architecture (Bronze → Silver → Gold) — orchestrated with Azure Data Factory (or AWS Glue).

Business Problem

An e-commerce company generates high-volume event data from multiple touchpoints — customer orders, product clicks, returns, and payment events — arriving as JSON files throughout the day. Each event source has its own schema and lands independently in blob storage.

Without a unified pipeline, analytics teams cannot answer cross-domain questions like "which product categories drive the most returns?" or "which customer segments churn fastest after discount campaigns?"

This platform builds a cloud ELT pipeline that ingests raw JSON events into a Bronze landing zone, processes and normalizes them with PySpark on Databricks, and materializes a Snowflake Schema in the Gold layer — enabling fast analytical queries on revenue, churn, and customer behavior.

Business Questions Answered

What is the revenue trend by product category over the last 90 days?
Which customer segments have the highest return rates?
Which products are most frequently viewed but least purchased (conversion gap)?
What is the average time from order placement to delivery by region?
Which discount campaigns drove the most incremental revenue?
Which cities show the fastest growth in new customers?
What percentage of revenue comes from repeat vs. first-time customers?

Architecture

Raw JSON Events (Orders / Clicks / Returns / Payments)
        ↓
  Blob Storage           ← Azure Data Lake / S3 (raw landing zone)
        ↓
  Azure Data Factory     ← Ingest trigger → copy raw files to Bronze
  (or AWS Glue)
        ↓
  Bronze Layer           ← Raw JSON → Parquet, no transformation
        ↓
  PySpark on Databricks  ← Large-scale transformation & normalization
        ↓
  Silver Layer           ← Cleaned, typed, deduplicated Parquet tables
        ↓
  Business Transformations (PySpark SQL)
        ↓
  Gold Layer             ← Snowflake Schema, analytical model
        ↓
  SQL Analytics Layer    ← Views, aggregations, KPI queries
        ↓
  Power BI Dashboard     ← Business reporting & visualization

Medallion Layers

🥉 Bronze — Raw Ingestion

Stores raw event data exactly as received, converted to Parquet for efficient storage.

No transformations or business logic applied
Preserves original source schema
Partitioned by event date and event type
Captures ingestion metadata and file timestamps
Immutable — never overwritten, only appended

🥈 Silver — Cleaned & Normalized

PySpark jobs clean and standardize all event types into consistent schemas.

JSON field extraction and flattening
Column renaming to snake_case standards
Data type casting (timestamps, decimals, booleans)
Null value handling by column business rules
Duplicate event deduplication using event_id
Referential integrity checks across event types
Repartitioned and written as Parquet for Gold consumption

🥇 Gold — Business-Ready Analytics

Produces the final Snowflake Schema by joining normalized Silver tables.

Source	Description
Orders	Confirmed purchase transactions
Clicks	Product browse and view events
Returns	Return and refund events
Payments	Payment method and status
Customers	Customer profile and segment
Products	Product catalog with hierarchy

Snowflake Schema Design

The warehouse follows a Snowflake Schema — dimensions are normalized into sub-dimension tables to eliminate redundancy.

Fact Table

fact_orders

Column	Type	Description
`order_key`	BIGINT PRIMARY KEY	Surrogate key
`order_id`	STRING	Source event ID
`date_key`	INT (FK)	Foreign key → dim_date
`customer_key`	INT (FK)	Foreign key → dim_customer
`product_key`	INT (FK)	Foreign key → dim_product
`payment_key`	INT (FK)	Foreign key → dim_payment
`quantity`	INT	Units ordered
`unit_price`	DECIMAL	Price per unit
`discount_amount`	DECIMAL	Discount applied
`total_revenue`	DECIMAL	Net revenue
`is_returned`	BOOLEAN	Return flag
`delivery_days`	INT	Days from order to delivery
`ingested_at`	TIMESTAMP	Pipeline load timestamp

Dimension Tables

dim_date — Standard date dimension with day/month/quarter/year/weekday attributes

dim_customer

Column	Type	Description
`customer_key`	INT PRIMARY KEY	Surrogate key
`customer_id`	STRING	Source ID
`full_name`	STRING	Customer name
`email`	STRING	Email
`segment_key`	INT (FK)	Foreign key → dim_segment (normalized)
`city_key`	INT (FK)	Foreign key → dim_city (normalized)
`registration_date`	DATE	Sign-up date

dim_segment (sub-dimension of dim_customer)

Column	Type	Description
`segment_key`	INT PRIMARY KEY	Surrogate key
`segment_name`	STRING	VIP / Regular / New
`segment_tier`	STRING	Tier classification

dim_product

Column	Type	Description
`product_key`	INT PRIMARY KEY	Surrogate key
`product_id`	STRING	Source ID
`product_name`	STRING	Product name
`category_key`	INT (FK)	Foreign key → dim_category (normalized)
`unit_cost`	DECIMAL	Cost of goods

dim_category (sub-dimension of dim_product)

Column	Type	Description
`category_key`	INT PRIMARY KEY	Surrogate key
`category_name`	STRING	Electronics / Apparel…
`sub_category`	STRING	Sub-category
`department`	STRING	Department

dim_payment

Column	Type	Description
`payment_key`	INT PRIMARY KEY	Surrogate key
`payment_method`	STRING	Card / Wallet / COD
`payment_status`	STRING	Completed / Failed / Refunded
`gateway`	STRING	Payment gateway name

dim_city (sub-dimension of dim_customer)

Column	Type	Description
`city_key`	INT PRIMARY KEY	Surrogate key
`city_name`	STRING	City
`region`	STRING	Region
`country`	STRING	Country

Technologies

Category	Tool / Library
Language	Python 3.11+
Big Data Processing	Apache Spark / PySpark
Cloud Platform	Databricks (Community Edition or workspace)
Orchestration	Azure Data Factory (or AWS Glue)
Storage	Azure Data Lake Gen2 / AWS S3 (simulated locally)
Database	SQL Server / Azure Synapse (or local PostgreSQL for dev)
ORM / DB Driver	SQLAlchemy, pyodbc
Data Processing	PySpark DataFrame API + Spark SQL
Data Modeling	Snowflake Schema (Dimensional Modeling)
Pipeline Pattern	ELT + Medallion Architecture
File Format	JSON (source) → Parquet (Bronze/Silver/Gold)
Version Control	Git + GitHub
CI/CD	GitHub Actions
Visualization (upcoming)	Power BI

Data Sources

Simulated e-commerce event data (JSON files — generated with Faker or sourced from Kaggle):

File / Stream	Description
`orders/*.json`	Purchase order events
`clicks/*.json`	Product click and browse events
`returns/*.json`	Return and refund events
`payments/*.json`	Payment transaction events
`customers.json`	Customer master records
`products.json`	Product catalog

All files land in storage/raw/ (simulating blob storage) before pipeline execution.

Project Structure

ecommerce-streaming-pipeline/
│
├── .github/
│   └── workflows/
│       └── ci.yml                        # GitHub Actions CI pipeline
│
├── databricks/
│   └── notebooks/
│       ├── bronze_ingest.py              # Raw JSON → Parquet (Bronze)
│       ├── silver_transform.py           # PySpark cleaning & normalization
│       ├── gold_build.py                 # Snowflake schema construction
│       └── analytics_views.py            # Gold layer SQL views
│
├── adf/                                  # Azure Data Factory pipeline definitions
│   └── pipelines/
│       ├── ingest_orders_pipeline.json
│       ├── ingest_clicks_pipeline.json
│       └── trigger_databricks_job.json
│
├── etl/
│   ├── bronze/
│   │   ├── __init__.py
│   │   ├── orders_ingestor.py            # Read orders JSON → Bronze Parquet
│   │   ├── clicks_ingestor.py
│   │   ├── returns_ingestor.py
│   │   └── payments_ingestor.py
│   │
│   ├── silver/
│   │   ├── __init__.py
│   │   ├── orders_transformer.py         # PySpark Silver transformations
│   │   ├── clicks_transformer.py
│   │   ├── returns_transformer.py
│   │   └── payments_transformer.py
│   │
│   └── gold/
│       ├── __init__.py
│       ├── dim_builder.py                # Build all dimension tables
│       ├── fact_builder.py               # Build fact_orders from Silver joins
│       └── loader.py                     # Write Gold → SQL Server / Synapse
│
├── db/
│   └── migrations/
│       ├── create_dimensions.sql         # DDL for Snowflake Schema dimensions
│       ├── create_fact.sql               # DDL for fact_orders
│       └── create_views.sql              # Analytical views on Gold layer
│
├── analytics/
│   └── queries/
│       ├── revenue_by_category.sql       # Revenue trend by product category
│       ├── return_rate_by_segment.sql    # Return rate by customer segment
│       ├── conversion_gap.sql            # Views vs purchases analysis
│       ├── repeat_vs_new.sql             # Repeat customer revenue split
│       └── delivery_performance.sql      # Avg delivery days by region
│
├── config/
│   ├── settings.py                       # DB credentials, paths, Spark config
│   └── schemas/
│       ├── bronze_schemas.py             # Expected JSON field definitions
│       ├── silver_schemas.py             # Silver Parquet column types
│       └── gold_schemas.py               # Gold table column definitions
│
├── storage/
│   ├── raw/                              # Simulated blob storage landing zone
│   │   ├── orders/
│   │   ├── clicks/
│   │   ├── returns/
│   │   └── payments/
│   ├── bronze/                           # Bronze Parquet tables
│   ├── silver/                           # Silver Parquet tables
│   └── gold/                             # Gold Parquet tables (pre-SQL load)
│
├── tests/
│   ├── test_bronze.py                    # Unit tests for ingestors
│   ├── test_silver.py                    # PySpark transform unit tests
│   ├── test_gold.py                      # Gold join logic tests
│   └── test_quality.py                   # Data quality assertion tests
│
├── logs/                                 # Pipeline execution logs
├── main.py                               # Local pipeline entry point (without ADF)
├── requirements.txt                      # Python dependencies
├── docker-compose.yml                    # Local Spark + PostgreSQL setup
├── .env.example                          # Environment variable template
├── .gitignore
└── README.md

Pipeline Stages

1. Ingest (Bronze)

Azure Data Factory (or AWS Glue) detects new JSON files in the raw landing zone and copies them to the Bronze layer. Files are converted to Parquet and partitioned by event_date and event_type. No business logic applied — raw data is preserved as-is.

2. Transform (Silver) — PySpark on Databricks

A PySpark notebook runs on a Databricks cluster to process each Bronze table:

Flatten nested JSON fields into flat columns
Cast all fields to correct Spark data types
Apply null handling rules per column business definition
Deduplicate events using event_id as the unique key
Validate referential keys (customer_id, product_id) exist in master tables
Write output as partitioned Parquet to the Silver layer

3. Build (Gold) — Snowflake Schema

A second PySpark job joins all Silver tables to produce the Snowflake Schema:

Build sub-dimension tables first (dim_segment, dim_category, dim_city)
Build dimension tables with FK references to sub-dimensions
Build fact_orders by joining all dimension keys onto Silver order records
Compute derived metrics: total_revenue, is_returned, delivery_days
Write Gold tables to SQL Server (or Azure Synapse) via SQLAlchemy / JDBC

4. Analytics Layer

SQL views on the Gold layer pre-aggregate common business queries for dashboard consumption.

Orchestration

Azure Data Factory (Primary)

Pipeline 1 — Ingest Triggered on new file arrival in raw storage. Copies JSON → Bronze Parquet. Runs per event type (orders, clicks, returns, payments).

Pipeline 2 — Transform and Build Triggered on completion of all four ingest pipelines. Calls the Databricks job via ADF Databricks Notebook activity. Runs Silver → Gold in sequence.

Schedule: Hourly ingest trigger; daily full Gold rebuild at 03:00 UTC.

AWS Glue (Alternative)

Replace ADF pipelines with Glue Jobs and Glue Workflows. The same PySpark transformation logic runs on Glue ETL jobs without modification.

Data Quality Checks

Run automatically after each Silver and Gold write:

Check	Layer	Description
Schema validation	Bronze	All expected JSON fields present
Null check	Silver	No nulls in key business columns
Duplicate event check	Silver	`event_id` unique per table
Referential integrity	Gold	All FK keys resolve to dimension rows
Row count validation	Silver & Gold	Output rows ≥ expected threshold
Revenue sanity	Gold	`total_revenue` = `unit_price` × `quantity` − `discount`
Date range check	Gold	No future-dated orders
Return rate anomaly	Gold	Return rate < 50% (alert if breached)

How to Run

Option A — Local (without Databricks / ADF)

# Clone the repo
git clone https://github.com/your-username/ecommerce-streaming-pipeline.git
cd ecommerce-streaming-pipeline

# Copy environment config
cp .env.example .env
# Edit .env with DB credentials and storage paths

# Install dependencies
pip install -r requirements.txt

# Start local PostgreSQL with Docker
docker-compose up -d

# Generate sample JSON data
python scripts/generate_sample_data.py

# Run full pipeline locally (Bronze → Silver → Gold)
python main.py

Option B — Databricks Community Edition

Sign up at community.cloud.databricks.com
Upload storage/raw/ files to DBFS (/FileStore/raw/)
Import notebooks from databricks/notebooks/ into your workspace
Run notebooks in order: bronze_ingest → silver_transform → gold_build
Query Gold tables using Databricks SQL editor

Option C — Azure Data Factory

Deploy ADF pipeline JSON definitions from adf/pipelines/
Configure linked services for Azure Data Lake and Databricks
Set up trigger on raw storage container
Monitor pipeline runs in ADF Monitor view

Development Progress

✅ Completed

Component	Status	Details
Project Structure	✅ Done	Modular Bronze/Silver/Gold layout
Bronze Ingestors	✅ Done	4 JSON → Parquet ingestor modules
Silver PySpark Jobs	✅ Done	Cleaning, dedup, type casting
Gold Dim Builder	✅ Done	All dimension + sub-dimension tables
Gold Fact Builder	✅ Done	fact_orders with all FK joins
SQL Migrations	✅ Done	DDL for Snowflake Schema
Databricks Notebooks	✅ Done	Bronze/Silver/Gold notebooks
Config Management	✅ Done	Centralized settings.py + .env
CI/CD	✅ Done	GitHub Actions workflow

🔄 In Progress

ADF pipeline JSON definitions
Data quality check automation
Unit tests for PySpark transforms

⏳ Upcoming

Power BI dashboard connected to Gold layer
Incremental loading (process only new partitions)
AWS Glue alternative implementation
Docker deployment for local Spark

Future Improvements

Version	Feature
v2	Incremental / partition-aware loading
v3	Real-time streaming with Apache Kafka
v4	Data quality framework (Great Expectations / Soda)
v5	dbt for Gold layer transformations
v6	Azure Synapse Analytics as the serving layer
v7	Power BI dashboard with live DirectQuery
v8	ML feature store on Gold layer

Learning Outcomes

This project demonstrates practical knowledge of:

ELT Pipeline Design (cloud-native)
Medallion Architecture (Bronze / Silver / Gold)
Dimensional Modeling — Snowflake Schema
PySpark DataFrame API and Spark SQL
Databricks notebook workflow
Azure Data Factory pipeline orchestration (or AWS Glue)
Large-scale data processing at scale
SQL Server / Azure Synapse as the serving layer
SQLAlchemy for Python → SQL integration
Data Quality Validation at each layer
CI/CD with GitHub Actions
Parquet file format and partition strategies

Name		Name	Last commit message	Last commit date
Latest commit History 1 Commit
README.MD		README.MD

Folders and files

Latest commit

History

Repository files navigation

E-Commerce Streaming Pipeline

Table of Contents

Business Problem

Business Questions Answered

Architecture

Medallion Layers

🥉 Bronze — Raw Ingestion

🥈 Silver — Cleaned & Normalized

🥇 Gold — Business-Ready Analytics

Snowflake Schema Design

Fact Table

Dimension Tables

Technologies

Data Sources

Project Structure

Pipeline Stages

1. Ingest (Bronze)

2. Transform (Silver) — PySpark on Databricks

3. Build (Gold) — Snowflake Schema

4. Analytics Layer

Orchestration

Azure Data Factory (Primary)

AWS Glue (Alternative)

Data Quality Checks

How to Run

Option A — Local (without Databricks / ADF)

Option B — Databricks Community Edition

Option C — Azure Data Factory

Development Progress

✅ Completed

🔄 In Progress

⏳ Upcoming

Future Improvements

Learning Outcomes

About

Resources

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Uh oh!

Contributors

Uh oh!

Packages