Ecommerce Data Pipeline

Overview

This project demonstrates an end-to-end data engineering pipeline built using modern data stack tools. It integrates structured data from MySQL and semi-structured JSON data into Snowflake, followed by layered transformations using dbt.

The pipeline enables advanced analytics such as customer lifetime value (CLV), product performance, customer segmentation, time-series trends, and anomaly detection.

Architecture

CSV / MySQL (DBeaver)
        ↓
StreamSets (Ingestion Pipelines)
        ↓
Snowflake (RAW Layer + JSON Stage)
        ↓
dbt (Transformations)
        ↓
STAGING → INTERMEDIATE → MARTS
        ↓
Analytics Tables + Tests + Documentation

• mysql acts as the oltp system
• streamsets handles ingestion
• snowflake is the analytical warehouse
• and dbt is used for transformation and modeling

Tech Stack

• MySQL (source system)
• DBeaver (data setup and management)
• StreamSets Data Collector (data ingestion)
• Snowflake (cloud data warehouse)
• dbt (data transformation and modeling)
• GitHub (version control)

---

Data Sources

1. Structured Data (MySQL via CSV)

Loaded manually into MySQL using DBeaver:

• customers
• products
• orders

These were then ingested into Snowflake using StreamSets.

---

2. Semi-Structured Data (JSON Files)

Multiple JSON files were created, each representing product-level reviews:

product_101_reviews.json
product_102_reviews.json
product_103_reviews.json
product_104_reviews.json
product_105_reviews.json

Each file contains:

• product_id
• rating
• review_text
• timestamp

These were uploaded to Snowflake using:

• Internal stage (reviews_stage)
• COPY INTO commands
• Flattening via dbt

---

Data Ingestion

StreamSets Pipelines

Separate pipelines were configured for:

• customers
• products
• orders

Configuration highlights:

• Origin: JDBC Query Consumer (MySQL)
• Mode: Incremental
• Offset Columns:
  • customers → customer_id
  • products → product_id
  • orders → order_timestamp
• Destination: Snowflake

---

Snowflake JSON Ingestion

Steps:

1. Create stage
2. Upload JSON files
3. Load into raw table (preserved in VARIANT format)
4. Parse using FLATTEN in dbt

---

dbt Project Structure

models/
  staging/
    stg_customers.sql
    stg_products.sql
    stg_orders.sql
    stg_reviews.sql

  intermediate/
    int_customer_orders.sql
    int_product_sales.sql
    int_product_reviews.sql

  marts/
    mart_customer_segment.sql
    mart_product_performance.sql
    mart_sales_trends.sql
    mart_sales_anomalies.sql
    mart_sales_trends_with_rolling.sql
    mart_reviews_summary.sql

---

Transformation Layers

Staging Layer

• Cleans raw data
• Standardizes column names
• Parses JSON reviews and flattens them to convert nested data into rows

---

Intermediate Layer

• Combines core datasets
• Prepares reusable aggregations

Examples:

• Customer + Orders join
• Product-level revenue
• Review aggregation

---

Mart Layer

Final analytics tables used for business insights:

• Customer segmentation (CLV-based)
• Product performance (sales + reviews)
• Sales trends (time-series)
• Anomaly detection (statistical)
• Rolling averages
• Review summaries

---

Key Metrics

• Customer Lifetime Value (CLV): total spend per customer
• Average Product Rating
• Total Revenue
• Total Orders
• Daily Revenue Trends
• Loyalty Tier Segmentation

---

Customer Segmentation Logic

CLV Range	Segment
< 100	Low
< 500	Medium
< 1000	High
>= 1000	VIP

---

Time-Series Analysis

• Daily aggregation of revenue and orders
• Rolling 7-day average
• Trend monitoring over time

---

Anomaly Detection

Sales anomalies are identified using statistical thresholds:

• Spike → above mean + 2 * std deviation
• Drop → below mean - 2 * std deviation

These advanced analytics help identify unusual spikes or drops in sales

Data Quality Tests

Implemented using dbt:

• not_null
• unique
• relationships
• accepted_values

Example:

• Order must have valid customer_id
• Rating must be between 1 and 5
• Order IDs must be unique

Run tests:

dbt test

---

Running the Project

Run transformations

dbt run

Run tests

dbt test

Generate documentation

dbt docs generate
dbt docs serve

---

Documentation & Lineage

dbt automatically generates:

• Model documentation
• Data lineage graph
• Dependency tracking

End-to-End Pipeline Lineage

The following graph illustrates the complete data flow from raw ingestion in Snowflake to final analytics-ready mart tables.

Product Performance Lineage

This lineage highlights how structured (orders, products) and semi-structured (JSON reviews) data are combined to produce enriched product performance metrics.

---

Snowflake Output Layers

• STAGING → cleaned raw data
• INTERMEDIATE → reusable logic
• MARTS → final analytics tables

What This Project Demonstrates

• End-to-end ELT pipeline design
• Handling structured and semi-structured data
• Incremental ingestion with StreamSets
• Data modeling with dbt (layered architecture)
• Business metric computation (CLV, segmentation)
• Time-series analytics
• Anomaly detection
• Data quality validation
• Documentation and lineage tracking