🏢 Medallion Architecture DWH on PostgreSQL

Modern Data Warehouse Implementation with PostgreSQL & Medallion Architecture

📖 Overview

This project implements an enterprise-grade data warehouse using PostgreSQL and the Medallion Architecture pattern (Bronze → Silver → Gold). It demonstrates modern data engineering practices with automated ETL pipelines, comprehensive data quality checks, and dimensional modeling.

🎯 Business Value

360° Customer View: Unified customer data from multiple sources
Product Intelligence: Comprehensive product performance tracking
Sales Analytics: Real-time sales trends and revenue insights
Data Quality: Built-in validation and monitoring
Scalability: Docker-based deployment for easy scaling

✨ Key Features

🏗️ Architecture

✅ Medallion Architecture (Bronze, Silver, Gold)
✅ Star Schema dimensional modeling
✅ Multi-source data integration
✅ Scalable Docker deployment

🔧 Engineering

✅ Automated ETL with stored procedures
✅ Data Quality validation checks
✅ Version Control for all scripts
✅ Documentation & data catalog

🚀 Quick Start

Get your data warehouse running in under 10 minutes:

Step 1: Clone the Repository

git clone https://github.com/yourusername/enterprise-data-warehouse.git
cd enterprise-data-warehouse

Step 2: Start PostgreSQL Container

docker-compose up -d

Wait for the container to be ready (about 10-15 seconds):

docker-compose ps

Step 3: Initialize Database & Create Schemas

docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -f /scripts/init_database.sql

Step 4: Create Bronze Layer

# Create tables
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -f /scripts/bronze/ddl_bronze.sql

# Create stored procedure
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -f /scripts/bronze/proc_load_bronze.sql

Step 5: Create Silver Layer

# Create tables
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -f /scripts/silver/ddl_silver.sql

# Create stored procedure
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -f /scripts/silver/proc_load_silver.sql

Step 6: Create Gold Layer

docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -f /scripts/gold/ddl_gold.sql

Step 7: Load Data

# Load Bronze layer (from CSV files)
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -c "CALL bronze.load_bronze();"

# Load Silver layer (cleaned and validated)
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -c "CALL silver.load_silver();"

Step 8: Verify Installation

# Check data counts
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -c "SELECT COUNT(*) as customers FROM gold.dim_customers;"
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -c "SELECT COUNT(*) as products FROM gold.dim_products;"
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -c "SELECT COUNT(*) as sales FROM gold.fact_sales;"

Step 9: Run Quality Checks (Optional)

docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -f /tests/quality_checks_silver.sql
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -f /tests/quality_checks_gold.sql

That's it! Your data warehouse is ready. 🎉

🏛️ Architecture

Medallion Architecture Pattern

┌─────────────┐    ┌──────────────┐    ┌──────────────┐    ┌─────────────┐
│   Sources   │───▶│ Bronze Layer │───▶│ Silver Layer │───▶│ Gold Layer  │
│    (CSV)    │    │  (Raw Data)  │    │   (Cleaned)  │    │ (Analytics) │
└─────────────┘    └──────────────┘    └──────────────┘    └─────────────┘
                          │                    │                    │
                      Tables               Tables               Views
                    No Transform          Validated          Star Schema

Data Layers

Layer	Purpose	Technology	Transformations
🥉 Bronze	Raw data ingestion	PostgreSQL Tables	None (as-is from source)
🥈 Silver	Cleaned & validated	PostgreSQL Tables	Deduplication, Standardization, Validation
🥇 Gold	Business-ready analytics	PostgreSQL Views	Dimensional modeling, Aggregations, Business logic

Star Schema (Gold Layer)

📁 Project Structure

Medallion-Architecture-DWH-on-PostgreSQL/
│
├── 📂 datasets/              # Source data files
│   ├── source_crm/          # Customer Relationship Management data
│   │   ├── cust_info.csv
│   │   ├── prd_info.csv
│   │   └── sales_details.csv
│   └── source_erp/          # Enterprise Resource Planning data
│       ├── CUST_AZ12.csv
│       ├── LOC_A101.csv
│       └── PX_CAT_G1V2.csv
│
├── 📂 scripts/              # SQL scripts organized by layer
│   ├── init_database.sql   # Database initialization
│   ├── bronze/             # Bronze layer (raw data)
│   │   ├── ddl_bronze.sql
│   │   └── proc_load_bronze.sql
│   ├── silver/             # Silver layer (cleaned data)
│   │   ├── ddl_silver.sql
│   │   └── proc_load_silver.sql
│   └── gold/               # Gold layer (analytics)
│       └── ddl_gold.sql
│
├── 📂 tests/                # Data quality validation
│   ├── quality_checks_silver.sql
│   └── quality_checks_gold.sql
│
├── 📂 docs/                 # Comprehensive documentation
│   ├── data_catalog.md     # Schema documentation
│   ├── naming_conventions.md
│   └── diagrams/           # Visual architecture diagrams
│
├── 🐳 docker-compose.yml    # Container orchestration
└── 📖 README.md            # This file

💻 Usage

Connecting to the Database

# Using Docker container (psql inside container)
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse

# Using external psql client
psql -h localhost -p 5432 -U dwh_user -d datawarehouse
# Password: dwh_password

Example Queries

📊 Top 10 Customers by Revenue

SELECT 
    c.first_name || ' ' || c.last_name as customer_name,
    c.country,
    COUNT(DISTINCT f.order_number) as total_orders,
    SUM(f.sales_amount) as lifetime_value
FROM gold.fact_sales f
JOIN gold.dim_customers c ON f.customer_key = c.customer_key
GROUP BY c.customer_key, customer_name, c.country
ORDER BY lifetime_value DESC
LIMIT 10;

📈 Monthly Sales Trend

SELECT 
    DATE_TRUNC('month', order_date) as month,
    COUNT(DISTINCT order_number) as orders,
    SUM(sales_amount) as revenue,
    AVG(sales_amount) as avg_order_value
FROM gold.fact_sales
WHERE order_date >= CURRENT_DATE - INTERVAL '12 months'
GROUP BY DATE_TRUNC('month', order_date)
ORDER BY month DESC;

🏆 Product Category Performance

SELECT 
    p.category,
    p.subcategory,
    COUNT(DISTINCT p.product_key) as products,
    SUM(f.quantity) as units_sold,
    SUM(f.sales_amount) as revenue
FROM gold.dim_products p
LEFT JOIN gold.fact_sales f ON p.product_key = f.product_key
GROUP BY p.category, p.subcategory
ORDER BY revenue DESC;

🎯 Customer Lifetime Value Analysis

SELECT 
    c.customer_key,
    c.first_name || ' ' || c.last_name as customer_name,
    c.country,
    c.marital_status,
    COUNT(DISTINCT f.order_number) as total_orders,
    SUM(f.quantity) as total_items_purchased,
    SUM(f.sales_amount) as lifetime_value,
    AVG(f.sales_amount) as avg_order_value,
    MIN(f.order_date) as first_purchase_date,
    MAX(f.order_date) as last_purchase_date,
    MAX(f.order_date) - MIN(f.order_date) as customer_tenure_days
FROM gold.dim_customers c
JOIN gold.fact_sales f ON c.customer_key = f.customer_key
GROUP BY c.customer_key, customer_name, c.country, c.marital_status
HAVING COUNT(DISTINCT f.order_number) > 5
ORDER BY lifetime_value DESC
LIMIT 20;

📦 Product Performance with Profitability

SELECT 
    p.product_name,
    p.category,
    p.subcategory,
    p.product_line,
    COUNT(DISTINCT f.order_number) as total_orders,
    SUM(f.quantity) as units_sold,
    SUM(f.sales_amount) as total_revenue,
    p.cost * SUM(f.quantity) as total_cost,
    SUM(f.sales_amount) - (p.cost * SUM(f.quantity)) as gross_profit,
    ROUND(
        ((SUM(f.sales_amount) - (p.cost * SUM(f.quantity))) / 
         NULLIF(SUM(f.sales_amount), 0) * 100)::NUMERIC, 2
    ) as profit_margin_pct
FROM gold.dim_products p
LEFT JOIN gold.fact_sales f ON p.product_key = f.product_key
GROUP BY p.product_key, p.product_name, p.category, p.subcategory, p.product_line, p.cost
HAVING SUM(f.quantity) IS NOT NULL
ORDER BY gross_profit DESC
LIMIT 15;

📚 More advanced query examples

Customer Segmentation by Purchase Frequency

SELECT 
    CASE 
        WHEN order_count >= 10 THEN 'VIP'
        WHEN order_count >= 5 THEN 'Loyal'
        WHEN order_count >= 2 THEN 'Regular'
        ELSE 'New'
    END as customer_segment,
    COUNT(*) as customers,
    AVG(lifetime_value) as avg_lifetime_value,
    SUM(lifetime_value) as total_revenue
FROM (
    SELECT 
        c.customer_key,
        COUNT(DISTINCT f.order_number) as order_count,
        SUM(f.sales_amount) as lifetime_value
    FROM gold.dim_customers c
    JOIN gold.fact_sales f ON c.customer_key = f.customer_key
    GROUP BY c.customer_key
) segments
GROUP BY customer_segment
ORDER BY avg_lifetime_value DESC;

Sales by Country and Product Category

SELECT 
    c.country,
    p.category,
    COUNT(DISTINCT f.order_number) as orders,
    SUM(f.quantity) as units_sold,
    SUM(f.sales_amount) as revenue,
    AVG(f.sales_amount) as avg_order_value
FROM gold.fact_sales f
JOIN gold.dim_customers c ON f.customer_key = c.customer_key
JOIN gold.dim_products p ON f.product_key = p.product_key
GROUP BY c.country, p.category
ORDER BY revenue DESC;

🔍 Data Quality

Automated Quality Checks

The warehouse includes comprehensive data quality validation:

Run All Quality Checks

# Silver layer checks
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -f /tests/quality_checks_silver.sql

# Gold layer checks
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -f /tests/quality_checks_gold.sql

Quality Dimensions

✅ Completeness: No null values in required fields
✅ Uniqueness: No duplicate primary keys
✅ Accuracy: Correct calculations (e.g., sales = quantity × price)
✅ Consistency: Valid date ranges and referential integrity
✅ Timeliness: Data freshness monitoring
✅ Validity: Standardized values (e.g., gender, marital status)

Quality Check Examples

View quality check queries

-- Check for duplicate customer keys
SELECT 
    customer_key,
    COUNT(*) AS duplicate_count
FROM gold.dim_customers
GROUP BY customer_key
HAVING COUNT(*) > 1;

-- Verify sales calculation accuracy
SELECT DISTINCT 
    sls_sales,
    sls_quantity,
    sls_price 
FROM silver.crm_sales_details
WHERE sls_sales != sls_quantity * sls_price
   OR sls_sales IS NULL 
   OR sls_quantity IS NULL 
   OR sls_price IS NULL;

-- Check referential integrity
SELECT COUNT(*) as orphaned_records
FROM gold.fact_sales f
LEFT JOIN gold.dim_customers c ON c.customer_key = f.customer_key
LEFT JOIN gold.dim_products p ON p.product_key = f.product_key
WHERE p.product_key IS NULL OR c.customer_key IS NULL;

🔄 Data Pipeline Operations

Complete Data Reload

# Step 1: Reload Bronze layer from source files
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -c "CALL bronze.load_bronze();"

# Step 2: Process and load Silver layer
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -c "CALL silver.load_silver();"

# Step 3: Verify Gold layer views (automatic, no load needed)
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -c "SELECT COUNT(*) FROM gold.fact_sales;"

Incremental Updates

Currently configured for full load only. Click for incremental strategy.

To implement incremental loads:

Add dwh_load_date column to track loading timestamps
Modify stored procedures to process only new/changed records
Implement change data capture (CDC) logic
Add watermark tables to track last processed timestamps

Example incremental load pattern:

-- Track last load time
CREATE TABLE dwh.load_watermarks (
    table_name VARCHAR(100),
    last_load_timestamp TIMESTAMP
);

-- Incremental insert example
INSERT INTO silver.crm_sales_details
SELECT * FROM bronze.crm_sales_details
WHERE dwh_create_date > (
    SELECT last_load_timestamp 
    FROM dwh.load_watermarks 
    WHERE table_name = 'crm_sales_details'
);

🛠️ Container Management

Basic Operations

# Start the database
docker-compose up -d

# Stop the database
docker-compose down

# Restart the database
docker-compose restart

# View logs
docker-compose logs -f postgres

# Check container status
docker-compose ps

# View resource usage
docker stats dwh_postgres

Database Backup & Restore

Create Backup

# Create backup file
docker exec dwh_postgres pg_dump -U dwh_user datawarehouse > backup_$(date +%Y%m%d_%H%M%S).sql

# Create compressed backup
docker exec dwh_postgres pg_dump -U dwh_user datawarehouse | gzip > backup_$(date +%Y%m%d_%H%M%S).sql.gz

Restore from Backup

# Restore from SQL file
cat backup_20241208_120000.sql | docker exec -i dwh_postgres psql -U dwh_user -d datawarehouse

# Restore from compressed backup
gunzip -c backup_20241208_120000.sql.gz | docker exec -i dwh_postgres psql -U dwh_user -d datawarehouse

Cleanup

# Stop and remove container (keeps volumes)
docker-compose down

# Stop and remove everything including data
docker-compose down -v

# Remove old/unused Docker resources
docker system prune -a

🔧 Configuration

Database Connection Details

Host:     localhost
Port:     5432
Database: datawarehouse
User:     dwh_user
Password: dwh_password

Environment Variables

You can customize these in docker-compose.yml:

environment:
  POSTGRES_USER: dwh_user
  POSTGRES_PASSWORD: dwh_password
  POSTGRES_DB: datawarehouse

📊 Performance Optimization

Create Indexes

Improve query performance by adding indexes:

-- Indexes on fact table
CREATE INDEX idx_sales_order_date ON gold.fact_sales(order_date);
CREATE INDEX idx_sales_customer_key ON gold.fact_sales(customer_key);
CREATE INDEX idx_sales_product_key ON gold.fact_sales(product_key);

-- Indexes on dimensions
CREATE INDEX idx_customers_country ON gold.dim_customers(country);
CREATE INDEX idx_products_category ON gold.dim_products(category);

Materialized Views

For frequently accessed aggregations:

-- Create materialized view
CREATE MATERIALIZED VIEW gold.mv_monthly_sales AS
SELECT 
    DATE_TRUNC('month', order_date) as month,
    COUNT(DISTINCT order_number) as orders,
    SUM(sales_amount) as revenue
FROM gold.fact_sales
GROUP BY DATE_TRUNC('month', order_date);

-- Refresh when needed
REFRESH MATERIALIZED VIEW gold.mv_monthly_sales;

Query Performance Analysis

-- Enable query timing
\timing on

-- Analyze query execution plan
EXPLAIN ANALYZE 
SELECT * FROM gold.fact_sales 
WHERE order_date >= '2024-01-01';

-- View slow queries
SELECT 
    query,
    calls,
    total_time,
    mean_time
FROM pg_stat_statements
ORDER BY total_time DESC
LIMIT 10;

🧪 Testing

Data Validation Tests

# Run Silver layer quality checks
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -f /tests/quality_checks_silver.sql

# Run Gold layer quality checks
docker exec -it dwh_postgres psql -U dwh_user -d datawarehouse -f /tests/quality_checks_gold.sql

Manual Verification Queries

-- Check record counts
SELECT 
    'dim_customers' as table_name, 
    COUNT(*) as records 
FROM gold.dim_customers
UNION ALL
SELECT 
    'dim_products', 
    COUNT(*) 
FROM gold.dim_products
UNION ALL
SELECT 
    'fact_sales', 
    COUNT(*) 
FROM gold.fact_sales;

-- Verify date ranges
SELECT 
    MIN(order_date) as earliest_order,
    MAX(order_date) as latest_order,
    MAX(order_date) - MIN(order_date) as date_range_days
FROM gold.fact_sales;

-- Check for data quality issues
SELECT 
    COUNT(*) FILTER (WHERE customer_key IS NULL) as null_customers,
    COUNT(*) FILTER (WHERE product_key IS NULL) as null_products,
    COUNT(*) FILTER (WHERE sales_amount IS NULL) as null_sales,
    COUNT(*) FILTER (WHERE sales_amount <= 0) as invalid_sales
FROM gold.fact_sales;

📈 Features

✅ Medallion architecture (Bronze, Silver, Gold)
✅ Star schema dimensional model
✅ Automated ETL with stored procedures
✅ Data quality validation
✅ Docker containerization
✅ Comprehensive documentation

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

Name		Name	Last commit message	Last commit date
datasets		datasets
docs		docs
scripts		scripts
tests		tests
LICENSE		LICENSE
README.md		README.md
docker-compose.yml		docker-compose.yml

Latest commit

History

Folders and files

Repository files navigation