Skip to content
BranchesTags

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

1 Commit
README.MD
README.MD
 
 

Repository files navigation

Retail Sales Data Warehouse

An end-to-end ETL data engineering project that ingests retail sales data from multiple CSV sources, transforms it using Pandas, and loads it into a PostgreSQL data warehouse following a Star Schema dimensional model — orchestrated with Apache Airflow.

Table of Contents

Business Problem

A retail company collects transactional data from multiple departments — sales orders, product catalogs, customer records, and store locations — all stored in separate CSV exports with inconsistent formats and no unified reporting model.

Without a centralized data warehouse, analysts spend hours manually combining spreadsheets every time a business report is needed. This pipeline automates the full ETL process: extracting raw files, cleaning and validating the data with Pandas, loading it into PostgreSQL using SQLAlchemy, and structuring it as a Star Schema — ready for analytical queries. Apache Airflow orchestrates the entire pipeline on a daily schedule.

Business Questions Answered

  • What is the total revenue by product category per month?
  • Which stores generate the highest sales volume?
  • Who are the top 10 customers by lifetime value?
  • Which products have declining sales trends over time?
  • What is the average order value by region?
  • How do sales figures compare year-over-year?
  • Which day of the week drives the most transactions?

Architecture

Raw CSV Files (Sales / Products / Customers / Stores)
        ↓
  Extract Stage         ← Read CSVs with Pandas
        ↓
  Transform Stage       ← Clean, validate, normalize, type-cast
        ↓
  Load Stage            ← Write to PostgreSQL via SQLAlchemy
        ↓
  Star Schema DW        ← fact_sales + dimension tables
        ↓
  Analytics Layer       ← SQL views & window function queries
        ↓
  Airflow DAG           ← Daily orchestration & scheduling

Star Schema Design

The warehouse follows a Star Schema with one central fact table and four dimension tables.

Fact Table

fact_sales

Column Type Description
sale_id SERIAL PRIMARY KEY Surrogate key
date_key INT (FK) Foreign key → dim_date
product_key INT (FK) Foreign key → dim_product
customer_key INT (FK) Foreign key → dim_customer
store_key INT (FK) Foreign key → dim_store
quantity INT Units sold
unit_price DECIMAL Price per unit
discount DECIMAL Discount applied
total_amount DECIMAL Calculated revenue
created_at TIMESTAMP Record load time

Dimension Tables

dim_date

Column Type Description
date_key INT PRIMARY KEY YYYYMMDD integer key
full_date DATE Actual date
day INT Day of month
month INT Month number
month_name VARCHAR Month name
quarter INT Quarter (1–4)
year INT Year
day_of_week VARCHAR Monday, Tuesday…
is_weekend BOOLEAN Weekend flag

dim_product

Column Type Description
product_key SERIAL PRIMARY KEY Surrogate key
product_id VARCHAR Source system ID
product_name VARCHAR Product name
category VARCHAR Product category
sub_category VARCHAR Sub-category
brand VARCHAR Brand name
unit_cost DECIMAL Cost of goods

dim_customer

Column Type Description
customer_key SERIAL PRIMARY KEY Surrogate key
customer_id VARCHAR Source system ID
full_name VARCHAR Customer full name
email VARCHAR Email address
city VARCHAR City
country VARCHAR Country
segment VARCHAR Customer segment
registration_date DATE Sign-up date

dim_store

Column Type Description
store_key SERIAL PRIMARY KEY Surrogate key
store_id VARCHAR Source system ID
store_name VARCHAR Store name
city VARCHAR City
region VARCHAR Region
country VARCHAR Country
open_date DATE Store opening date

Technologies

Category Tool / Library
Language Python 3.11+
Database PostgreSQL 15
ORM / DB Driver SQLAlchemy, psycopg2
Data Processing Pandas
Pipeline Orchestration Apache Airflow 2.x
Data Modeling Star Schema (Dimensional Modeling)
Pipeline Pattern ETL
Version Control Git + GitHub
CI/CD GitHub Actions
Containerization (upcoming) Docker
Visualization (upcoming) Power BI / Metabase

Data Sources

Simulated retail data (CSV files generated or sourced from Kaggle):

File Description
sales.csv Raw transactional sales records
products.csv Product catalog with categories
customers.csv Customer master data
stores.csv Store location and region data

All files land in storage/raw/ before pipeline execution.

Project Structure

retail-sales-warehouse/
│
├── .github/
│   └── workflows/
│       └── ci.yml                    # GitHub Actions CI pipeline
│
├── dags/
│   └── retail_etl_dag.py             # Airflow DAG definition (daily schedule)
│
├── etl/
│   ├── extract/
│   │   ├── __init__.py
│   │   ├── sales_extractor.py        # Read sales.csv → DataFrame
│   │   ├── products_extractor.py     # Read products.csv → DataFrame
│   │   ├── customers_extractor.py    # Read customers.csv → DataFrame
│   │   └── stores_extractor.py      # Read stores.csv → DataFrame
│   │
│   ├── transform/
│   │   ├── __init__.py
│   │   ├── clean.py                  # Null handling, deduplication, type casting
│   │   ├── normalize.py              # Column renaming, standardization
│   │   ├── dim_builder.py            # Build dimension DataFrames
│   │   └── fact_builder.py           # Join dims → build fact_sales DataFrame
│   │
│   └── load/
│       ├── __init__.py
│       ├── models.py                 # SQLAlchemy ORM table models
│       └── loader.py                 # Write DataFrames → PostgreSQL
│
├── db/
│   └── migrations/
│       ├── create_dimensions.sql     # DDL for all dim tables
│       ├── create_fact.sql           # DDL for fact_sales table
│       └── create_views.sql          # Analytical SQL views
│
├── analytics/
│   └── queries/
│       ├── revenue_by_category.sql   # Monthly revenue by product category
│       ├── top_customers.sql         # Top 10 customers by spend
│       ├── store_performance.sql     # Revenue per store per region
│       └── yoy_comparison.sql        # Year-over-year sales comparison
│
├── config/
│   ├── settings.py                   # DB credentials, file paths, env vars
│   └── schemas/
│       ├── sales_schema.py           # Pandas dtype + column definitions
│       ├── products_schema.py
│       ├── customers_schema.py
│       └── stores_schema.py
│
├── storage/
│   └── raw/                          # Source CSV files land here
│       ├── sales.csv
│       ├── products.csv
│       ├── customers.csv
│       └── stores.csv
│
├── tests/
│   ├── test_extract.py               # Unit tests for extractors
│   ├── test_transform.py             # Unit tests for transformations
│   └── test_load.py                  # Integration tests for DB writes
│
├── logs/                             # Pipeline execution logs
├── main.py                           # Manual pipeline entry point
├── requirements.txt                  # Python dependencies
├── docker-compose.yml                # PostgreSQL + Airflow local setup
├── .env.example                      # Environment variable template
├── .gitignore
└── README.md

Pipeline Stages

1. Extract

Each source CSV is read into a Pandas DataFrame by a dedicated extractor module. Basic file validation (file exists, non-empty, expected columns present) is performed at this stage.

2. Transform

  • Null value handling (drop or fill based on column rules)
  • Duplicate record removal
  • Data type enforcement (dates, decimals, integers)
  • Column renaming to match warehouse schema
  • Surrogate key generation for dimension tables
  • Date dimension population (derived columns: day, month, quarter, year, weekday, weekend flag)
  • Fact table construction by joining dimension DataFrames on business keys

3. Load

  • SQLAlchemy ORM models define table structure
  • Dimensions loaded first (referential integrity)
  • Fact table loaded last
  • upsert logic to handle reruns without duplicates
  • Row count logged after each table write

Airflow DAG

File: dags/retail_etl_dag.py

Schedule: Daily at 06:00 UTC

DAG Tasks:

start
  └── extract_sales
  └── extract_products
  └── extract_customers
  └── extract_stores
        └── transform_and_build_dims
              └── load_dimensions
                    └── build_fact_table
                          └── load_fact
                                └── run_data_quality_checks
                                      └── end

Each task is a Python operator calling the corresponding ETL module. Task failures trigger email alerts and halt downstream tasks.

Data Quality Checks

Run automatically after each pipeline load:

Check Description
Null check No nulls in primary key and foreign key columns
Duplicate check No duplicate sale_id in fact_sales
Referential integrity Every date_key, product_key, customer_key, store_key in fact resolves to a dim row
Row count validation Loaded row count matches source CSV row count (after dedup)
Date range check No future-dated transactions
Revenue sanity total_amount = unit_price × quantity − discount (within tolerance)

How to Run

Prerequisites

  • Python 3.11+
  • Docker + Docker Compose (for PostgreSQL + Airflow)
  • Git

Setup

# Clone the repo
git clone https://github.com/your-username/retail-sales-warehouse.git
cd retail-sales-warehouse

# Copy environment config
cp .env.example .env
# Edit .env with your DB credentials

# Install dependencies
pip install -r requirements.txt

# Start PostgreSQL + Airflow with Docker
docker-compose up -d

# Run DB migrations
psql -U postgres -d retail_dw -f db/migrations/create_dimensions.sql
psql -U postgres -d retail_dw -f db/migrations/create_fact.sql
psql -U postgres -d retail_dw -f db/migrations/create_views.sql

# Run pipeline manually (without Airflow)
python main.py

# Or open Airflow UI at http://localhost:8080 and trigger the DAG

Development Progress

✅ Completed

Component Status Details
Project Structure ✅ Done Modular ETL folder layout
Extract Layer ✅ Done 4 CSV extractor modules
Transform Layer ✅ Done Cleaning, normalization, dim/fact builders
SQLAlchemy Models ✅ Done ORM models for all 5 tables
PostgreSQL Migrations ✅ Done DDL for all dimensions and fact table
Load Layer ✅ Done Upsert logic with row count logging
Airflow DAG ✅ Done Daily scheduled DAG with task dependencies
Config Management ✅ Done Centralized settings.py + .env support
CI/CD ✅ Done GitHub Actions workflow

🔄 In Progress

  • Data quality check automation post-load
  • Unit tests for transform module

⏳ Upcoming

  • Docker deployment (full stack)
  • Analytical SQL views
  • Power BI / Metabase dashboard
  • Incremental loading (load only new records)

Future Improvements

Version Feature
v2 Incremental / delta loading
v3 Data quality framework (Great Expectations)
v4 Migrate orchestration to Airflow 2 TaskFlow API
v5 Add Spark for large-scale transform
v6 Cloud deployment (AWS RDS + MWAA)
v7 Power BI dashboard connected to PostgreSQL

Learning Outcomes

This project demonstrates practical knowledge of:

  • ETL Pipeline Design
  • Dimensional Modeling (Star Schema)
  • SQL & PostgreSQL
  • Python data engineering (Pandas, SQLAlchemy)
  • Apache Airflow DAG authoring and scheduling
  • Data Quality Validation
  • Pipeline Orchestration
  • CI/CD with GitHub Actions

About

No description, website, or topics provided.

Resources

Readme
Activity

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors