Build a complete production-ready data warehouse from scratch. Comprehensive capstone project integrating all data warehousing concepts including dimensional modeling, ETL pipelines, BI integration, and more.
Build a complete production-ready data warehouse from scratch. Comprehensive capstone project integrating all data warehousing concepts including dimensional modeling, ETL pipelines, BI integration, and more.
Welcome to the capstone project of the Data Warehousing course! In this comprehensive project, you'll build a complete, production-ready data warehouse for an e-commerce company called "Global Retail Analytics."
This project integrates all the concepts you've learned throughout the course, including dimensional modeling, ETL pipelines, slowly changing dimensions, data quality validation, performance optimization, real-time analytics, and BI tool integration.
Global Retail Analytics is an online retail company that needs a data warehouse to:
Track sales performance across products, customers, and regions
Analyze customer behavior and purchasing patterns
Monitor inventory levels and supply chain efficiency
Support real-time operational dashboards for executives
Enable self-service analytics for business users
Maintain historical data for trend analysis
PostgreSQL as the data warehouse database
Star schema dimensional model
Python-based ETL pipelines
Data quality validation framework
Slowly Changing Dimensions (Type 2) for customer data
Aggregate tables for performance optimization
BI tool integration (Metabase or similar)
Real-time metrics using streaming data
Comprehensive documentation
Learn about Project Architecture.
1
2┌─────────────────────────────────────────────────────────────────────┐
3│ Complete Data Warehouse Architecture │
4├─────────────────────────────────────────────────────────────────────┤
5│ │
6│ DATA SOURCES │
7│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
8│ │ Orders DB │ │ Customer │ │ Product │ │
9│ │ (Postgres) │ │ CRM │ │ Catalog │ │
10│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
11│ │ │ │ │
12│ └──────────────────┴──────────────────┘ │
13│ │ │
14│ ▼ │
15│ ┌─────────────────────────────────────────────────────────────┐ │
16│ │ STAGING LAYER │ │
17│ │ - Raw data extraction │ │
18│ │ - Initial data validation │ │
19│ │ - Change data capture (CDC) │ │
20│ └────────────────────────┬─────────────────────────────────────┘ │
21│ │ │
22│ ▼ │
23│ ┌─────────────────────────────────────────────────────────────┐ │
24│ │ ETL PROCESSING LAYER │ │
25│ │ ┌──────────────────────────────────────────────────────┐ │ │
26│ │ │ Data Transformation & Cleansing │ │ │
27│ │ │ - Type 2 SCD processing │ │ │
28│ │ │ - Business rule application │ │ │
29│ │ │ - Data quality validation │ │ │
30│ │ │ - Surrogate key generation │ │ │
31│ │ └──────────────────────────────────────────────────────┘ │ │
32│ └────────────────────────┬─────────────────────────────────────┘ │
33│ │ │
34│ ▼ │
35│ ┌─────────────────────────────────────────────────────────────┐ │
36│ │ DATA WAREHOUSE (STAR SCHEMA) │ │
37│ │ │ │
38│ │ FACT TABLES: DIMENSION TABLES: │ │
39│ │ - fact_sales - dim_customer │ │
40│ │ - fact_inventory - dim_product │ │
41│ │ - fact_order_line_items - dim_time │ │
42│ │ - dim_location │ │
43│ │ - dim_supplier │ │
44│ │ │ │
45│ │ AGGREGATE TABLES: │ │
46│ │ - agg_sales_daily │ │
47│ │ - agg_sales_monthly │ │
48│ │ - agg_product_performance │ │
49│ └────────────────────────┬─────────────────────────────────────┘ │
50│ │ │
51│ ▼ │
52│ ┌─────────────────────────────────────────────────────────────┐ │
53│ │ SEMANTIC LAYER (dbt metrics) │ │
54│ │ - Business metrics definitions │ │
55│ │ - Consistent calculations │ │
56│ │ - Data access control │ │
57│ └────────────────────────┬─────────────────────────────────────┘ │
58│ │ │
59│ ▼ │
60│ ┌─────────────────────────────────────────────────────────────┐ │
61│ │ BI & ANALYTICS LAYER │ │
62│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
63│ │ │ Metabase │ │ Tableau │ │ Python │ │ │
64│ │ │Dashboard │ │ Reports │ │ Notebooks│ │ │
65│ │ └──────────┘ └──────────┘ └──────────┘ │ │
66│ └─────────────────────────────────────────────────────────────┘ │
67│ │
68│ REAL-TIME STREAM │
69│ ┌──────────────┐ ┌──────────────┐ │
70│ │ Kafka │───▶│ Flink │───▶ Real-time Metrics │
71│ │ Topics │ │ Processing │ │
72│ └──────────────┘ └──────────────┘ │
73│ │
74└─────────────────────────────────────────────────────────────────────┘
75Design and implement a complete star schema for the retail data warehouse.
Learn about Step 1.1: Create Dimension Tables.
1-- Create schema for data warehouse
2CREATE SCHEMA IF NOT EXISTS dw;
3
4-- Time Dimension (pre-populated for 10 years)
5CREATE TABLE dw.dim_time (
6 time_key SERIAL PRIMARY KEY,
7 date_actual DATE NOT NULL UNIQUE,
8 day_of_week INTEGER NOT NULL,
9 day_name VARCHAR(10) NOT NULL,
10 day_of_month INTEGER NOT NULL,
11 day_of_quarter INTEGER NOT NULL,
12 day_of_year INTEGER NOT NULL,
13 week_of_month INTEGER NOT NULL,
14 week_of_year INTEGER NOT NULL,
15 month_actual INTEGER NOT NULL,
16 month_name VARCHAR(10) NOT NULL,
17 quarter_actual INTEGER NOT NULL,
18 quarter_name VARCHAR(2) NOT NULL,
19 year_actual INTEGER NOT NULL,
20 is_weekend BOOLEAN NOT NULL,
21 is_holiday BOOLEAN DEFAULT FALSE,
22 holiday_name VARCHAR(50),
23 fiscal_year INTEGER,
24 fiscal_quarter INTEGER,
25 fiscal_month INTEGER
26);
27
28-- Populate time dimension
29INSERT INTO dw.dim_time (
30 date_actual, day_of_week, day_name, day_of_month, day_of_quarter,
31 day_of_year, week_of_month, week_of_year, month_actual, month_name,
32 quarter_actual, quarter_name, year_actual, is_weekend
33)
34SELECT
35 datum AS date_actual,
36 EXTRACT(ISODOW FROM datum) AS day_of_week,
37 TO_CHAR(datum, 'Day') AS day_name,
38 EXTRACT(DAY FROM datum) AS day_of_month,
39 datum - DATE_TRUNC('quarter', datum)::DATE + 1 AS day_of_quarter,
40 EXTRACT(DOY FROM datum) AS day_of_year,
41 EXTRACT(WEEK FROM datum) - EXTRACT(WEEK FROM DATE_TRUNC('month', datum)) + 1 AS week_of_month,
42 EXTRACT(WEEK FROM datum) AS week_of_year,
43 EXTRACT(MONTH FROM datum) AS month_actual,
44 TO_CHAR(datum, 'Month') AS month_name,
45 EXTRACT(QUARTER FROM datum) AS quarter_actual,
46 'Q' || EXTRACT(QUARTER FROM datum) AS quarter_name,
47 EXTRACT(YEAR FROM datum) AS year_actual,
48 CASE WHEN EXTRACT(ISODOW FROM datum) IN (6, 7) THEN TRUE ELSE FALSE END AS is_weekend
49FROM generate_series('2020-01-01'::DATE, '2030-12-31'::DATE, '1 day'::INTERVAL) AS datum;
50
51CREATE INDEX idx_dim_time_date ON dw.dim_time(date_actual);
52CREATE INDEX idx_dim_time_year_month ON dw.dim_time(year_actual, month_actual);
53
54
55-- Customer Dimension (Type 2 SCD)
56CREATE TABLE dw.dim_customer (
57 customer_key SERIAL PRIMARY KEY,
58 customer_id VARCHAR(50) NOT NULL, -- Business key
59 customer_name VARCHAR(200) NOT NULL,
60 email VARCHAR(200),
61 phone VARCHAR(50),
62 customer_segment VARCHAR(50),
63 customer_tier VARCHAR(20),
64 address_line1 VARCHAR(200),
65 address_line2 VARCHAR(200),
66 city VARCHAR(100),
67 state VARCHAR(50),
68 postal_code VARCHAR(20),
69 country VARCHAR(50),
70 region VARCHAR(50),
71
72 -- Type 2 SCD columns
73 effective_date DATE NOT NULL,
74 end_date DATE,
75 is_current BOOLEAN NOT NULL DEFAULT TRUE,
76 version INTEGER NOT NULL DEFAULT 1,
77
78 -- Audit columns
79 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
80 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
81);
82
83CREATE INDEX idx_dim_customer_id ON dw.dim_customer(customer_id);
84CREATE INDEX idx_dim_customer_current ON dw.dim_customer(customer_id, is_current);
85CREATE INDEX idx_dim_customer_segment ON dw.dim_customer(customer_segment);
86
87
88-- Product Dimension
89CREATE TABLE dw.dim_product (
90 product_key SERIAL PRIMARY KEY,
91 product_id VARCHAR(50) NOT NULL UNIQUE, -- Business key
92 product_name VARCHAR(200) NOT NULL,
93 product_description TEXT,
94 category VARCHAR(100),
95 subcategory VARCHAR(100),
96 brand VARCHAR(100),
97 supplier_id VARCHAR(50),
98 unit_cost DECIMAL(10, 2),
99 unit_price DECIMAL(10, 2),
100 package_type VARCHAR(50),
101 package_size VARCHAR(50),
102 weight_kg DECIMAL(8, 2),
103 is_active BOOLEAN DEFAULT TRUE,
104 discontinued_date DATE,
105
106 -- Audit columns
107 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
108 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
109);
110
111CREATE INDEX idx_dim_product_id ON dw.dim_product(product_id);
112CREATE INDEX idx_dim_product_category ON dw.dim_product(category, subcategory);
113CREATE INDEX idx_dim_product_brand ON dw.dim_product(brand);
114
115
116-- Location Dimension
117CREATE TABLE dw.dim_location (
118 location_key SERIAL PRIMARY KEY,
119 location_id VARCHAR(50) NOT NULL UNIQUE,
120 location_name VARCHAR(200) NOT NULL,
121 location_type VARCHAR(50), -- Warehouse, Store, Distribution Center
122 address_line1 VARCHAR(200),
123 address_line2 VARCHAR(200),
124 city VARCHAR(100),
125 state VARCHAR(50),
126 postal_code VARCHAR(20),
127 country VARCHAR(50),
128 region VARCHAR(50),
129 timezone VARCHAR(50),
130 square_footage INTEGER,
131 opened_date DATE,
132 is_active BOOLEAN DEFAULT TRUE,
133
134 -- Audit columns
135 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
136 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
137);
138
139CREATE INDEX idx_dim_location_type ON dw.dim_location(location_type);
140CREATE INDEX idx_dim_location_region ON dw.dim_location(region);
141
142
143-- Supplier Dimension
144CREATE TABLE dw.dim_supplier (
145 supplier_key SERIAL PRIMARY KEY,
146 supplier_id VARCHAR(50) NOT NULL UNIQUE,
147 supplier_name VARCHAR(200) NOT NULL,
148 contact_name VARCHAR(100),
149 email VARCHAR(200),
150 phone VARCHAR(50),
151 address_line1 VARCHAR(200),
152 city VARCHAR(100),
153 country VARCHAR(50),
154 lead_time_days INTEGER,
155 quality_rating DECIMAL(3, 2), -- 0-5 scale
156 is_active BOOLEAN DEFAULT TRUE,
157
158 -- Audit columns
159 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
160 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
161);
162
163CREATE INDEX idx_dim_supplier_id ON dw.dim_supplier(supplier_id);
164Learn about Step 1.2: Create Fact Tables.
1-- Sales Fact Table
2CREATE TABLE dw.fact_sales (
3 sales_key BIGSERIAL PRIMARY KEY,
4
5 -- Foreign keys to dimensions
6 time_key INTEGER NOT NULL REFERENCES dw.dim_time(time_key),
7 customer_key INTEGER NOT NULL REFERENCES dw.dim_customer(customer_key),
8 product_key INTEGER NOT NULL REFERENCES dw.dim_product(product_key),
9 location_key INTEGER NOT NULL REFERENCES dw.dim_location(location_key),
10
11 -- Degenerate dimensions
12 order_id VARCHAR(50) NOT NULL,
13 order_line_number INTEGER NOT NULL,
14
15 -- Measures
16 quantity INTEGER NOT NULL,
17 unit_price DECIMAL(10, 2) NOT NULL,
18 unit_cost DECIMAL(10, 2) NOT NULL,
19 discount_amount DECIMAL(10, 2) DEFAULT 0,
20 tax_amount DECIMAL(10, 2) DEFAULT 0,
21 shipping_amount DECIMAL(10, 2) DEFAULT 0,
22 total_amount DECIMAL(10, 2) NOT NULL,
23
24 -- Calculated measures (for query performance)
25 gross_profit DECIMAL(10, 2) GENERATED ALWAYS AS (total_amount - (quantity * unit_cost)) STORED,
26 profit_margin DECIMAL(5, 2) GENERATED ALWAYS AS (
27 CASE WHEN total_amount > 0
28 THEN ((total_amount - (quantity * unit_cost)) / total_amount) * 100
29 ELSE 0 END
30 ) STORED,
31
32 -- Order status
33 order_status VARCHAR(50) DEFAULT 'pending',
34
35 -- Audit
36 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
37
38 -- Constraint
39 CONSTRAINT uq_fact_sales_order_line UNIQUE (order_id, order_line_number)
40);
41
42-- Partitioning by time for better performance
43CREATE INDEX idx_fact_sales_time ON dw.fact_sales(time_key);
44CREATE INDEX idx_fact_sales_customer ON dw.fact_sales(customer_key);
45CREATE INDEX idx_fact_sales_product ON dw.fact_sales(product_key);
46CREATE INDEX idx_fact_sales_location ON dw.fact_sales(location_key);
47CREATE INDEX idx_fact_sales_order ON dw.fact_sales(order_id);
48CREATE INDEX idx_fact_sales_status ON dw.fact_sales(order_status);
49
50
51-- Inventory Fact Table (Periodic Snapshot)
52CREATE TABLE dw.fact_inventory (
53 inventory_key BIGSERIAL PRIMARY KEY,
54
55 -- Foreign keys
56 time_key INTEGER NOT NULL REFERENCES dw.dim_time(time_key),
57 product_key INTEGER NOT NULL REFERENCES dw.dim_product(product_key),
58 location_key INTEGER NOT NULL REFERENCES dw.dim_location(location_key),
59
60 -- Measures
61 quantity_on_hand INTEGER NOT NULL,
62 quantity_reserved INTEGER DEFAULT 0,
63 quantity_available INTEGER NOT NULL,
64 reorder_point INTEGER,
65 reorder_quantity INTEGER,
66 quantity_on_order INTEGER DEFAULT 0,
67 days_of_supply INTEGER,
68
69 -- Calculated measures
70 inventory_value DECIMAL(12, 2),
71
72 -- Audit
73 snapshot_date DATE NOT NULL,
74 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
75
76 -- Constraint
77 CONSTRAINT uq_fact_inventory_snapshot UNIQUE (time_key, product_key, location_key)
78);
79
80CREATE INDEX idx_fact_inventory_time ON dw.fact_inventory(time_key);
81CREATE INDEX idx_fact_inventory_product ON dw.fact_inventory(product_key);
82CREATE INDEX idx_fact_inventory_location ON dw.fact_inventory(location_key);
83CREATE INDEX idx_fact_inventory_snapshot ON dw.fact_inventory(snapshot_date);
84Learn about Step 1.3: Create Aggregate Tables.
1-- Daily Sales Aggregates
2CREATE TABLE dw.agg_sales_daily (
3 time_key INTEGER NOT NULL REFERENCES dw.dim_time(time_key),
4 product_key INTEGER REFERENCES dw.dim_product(product_key),
5 customer_key INTEGER REFERENCES dw.dim_customer(customer_key),
6 location_key INTEGER REFERENCES dw.dim_location(location_key),
7
8 -- Aggregated measures
9 order_count INTEGER NOT NULL,
10 customer_count INTEGER NOT NULL,
11 total_quantity INTEGER NOT NULL,
12 total_revenue DECIMAL(12, 2) NOT NULL,
13 total_cost DECIMAL(12, 2) NOT NULL,
14 total_profit DECIMAL(12, 2) NOT NULL,
15 avg_order_value DECIMAL(10, 2) NOT NULL,
16 avg_profit_margin DECIMAL(5, 2),
17
18 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
19
20 PRIMARY KEY (time_key, product_key, customer_key, location_key)
21);
22
23CREATE INDEX idx_agg_sales_daily_time ON dw.agg_sales_daily(time_key);
24CREATE INDEX idx_agg_sales_daily_product ON dw.agg_sales_daily(product_key);
25
26
27-- Monthly Sales Aggregates
28CREATE TABLE dw.agg_sales_monthly (
29 year_month DATE NOT NULL,
30 product_key INTEGER REFERENCES dw.dim_product(product_key),
31 customer_segment VARCHAR(50),
32 region VARCHAR(50),
33
34 -- Aggregated measures
35 order_count INTEGER NOT NULL,
36 customer_count INTEGER NOT NULL,
37 total_quantity INTEGER NOT NULL,
38 total_revenue DECIMAL(12, 2) NOT NULL,
39 total_profit DECIMAL(12, 2) NOT NULL,
40 avg_order_value DECIMAL(10, 2) NOT NULL,
41
42 -- Year-over-year comparison
43 yoy_revenue_growth DECIMAL(5, 2),
44
45 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
46
47 PRIMARY KEY (year_month, product_key, customer_segment, region)
48);
49
50CREATE INDEX idx_agg_sales_monthly_date ON dw.agg_sales_monthly(year_month);
51Implement comprehensive ETL pipelines to extract data from source systems, transform it, and load it into the data warehouse.
Learn about Step 2.1: Staging Layer.
1-- Create staging schema
2CREATE SCHEMA IF NOT EXISTS staging;
3
4-- Staging table for orders
5CREATE TABLE staging.orders (
6 order_id VARCHAR(50),
7 customer_id VARCHAR(50),
8 order_date DATE,
9 order_status VARCHAR(50),
10 total_amount DECIMAL(10, 2),
11
12 -- Metadata
13 source_system VARCHAR(50),
14 extracted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
15 loaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
16);
17
18-- Staging table for customers
19CREATE TABLE staging.customers (
20 customer_id VARCHAR(50),
21 customer_name VARCHAR(200),
22 email VARCHAR(200),
23 phone VARCHAR(50),
24 customer_segment VARCHAR(50),
25 address_line1 VARCHAR(200),
26 city VARCHAR(100),
27 state VARCHAR(50),
28 country VARCHAR(50),
29
30 source_system VARCHAR(50),
31 extracted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
32 loaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
33);
34
35-- Staging table for products
36CREATE TABLE staging.products (
37 product_id VARCHAR(50),
38 product_name VARCHAR(200),
39 category VARCHAR(100),
40 subcategory VARCHAR(100),
41 brand VARCHAR(100),
42 unit_cost DECIMAL(10, 2),
43 unit_price DECIMAL(10, 2),
44
45 source_system VARCHAR(50),
46 extracted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
47 loaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
48);
49Learn about Step 2.2: Python ETL Framework.
1# etl_framework.py - Complete ETL framework
2import psycopg2
3from psycopg2.extras import execute_batch
4from datetime import datetime, timedelta
5import logging
6from typing import List, Dict, Any
7from dataclasses import dataclass
8import pandas as pd
9
10logging.basicConfig(level=logging.INFO)
11logger = logging.getLogger(__name__)
12
13
14@dataclass
15class ETLConfig:
16 """Configuration for ETL pipeline"""
17 source_conn_string: str
18 target_conn_string: str
19 batch_size: int = 1000
20 enable_scd2: bool = True
21 enable_validation: bool = True
22
23
24class DataWarehouseETL:
25 """Complete ETL framework for data warehouse"""
26
27 def __init__(self, config: ETLConfig):
28 self.config = config
29 self.source_conn = psycopg2.connect(config.source_conn_string)
30 self.target_conn = psycopg2.connect(config.target_conn_string)
31 self.stats = {
32 'extracted': 0,
33 'transformed': 0,
34 'loaded': 0,
35 'errors': 0
36 }
37
38 def run_full_etl(self):
39 """Execute complete ETL pipeline"""
40 try:
41 logger.info("Starting full ETL pipeline...")
42
43 # Phase 1: Extract
44 self.extract_all_sources()
45
46 # Phase 2: Transform & Load Dimensions
47 self.load_time_dimension()
48 self.load_customer_dimension_scd2()
49 self.load_product_dimension()
50 self.load_location_dimension()
51 self.load_supplier_dimension()
52
53 # Phase 3: Load Facts
54 self.load_sales_fact()
55 self.load_inventory_fact()
56
57 # Phase 4: Build Aggregates
58 self.build_daily_aggregates()
59 self.build_monthly_aggregates()
60
61 # Phase 5: Data Quality Checks
62 if self.config.enable_validation:
63 self.run_data_quality_checks()
64
65 logger.info(f"ETL completed successfully. Stats: {self.stats}")
66
67 except Exception as e:
68 logger.error(f"ETL pipeline failed: {e}")
69 self.target_conn.rollback()
70 raise
71 finally:
72 self.source_conn.close()
73 self.target_conn.close()
74
75 def extract_all_sources(self):
76 """Extract data from all source systems"""
77 logger.info("Extracting data from source systems...")
78
79 # Extract orders
80 self._extract_to_staging(
81 source_query="SELECT * FROM orders WHERE order_date >= CURRENT_DATE - INTERVAL '7 days'",
82 staging_table="staging.orders"
83 )
84
85 # Extract customers
86 self._extract_to_staging(
87 source_query="SELECT * FROM customers WHERE updated_at >= CURRENT_DATE - INTERVAL '7 days'",
88 staging_table="staging.customers"
89 )
90
91 # Extract products
92 self._extract_to_staging(
93 source_query="SELECT * FROM products",
94 staging_table="staging.products"
95 )
96
97 def _extract_to_staging(self, source_query: str, staging_table: str):
98 """Generic extraction to staging"""
99 with self.source_conn.cursor() as src_cursor:
100 src_cursor.execute(source_query)
101 rows = src_cursor.fetchall()
102 columns = [desc[0] for desc in src_cursor.description]
103
104 if not rows:
105 logger.info(f"No data to extract for {staging_table}")
106 return
107
108 # Clear staging table
109 with self.target_conn.cursor() as tgt_cursor:
110 tgt_cursor.execute(f"TRUNCATE TABLE {staging_table}")
111
112 # Insert into staging
113 placeholders = ','.join(['%s'] * len(columns))
114 insert_sql = f"INSERT INTO {staging_table} ({','.join(columns)}) VALUES ({placeholders})"
115 execute_batch(tgt_cursor, insert_sql, rows, page_size=self.config.batch_size)
116
117 self.target_conn.commit()
118 self.stats['extracted'] += len(rows)
119 logger.info(f"Extracted {len(rows)} rows to {staging_table}")
120
121 def load_customer_dimension_scd2(self):
122 """Load customer dimension with Type 2 SCD"""
123 logger.info("Loading customer dimension with SCD Type 2...")
124
125 with self.target_conn.cursor() as cursor:
126 # Call stored procedure for SCD2 processing
127 cursor.execute("""
128 CALL dw.sp_load_customer_dimension_scd2()
129 """)
130 self.target_conn.commit()
131 logger.info("Customer dimension loaded successfully")
132
133 def load_product_dimension(self):
134 """Load product dimension"""
135 logger.info("Loading product dimension...")
136
137 with self.target_conn.cursor() as cursor:
138 cursor.execute("""
139 INSERT INTO dw.dim_product (
140 product_id, product_name, category, subcategory,
141 brand, unit_cost, unit_price
142 )
143 SELECT
144 product_id, product_name, category, subcategory,
145 brand, unit_cost, unit_price
146 FROM staging.products
147 ON CONFLICT (product_id) DO UPDATE SET
148 product_name = EXCLUDED.product_name,
149 category = EXCLUDED.category,
150 subcategory = EXCLUDED.subcategory,
151 brand = EXCLUDED.brand,
152 unit_cost = EXCLUDED.unit_cost,
153 unit_price = EXCLUDED.unit_price,
154 updated_at = CURRENT_TIMESTAMP
155 """)
156 self.target_conn.commit()
157 logger.info(f"Product dimension loaded: {cursor.rowcount} rows")
158
159 def load_sales_fact(self):
160 """Load sales fact table"""
161 logger.info("Loading sales fact table...")
162
163 with self.target_conn.cursor() as cursor:
164 cursor.execute("""
165 INSERT INTO dw.fact_sales (
166 time_key, customer_key, product_key, location_key,
167 order_id, order_line_number, quantity, unit_price,
168 unit_cost, total_amount, order_status
169 )
170 SELECT
171 t.time_key,
172 c.customer_key,
173 p.product_key,
174 l.location_key,
175 o.order_id,
176 ol.line_number,
177 ol.quantity,
178 ol.unit_price,
179 p.unit_cost,
180 ol.line_total,
181 o.order_status
182 FROM staging.orders o
183 JOIN staging.order_lines ol ON o.order_id = ol.order_id
184 JOIN dw.dim_time t ON o.order_date = t.date_actual
185 JOIN dw.dim_customer c ON o.customer_id = c.customer_id AND c.is_current = TRUE
186 JOIN dw.dim_product p ON ol.product_id = p.product_id
187 JOIN dw.dim_location l ON o.location_id = l.location_id
188 WHERE NOT EXISTS (
189 SELECT 1 FROM dw.fact_sales fs
190 WHERE fs.order_id = o.order_id
191 AND fs.order_line_number = ol.line_number
192 )
193 """)
194 self.target_conn.commit()
195 logger.info(f"Sales fact loaded: {cursor.rowcount} rows")
196 self.stats['loaded'] += cursor.rowcount
197
198 def build_daily_aggregates(self):
199 """Build daily aggregate tables"""
200 logger.info("Building daily aggregates...")
201
202 with self.target_conn.cursor() as cursor:
203 cursor.execute("""
204 INSERT INTO dw.agg_sales_daily (
205 time_key, product_key, customer_key, location_key,
206 order_count, customer_count, total_quantity,
207 total_revenue, total_cost, total_profit,
208 avg_order_value, avg_profit_margin
209 )
210 SELECT
211 fs.time_key,
212 fs.product_key,
213 fs.customer_key,
214 fs.location_key,
215 COUNT(DISTINCT fs.order_id) as order_count,
216 COUNT(DISTINCT fs.customer_key) as customer_count,
217 SUM(fs.quantity) as total_quantity,
218 SUM(fs.total_amount) as total_revenue,
219 SUM(fs.quantity * fs.unit_cost) as total_cost,
220 SUM(fs.gross_profit) as total_profit,
221 AVG(fs.total_amount) as avg_order_value,
222 AVG(fs.profit_margin) as avg_profit_margin
223 FROM dw.fact_sales fs
224 WHERE fs.order_status = 'completed'
225 GROUP BY 1, 2, 3, 4
226 ON CONFLICT (time_key, product_key, customer_key, location_key)
227 DO UPDATE SET
228 order_count = EXCLUDED.order_count,
229 customer_count = EXCLUDED.customer_count,
230 total_quantity = EXCLUDED.total_quantity,
231 total_revenue = EXCLUDED.total_revenue,
232 total_cost = EXCLUDED.total_cost,
233 total_profit = EXCLUDED.total_profit,
234 avg_order_value = EXCLUDED.avg_order_value,
235 avg_profit_margin = EXCLUDED.avg_profit_margin
236 """)
237 self.target_conn.commit()
238 logger.info(f"Daily aggregates built: {cursor.rowcount} rows")
239
240 def run_data_quality_checks(self):
241 """Run comprehensive data quality checks"""
242 logger.info("Running data quality checks...")
243
244 checks = [
245 ("Orphaned Sales Records", """
246 SELECT COUNT(*) FROM dw.fact_sales fs
247 LEFT JOIN dw.dim_customer c ON fs.customer_key = c.customer_key
248 WHERE c.customer_key IS NULL
249 """),
250 ("Negative Quantities", """
251 SELECT COUNT(*) FROM dw.fact_sales
252 WHERE quantity < 0
253 """),
254 ("Future Dates", """
255 SELECT COUNT(*) FROM dw.fact_sales fs
256 JOIN dw.dim_time t ON fs.time_key = t.time_key
257 WHERE t.date_actual > CURRENT_DATE
258 """),
259 ("Duplicate Records", """
260 SELECT COUNT(*) FROM (
261 SELECT order_id, order_line_number, COUNT(*)
262 FROM dw.fact_sales
263 GROUP BY order_id, order_line_number
264 HAVING COUNT(*) > 1
265 ) dup
266 """)
267 ]
268
269 with self.target_conn.cursor() as cursor:
270 for check_name, check_sql in checks:
271 cursor.execute(check_sql)
272 result = cursor.fetchone()[0]
273
274 if result > 0:
275 logger.warning(f"Data Quality Check FAILED: {check_name} - {result} issues found")
276 self.stats['errors'] += result
277 else:
278 logger.info(f"Data Quality Check PASSED: {check_name}")
279
280 def close(self):
281 """Close database connections"""
282 self.source_conn.close()
283 self.target_conn.close()
284
285
286# Usage example
287if __name__ == "__main__":
288 config = ETLConfig(
289 source_conn_string="postgresql://user:pass@localhost:5432/source_db",
290 target_conn_string="postgresql://user:pass@localhost:5432/dw",
291 batch_size=1000,
292 enable_scd2=True,
293 enable_validation=True
294 )
295
296 etl = DataWarehouseETL(config)
297 etl.run_full_etl()
298Learn about Step 2.3: SCD Type 2 Implementation.
1-- Stored procedure for SCD Type 2 customer dimension
2CREATE OR REPLACE PROCEDURE dw.sp_load_customer_dimension_scd2()
3LANGUAGE plpgsql
4AS $$
5DECLARE
6 v_rows_inserted INT := 0;
7 v_rows_updated INT := 0;
8BEGIN
9 -- Step 1: Expire changed records
10 UPDATE dw.dim_customer dc
11 SET
12 end_date = CURRENT_DATE - 1,
13 is_current = FALSE
14 FROM staging.customers sc
15 WHERE dc.customer_id = sc.customer_id
16 AND dc.is_current = TRUE
17 AND (
18 dc.customer_name != sc.customer_name OR
19 dc.email != sc.email OR
20 dc.customer_segment != sc.customer_segment OR
21 dc.city != sc.city OR
22 dc.country != sc.country
23 );
24
25 GET DIAGNOSTICS v_rows_updated = ROW_COUNT;
26
27 -- Step 2: Insert new versions for changed records
28 INSERT INTO dw.dim_customer (
29 customer_id, customer_name, email, phone,
30 customer_segment, address_line1, city, state,
31 postal_code, country, region,
32 effective_date, end_date, is_current, version
33 )
34 SELECT
35 sc.customer_id,
36 sc.customer_name,
37 sc.email,
38 sc.phone,
39 sc.customer_segment,
40 sc.address_line1,
41 sc.city,
42 sc.state,
43 sc.postal_code,
44 sc.country,
45 sc.region,
46 CURRENT_DATE as effective_date,
47 NULL as end_date,
48 TRUE as is_current,
49 COALESCE(MAX(dc.version), 0) + 1 as version
50 FROM staging.customers sc
51 LEFT JOIN dw.dim_customer dc ON sc.customer_id = dc.customer_id
52 WHERE NOT EXISTS (
53 SELECT 1 FROM dw.dim_customer dc2
54 WHERE dc2.customer_id = sc.customer_id
55 AND dc2.is_current = TRUE
56 AND dc2.customer_name = sc.customer_name
57 AND dc2.email = sc.email
58 AND dc2.customer_segment = sc.customer_segment
59 )
60 GROUP BY sc.customer_id, sc.customer_name, sc.email, sc.phone,
61 sc.customer_segment, sc.address_line1, sc.city, sc.state,
62 sc.postal_code, sc.country, sc.region;
63
64 GET DIAGNOSTICS v_rows_inserted = ROW_COUNT;
65
66 RAISE NOTICE 'SCD Type 2 Customer Load Complete: % updated, % inserted',
67 v_rows_updated, v_rows_inserted;
68
69 COMMIT;
70END;
71$$;
72Integrate BI tools and create semantic layer for business users.
Learn about Step 3.1: dbt Metrics Semantic Layer.
1# models/schema.yml - dbt metrics for semantic layer
2version: 2
3
4metrics:
5 - name: total_revenue
6 label: "Total Revenue"
7 model: ref('fact_sales')
8 calculation_method: sum
9 expression: total_amount
10 timestamp: order_date
11 time_grains: [day, week, month, quarter, year]
12 dimensions:
13 - customer_segment
14 - product_category
15 - region
16 filters:
17 - field: order_status
18 operator: '='
19 value: "'completed'"
20
21 - name: gross_profit
22 label: "Gross Profit"
23 model: ref('fact_sales')
24 calculation_method: sum
25 expression: gross_profit
26 timestamp: order_date
27 time_grains: [day, week, month, quarter, year]
28
29 - name: order_count
30 label: "Number of Orders"
31 model: ref('fact_sales')
32 calculation_method: count_distinct
33 expression: order_id
34 timestamp: order_date
35 time_grains: [day, week, month, quarter, year]
36
37 - name: average_order_value
38 label: "Average Order Value"
39 model: ref('fact_sales')
40 calculation_method: average
41 expression: total_amount
42 timestamp: order_date
43 time_grains: [day, week, month, quarter, year]
44
45 - name: customer_lifetime_value
46 label: "Customer Lifetime Value"
47 model: ref('customer_metrics')
48 calculation_method: average
49 expression: lifetime_revenue
50 timestamp: calculation_date
51 time_grains: [month, quarter, year]
52Learn about Step 3.2: Create Business Views.
1-- Create views for common business queries
2CREATE OR REPLACE VIEW dw.v_sales_summary AS
3SELECT
4 t.date_actual as order_date,
5 t.year_actual as year,
6 t.month_name as month,
7 t.quarter_name as quarter,
8 c.customer_segment,
9 c.region,
10 p.category as product_category,
11 p.brand,
12 COUNT(DISTINCT fs.order_id) as order_count,
13 COUNT(DISTINCT fs.customer_key) as customer_count,
14 SUM(fs.quantity) as total_quantity,
15 SUM(fs.total_amount) as total_revenue,
16 SUM(fs.gross_profit) as total_profit,
17 AVG(fs.total_amount) as avg_order_value,
18 AVG(fs.profit_margin) as avg_profit_margin
19FROM dw.fact_sales fs
20JOIN dw.dim_time t ON fs.time_key = t.time_key
21JOIN dw.dim_customer c ON fs.customer_key = c.customer_key
22JOIN dw.dim_product p ON fs.product_key = p.product_key
23WHERE fs.order_status = 'completed'
24GROUP BY 1,2,3,4,5,6,7,8;
25
26
27-- Customer 360 view
28CREATE OR REPLACE VIEW dw.v_customer_360 AS
29SELECT
30 c.customer_id,
31 c.customer_name,
32 c.customer_segment,
33 c.region,
34 COUNT(DISTINCT fs.order_id) as lifetime_orders,
35 SUM(fs.total_amount) as lifetime_revenue,
36 AVG(fs.total_amount) as avg_order_value,
37 MIN(t.date_actual) as first_order_date,
38 MAX(t.date_actual) as last_order_date,
39 CURRENT_DATE - MAX(t.date_actual) as days_since_last_order,
40 COUNT(DISTINCT p.category) as categories_purchased
41FROM dw.dim_customer c
42LEFT JOIN dw.fact_sales fs ON c.customer_key = fs.customer_key
43LEFT JOIN dw.dim_time t ON fs.time_key = t.time_key
44LEFT JOIN dw.dim_product p ON fs.product_key = p.product_key
45WHERE c.is_current = TRUE
46GROUP BY 1,2,3,4;
47
48
49-- Product performance view
50CREATE OR REPLACE VIEW dw.v_product_performance AS
51SELECT
52 p.product_id,
53 p.product_name,
54 p.category,
55 p.subcategory,
56 p.brand,
57 COUNT(DISTINCT fs.order_id) as order_count,
58 SUM(fs.quantity) as units_sold,
59 SUM(fs.total_amount) as total_revenue,
60 SUM(fs.gross_profit) as total_profit,
61 AVG(fs.profit_margin) as avg_profit_margin,
62 SUM(fs.total_amount) / NULLIF(SUM(SUM(fs.total_amount)) OVER (), 0) * 100 as revenue_contribution_pct
63FROM dw.dim_product p
64JOIN dw.fact_sales fs ON p.product_key = fs.product_key
65WHERE fs.order_status = 'completed'
66 AND p.is_active = TRUE
67GROUP BY 1,2,3,4,5;
68Complete the following deliverables to finish the final project:
✓ Complete star schema with all dimension and fact tables
✓ Proper indexing strategy for query performance
✓ Type 2 SCD implementation for customer dimension
✓ Aggregate tables at multiple granularities
✓ Business-friendly views
✓ Staging layer for all source systems
✓ Python ETL framework with error handling
✓ Incremental load strategy
✓ Data quality validation framework
✓ Logging and monitoring
✓ Automated data quality checks
✓ Referential integrity validation
✓ Business rule validation
✓ Data profiling reports
✓ Semantic layer with dbt metrics
✓ BI tool connection (Metabase or Tableau)
✓ Executive dashboard
✓ Operational dashboards
✓ Data model documentation (ERD)
✓ ETL process documentation
✓ Business glossary
✓ User guide for BI tools
✓ Query performance analysis
✓ Index optimization
✓ Materialized views where appropriate
Learn about Testing & Validation Checklist.
[ ] All dimension tables populated correctly
[ ] Fact tables have referential integrity
[ ] SCD Type 2 logic works for customer changes
[ ] Aggregate tables match detail records
[ ] All business views return expected results
[ ] Dashboard queries execute in under 5 seconds
[ ] ETL pipeline completes within time window
[ ] Indexes are being used effectively
[ ] No table scans on large tables
[ ] No orphaned records in fact tables
[ ] No negative quantities or amounts
[ ] No future dates in historical data
[ ] No duplicate records
[ ] All required fields populated
[ ] Revenue numbers match source system
[ ] Customer counts are accurate
[ ] Product inventory balances
[ ] Time dimensions cover required range
Take your data warehouse to the next level with these advanced features:
Implement Kafka streaming for real-time order events
Use Apache Flink for stream processing
Create real-time dashboards with sub-second latency
Implement medallion architecture (Bronze, Silver, Gold)
Use Delta Lake for ACID transactions
Build a lakehouse with Iceberg or Hudi
Build customer churn prediction model
Implement product recommendation engine
Create demand forecasting models
Implement row-level security
Add column-level encryption for PII
Create audit trails for all data access
Automate ETL deployment with GitHub Actions
Implement automated testing
Set up monitoring and alerting
Learn about Submission Guidelines.
Database Scripts
: All DDL and stored procedures
: Complete Python ETL framework
Configuration Files
: dbt models, connection configs
Documentation
Data model ERD
ETL process flow diagrams
Business glossary
User guide
Dashboards
: Screenshots or exported reports
Test Results
: Data quality check results
: Setup instructions and architecture overview
Learn about Grading Rubric.
1
2┌────────────────────────────────────────────────────────────┐
3│ Grading Rubric │
4├────────────────────────────────────────────────────────────┤
5│ │
6│ Component Points Weight │
7│ ────────────────────────────────────────────────────── │
8│ Database Implementation 30 30% │
9│ ETL Pipeline 25 25% │
10│ Data Quality 15 15% │
11│ BI Integration 15 15% │
12│ Documentation 10 10% │
13│ Performance Optimization 5 5% │
14│ ────────────────────────────────────────────────────── │
15│ TOTAL 100 100% │
16│ │
17│ Bonus Challenges +40 (optional) │
18│ │
19│ Grade Scale: │
20│ A: 90-100 points │
21│ B: 80-89 points │
22│ C: 70-79 points │
23│ D: 60-69 points │
24│ F: Below 60 points │
25│ │
26└────────────────────────────────────────────────────────────┘
27Submit your project as a GitHub repository with the following structure:
1
2data-warehouse-project/
3├── README.md
4├── docs/
5│ ├── architecture.md
6│ ├── data-model.pdf (ERD)
7│ ├── etl-process.md
8│ ├── business-glossary.md
9│ └── user-guide.md
10├── database/
11│ ├── ddl/
12│ │ ├── 01-create-schemas.sql
13│ │ ├── 02-create-dimensions.sql
14│ │ ├── 03-create-facts.sql
15│ │ └── 04-create-aggregates.sql
16│ ├── stored-procedures/
17│ │ └── sp_load_customer_scd2.sql
18│ └── views/
19│ └── business-views.sql
20├── etl/
21│ ├── __init__.py
22│ ├── config.py
23│ ├── etl_framework.py
24│ ├── extractors/
25│ ├── transformers/
26│ └── loaders/
27├── dbt/
28│ ├── models/
29│ ├── macros/
30│ └── dbt_project.yml
31├── dashboards/
32│ ├── executive-dashboard.png
33│ └── operational-dashboard.png
34├── tests/
35│ ├── test_etl.py
36│ └── test_data_quality.py
37├── config/
38│ ├── database.yml
39│ └── metabase-setup.json
40└── requirements.txt
41Your data warehouse project is considered successful if it meets the following criteria:
✓ All tables created with proper schema design
✓ ETL pipeline runs without errors
✓ Data quality checks pass with 95%+ accuracy
✓ Dashboard queries execute in under 5 seconds
✓ No orphaned records or referential integrity violations
✓ SCD Type 2 correctly tracks historical changes
✓ Revenue figures match source system within 1%
✓ Business users can answer key questions without IT help
✓ Dashboards provide actionable insights
✓ Historical trend analysis is accurate
✓ Self-service analytics is enabled through semantic layer
✓ Code is well-documented and maintainable
✓ Architecture is scalable and extensible
✓ Documentation is clear and comprehensive
✓ Project follows industry best practices
✓ Security and access controls are implemented
Congratulations on reaching the final project of the Data Warehousing course! This comprehensive capstone project brings together everything you've learned: dimensional modeling, ETL pipelines, data quality, performance optimization, BI integration, and more.
By completing this project, you will have built a production-ready data warehouse that demonstrates your mastery of data warehousing concepts and your ability to deliver real business value through data analytics.
Take your time with each phase, test thoroughly, and don't hesitate to refer back to previous lessons for guidance. Good luck!
Learn about 🎉 Final Tips for Success.
Your progress is saved in this browser, on this device only.