Data Warehouse

Data Warehouse Design

Star Schema Basics

Fact Table Design

-- Star schema with sales fact table
CREATE TABLE fact_sales (
  sales_id BIGINT PRIMARY KEY,
  date_id INT NOT NULL,
  customer_id INT NOT NULL,
  product_id INT NOT NULL,
  store_id INT NOT NULL,
  quantity INT NOT NULL,
  unit_price DECIMAL(10, 2),
  sale_amount DECIMAL(12, 2),
  discount_amount DECIMAL(12, 2),
  net_sales DECIMAL(12, 2),
  tax_amount DECIMAL(12, 2),
  total_sale DECIMAL(12, 2),

  -- Foreign keys
  FOREIGN KEY (date_id) REFERENCES dim_date(date_id),
  FOREIGN KEY (customer_id) REFERENCES dim_customer(customer_id),
  FOREIGN KEY (product_id) REFERENCES dim_product(product_id),
  FOREIGN KEY (store_id) REFERENCES dim_store(store_id)
);

-- Create indexes on foreign keys for query performance
CREATE INDEX idx_fact_sales_date ON fact_sales(date_id);
CREATE INDEX idx_fact_sales_customer ON fact_sales(customer_id);
CREATE INDEX idx_fact_sales_product ON fact_sales(product_id);
CREATE INDEX idx_fact_sales_store ON fact_sales(store_id);

Dimension Table Design

-- Date dimension (conformed dimension - used across multiple facts)
CREATE TABLE dim_date (
  date_id INT PRIMARY KEY,
  full_date DATE UNIQUE,
  day_of_week INT,
  day_of_week_name VARCHAR(10),
  day_of_month INT,
  week_of_year INT,
  month_number INT,
  month_name VARCHAR(12),
  quarter INT,
  year INT,
  fiscal_quarter INT,
  fiscal_year INT,
  is_holiday BOOLEAN,
  is_weekend BOOLEAN,
  is_weekday BOOLEAN
);

-- Customer dimension
CREATE TABLE dim_customer (
  customer_id INT PRIMARY KEY,
  customer_code VARCHAR(20),
  first_name VARCHAR(50),
  last_name VARCHAR(50),
  email VARCHAR(100),
  phone VARCHAR(20),
  gender VARCHAR(10),
  birth_date DATE,
  -- Address hierarchy
  street_address VARCHAR(100),
  city VARCHAR(50),
  state_province VARCHAR(50),
  postal_code VARCHAR(10),
  country VARCHAR(50),
  region VARCHAR(50),
  -- Customer segment
  customer_segment VARCHAR(50),
  customer_lifetime_value DECIMAL(12, 2),
  -- Slowly changing dimension columns
  effective_date DATE,
  end_date DATE,
  is_current BOOLEAN,
  -- Audit columns
  created_date DATE,
  updated_date DATE
);

-- Product dimension
CREATE TABLE dim_product (
  product_id INT PRIMARY KEY,
  product_code VARCHAR(50),
  product_name VARCHAR(200),
  product_category VARCHAR(50),
  product_subcategory VARCHAR(50),
  product_line VARCHAR(50),
  supplier_id INT,
  brand VARCHAR(50),
  model VARCHAR(100),
  color VARCHAR(30),
  size VARCHAR(10),
  unit_cost DECIMAL(10, 2),
  list_price DECIMAL(10, 2),
  cost_to_list_ratio DECIMAL(5, 4),
  product_status VARCHAR(20),
  effective_date DATE,
  end_date DATE,
  is_current BOOLEAN
);

Slowly Changing Dimensions (SCD)

Type 1: Overwrite

-- Simply update the existing record
UPDATE dim_customer
SET
  email = 'new_email@example.com',
  phone = '555-9999',
  updated_date = CURRENT_DATE
WHERE customer_id = 1;

Type 2: Add New Row

-- Close old row
UPDATE dim_customer
SET
  is_current = FALSE,
  end_date = CURRENT_DATE - INTERVAL '1 day'
WHERE customer_id = 1 AND is_current = TRUE;

-- Insert new row
INSERT INTO dim_customer
VALUES (
  customer_id,
  customer_code,
  new_values...,
  CURRENT_DATE,  -- effective_date
  NULL,          -- end_date
  TRUE,          -- is_current
  CURRENT_DATE   -- created_date
);

Type 3: Add New Column

-- Add previous value columns
ALTER TABLE dim_customer ADD COLUMN previous_city VARCHAR(50);
ALTER TABLE dim_customer ADD COLUMN previous_city_start_date DATE;

-- Update previous columns when changing current
UPDATE dim_customer
SET
  previous_city = city,
  previous_city_start_date = CURRENT_DATE,
  city = 'New York'
WHERE customer_id = 1;

Conformed Dimensions

-- Single dim_date used across all fact tables
SELECT
  f.sales_id,
  f.quantity * f.unit_price as revenue,
  d.month_name,
  d.year
FROM fact_sales f
JOIN dim_date d ON f.date_id = d.date_id;

-- Reuse dim_customer in multiple facts
SELECT
  fs.sales_id,
  fc.call_id,
  c.customer_segment
FROM fact_sales fs
JOIN dim_customer c ON fs.customer_id = c.customer_id
LEFT JOIN fact_customer_calls fc ON c.customer_id = fc.customer_id;

Aggregate Tables (Materialized Views)

-- Pre-calculate common aggregations for performance
CREATE MATERIALIZED VIEW sales_summary_daily AS
SELECT
  d.full_date,
  d.month_name,
  d.year,
  p.product_category,
  c.customer_segment,
  COUNT(DISTINCT fs.sales_id) as transaction_count,
  SUM(fs.quantity) as total_quantity,
  ROUND(SUM(fs.net_sales), 2) as total_sales,
  ROUND(AVG(fs.net_sales), 2) as avg_sale,
  COUNT(DISTINCT fs.customer_id) as unique_customers
FROM fact_sales fs
JOIN dim_date d ON fs.date_id = d.date_id
JOIN dim_product p ON fs.product_id = p.product_id
JOIN dim_customer c ON fs.customer_id = c.customer_id
GROUP BY d.full_date, d.month_name, d.year, p.product_category, c.customer_segment;

-- Refresh materialized view
REFRESH MATERIALIZED VIEW sales_summary_daily;

-- Query aggregate table instead of fact table
SELECT
  month_name,
  product_category,
  SUM(total_sales) as monthly_sales
FROM sales_summary_daily
WHERE year = 2024
GROUP BY month_name, product_category;

Bridge Tables (Many-to-Many)

-- For many-to-many relationships (e.g., product to categories)
CREATE TABLE bridge_product_category (
  product_id INT,
  category_id INT,
  PRIMARY KEY (product_id, category_id),
  FOREIGN KEY (product_id) REFERENCES dim_product(product_id),
  FOREIGN KEY (category_id) REFERENCES dim_category(category_id)
);

-- Query with bridge table
SELECT
  p.product_name,
  STRING_AGG(DISTINCT c.category_name, ', ') as categories,
  COUNT(DISTINCT bc.category_id) as category_count
FROM dim_product p
LEFT JOIN bridge_product_category bc ON p.product_id = bc.product_id
LEFT JOIN dim_category c ON bc.category_id = c.category_id
GROUP BY p.product_id, p.product_name;

Data Quality Metrics

-- Monitor fact table metrics
SELECT
  COUNT(*) as total_records,
  COUNT(DISTINCT customer_id) as unique_customers,
  COUNT(DISTINCT product_id) as unique_products,
  MIN(sale_amount) as min_sale,
  MAX(sale_amount) as max_sale,
  ROUND(AVG(sale_amount), 2) as avg_sale,
  COUNT(CASE WHEN sale_amount < 0 THEN 1 END) as negative_sales,
  COUNT(CASE WHEN sale_amount IS NULL THEN 1 END) as null_sales,
  MAX(load_timestamp) as last_load_time
FROM fact_sales;

-- Dimension quality checks
SELECT
  'dim_customer' as dimension,
  COUNT(*) as total_records,
  COUNT(DISTINCT customer_id) as distinct_ids,
  COUNT(CASE WHEN first_name IS NULL THEN 1 END) as null_first_names,
  COUNT(CASE WHEN is_current = TRUE THEN 1 END) as current_records
FROM dim_customer;

Next Steps

Learn ETL/ELT pipeline design and data transformation patterns in the etl-pipelines skill.

ETL & Data Pipelines

Extract Patterns

From Relational Databases

-- Full extract of source table
SELECT * FROM source_system.customers;

-- Incremental extract using timestamp
SELECT *
FROM source_system.orders
WHERE updated_timestamp > (
  SELECT MAX(last_sync_time)
  FROM pipeline_control.sync_status
  WHERE source_table = 'orders'
);

-- Change Data Capture (CDC) simulation
CREATE TABLE source.orders_cdc (
  order_id INT,
  customer_id INT,
  amount DECIMAL(12, 2),
  operation_type VARCHAR(1),  -- 'I' = Insert, 'U' = Update, 'D' = Delete
  operation_timestamp TIMESTAMP
);

-- Extract only changed records
SELECT *
FROM source.orders_cdc
WHERE operation_timestamp >= (
  SELECT MAX(last_processed_time)
  FROM pipeline_control.cdc_status
)
ORDER BY operation_timestamp;

From APIs and Streaming

# Pseudocode for API extraction
def extract_from_api():
    url = "https://api.example.com/data"
    headers = {"Authorization": f"Bearer {api_token}"}

    # Incremental extraction with cursor
    last_cursor = get_last_cursor()

    params = {
        "limit": 1000,
        "cursor": last_cursor
    }

    response = requests.get(url, headers=headers, params=params)
    data = response.json()

    # Save extracted data
    save_to_raw_storage(data)

    # Update cursor for next run
    save_cursor(data['next_cursor'])

    return data

Transform Patterns

Data Cleaning

-- Standardize and clean data
SELECT
  UPPER(TRIM(first_name)) as first_name,
  UPPER(TRIM(last_name)) as last_name,
  LOWER(TRIM(email)) as email,
  COALESCE(phone, 'Unknown') as phone,
  CASE
    WHEN LENGTH(phone) < 10 THEN NULL
    ELSE phone
  END as cleaned_phone,
  -- Remove duplicates
  ROW_NUMBER() OVER (PARTITION BY email ORDER BY created_date DESC) as row_num
FROM raw_customers
WHERE first_name IS NOT NULL
  AND email IS NOT NULL
HAVING ROW_NUMBER() = 1;

Data Validation

-- Validation queries during transformation
SELECT
  'Email format' as validation,
  COUNT(*) as failed_count
FROM raw_customers
WHERE email NOT LIKE '%@%.%'

UNION ALL

SELECT
  'Missing required fields',
  COUNT(*)
FROM raw_customers
WHERE first_name IS NULL
  OR last_name IS NULL
  OR email IS NULL

UNION ALL

SELECT
  'Salary out of range',
  COUNT(*)
FROM raw_employees
WHERE salary < 0 OR salary > 10000000;

Complex Transformations

-- Multi-step transformation with CTEs
WITH raw_data AS (
  -- Step 1: Extract and clean
  SELECT
    customer_id,
    order_id,
    CAST(amount AS DECIMAL(12, 2)) as amount,
    CAST(order_date AS DATE) as order_date
  FROM raw_orders
  WHERE amount > 0 AND order_date IS NOT NULL
),
aggregated_data AS (
  -- Step 2: Aggregate
  SELECT
    customer_id,
    DATE_TRUNC('month', order_date)::DATE as month,
    COUNT(DISTINCT order_id) as transaction_count,
    SUM(amount) as total_amount,
    AVG(amount) as avg_amount
  FROM raw_data
  GROUP BY customer_id, DATE_TRUNC('month', order_date)
),
enriched_data AS (
  -- Step 3: Enrich with business logic
  SELECT
    customer_id,
    month,
    transaction_count,
    total_amount,
    avg_amount,
    CASE
      WHEN total_amount > 10000 THEN 'High Value'
      WHEN total_amount > 1000 THEN 'Medium Value'
      ELSE 'Low Value'
    END as customer_value_segment,
    CURRENT_TIMESTAMP as load_timestamp
  FROM aggregated_data
)
INSERT INTO transformed_customer_monthly
SELECT * FROM enriched_data;

Load Patterns

Append Load (Full)

-- Simple append - add all new records
INSERT INTO fact_sales
SELECT
  sale_id,
  date_id,
  customer_id,
  product_id,
  amount,
  CURRENT_TIMESTAMP as load_time
FROM staging_sales
WHERE load_status = 'Ready';

-- Update load control
UPDATE pipeline_control.load_status
SET last_load_time = CURRENT_TIMESTAMP,
    record_count = (SELECT COUNT(*) FROM staging_sales),
    status = 'Completed'
WHERE pipeline_name = 'fact_sales_load';

Incremental Load (Upsert)

-- Insert or update strategy using MERGE (or UPSERT if supported)
MERGE INTO fact_customer_monthly t
USING staging_customer_monthly s
ON t.customer_id = s.customer_id
  AND t.month = s.month
WHEN MATCHED THEN
  UPDATE SET
    transaction_count = s.transaction_count,
    total_sales = s.total_sales,
    updated_date = CURRENT_TIMESTAMP
WHEN NOT MATCHED THEN
  INSERT (customer_id, month, transaction_count, total_sales, created_date, updated_date)
  VALUES (s.customer_id, s.month, s.transaction_count, s.total_sales,
          CURRENT_TIMESTAMP, CURRENT_TIMESTAMP);

-- For databases without MERGE:
-- Step 1: Delete existing records
DELETE FROM fact_customer_monthly
WHERE (customer_id, month) IN (
  SELECT customer_id, month FROM staging_customer_monthly
);

-- Step 2: Insert new records
INSERT INTO fact_customer_monthly
SELECT * FROM staging_customer_monthly;

Dimension Load (SCD Type 2)

-- Slowly Changing Dimension - Type 2 (versioning)

-- Step 1: Identify changes
WITH changes AS (
  SELECT
    s.customer_id,
    s.email,
    s.phone,
    s.address,
    s.city,
    d.email as old_email,
    d.phone as old_phone,
    d.address as old_address,
    d.city as old_city,
    CASE
      WHEN s.email != d.email OR s.phone != d.phone
        OR s.address != d.address OR s.city != d.city
      THEN TRUE
      ELSE FALSE
    END as has_changes
  FROM staging_customers s
  LEFT JOIN dim_customer d ON s.customer_id = d.customer_id
    AND d.is_current = TRUE
)
-- Step 2: Close old records for customers with changes
UPDATE dim_customer
SET is_current = FALSE,
    end_date = CURRENT_DATE - INTERVAL '1 day'
WHERE customer_id IN (
  SELECT customer_id FROM changes WHERE has_changes = TRUE
)
AND is_current = TRUE;

-- Step 3: Insert new versions
INSERT INTO dim_customer (
  customer_id, email, phone, address, city,
  effective_date, end_date, is_current,
  created_date, updated_date
)
SELECT
  s.customer_id,
  s.email,
  s.phone,
  s.address,
  s.city,
  CURRENT_DATE,
  NULL,
  TRUE,
  CURRENT_TIMESTAMP,
  CURRENT_TIMESTAMP
FROM staging_customers s
WHERE s.customer_id NOT IN (
  SELECT DISTINCT customer_id FROM dim_customer WHERE is_current = TRUE
);

Pipeline Orchestration

Control Framework

-- Pipeline execution log table
CREATE TABLE pipeline_control.execution_log (
  execution_id BIGINT PRIMARY KEY AUTO_INCREMENT,
  pipeline_name VARCHAR(100),
  step_name VARCHAR(100),
  start_time TIMESTAMP,
  end_time TIMESTAMP,
  status VARCHAR(20),  -- 'Running', 'Success', 'Failed'
  record_count INT,
  error_message TEXT,
  created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Track last successful load
CREATE TABLE pipeline_control.load_watermark (
  pipeline_name VARCHAR(100) PRIMARY KEY,
  last_successful_load TIMESTAMP,
  records_processed INT,
  last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Mark pipeline start
INSERT INTO pipeline_control.execution_log
(pipeline_name, step_name, start_time, status)
VALUES ('customer_etl', 'extract', CURRENT_TIMESTAMP, 'Running');

-- Update on completion
UPDATE pipeline_control.execution_log
SET
  end_time = CURRENT_TIMESTAMP,
  status = 'Success',
  record_count = (SELECT COUNT(*) FROM staging_customers)
WHERE execution_id = @current_exec_id;

Error Handling

# Pseudocode for ETL error handling
def run_etl_pipeline():
    execution_id = log_pipeline_start('customer_etl')

    try:
        # Extract
        raw_data = extract_data()
        log_step(execution_id, 'extract', 'Success', len(raw_data))

        # Transform
        clean_data = transform_data(raw_data)
        validate_data(clean_data)
        log_step(execution_id, 'transform', 'Success', len(clean_data))

        # Load
        load_data(clean_data)
        log_step(execution_id, 'load', 'Success', len(clean_data))

        log_pipeline_completion(execution_id, 'Success')

    except ValidationError as e:
        log_error(execution_id, 'validation', str(e))
        send_alert(f"ETL failed: {e}")
        rollback_transaction()

    except DatabaseError as e:
        log_error(execution_id, 'database', str(e))
        send_alert(f"Database error: {e}")
        retry_with_backoff()

Performance Optimization

-- Batch processing for large volumes
-- Instead of: INSERT INTO target SELECT * FROM source (slow for millions)

-- Do: Process in batches
DECLARE @batch_size INT = 100000;
DECLARE @offset INT = 0;
DECLARE @total INT = (SELECT COUNT(*) FROM source_table);

WHILE @offset < @total
BEGIN
  INSERT INTO target_table
  SELECT * FROM source_table
  LIMIT @batch_size OFFSET @offset;

  SET @offset = @offset + @batch_size;
END;

-- Parallel processing
-- Process multiple dimensions in parallel
-- Fact table load only after all dimensions complete

-- Compression and partitioning
CREATE TABLE large_fact_table (
  date_id INT,
  customer_id INT,
  amount DECIMAL(12, 2)
)
PARTITION BY RANGE (YEAR(date_id)) (
  PARTITION p_2022 VALUES LESS THAN (2023),
  PARTITION p_2023 VALUES LESS THAN (2024),
  PARTITION p_2024 VALUES LESS THAN (2025),
  PARTITION p_future VALUES LESS THAN MAXVALUE
);

-- Create indexes on load columns
CREATE INDEX idx_staging_sales_date ON staging_sales(load_date);

Best Practices Checklist

✅ Implement comprehensive error logging and alerting ✅ Use staging tables for intermediate transformations ✅ Version your data transformations ✅ Implement data quality checks at each stage ✅ Use idempotent operations (safe to run multiple times) ✅ Implement recovery procedures for failed loads ✅ Monitor pipeline performance metrics ✅ Use partitioning for large tables ✅ Implement incremental loads where possible ✅ Document data lineage and transformations

data-engineer Guide