数仓建模：如何设计一个高性能的数据仓库模型?

销售事实表(Sales_Fact)
客户行为事实表(Customer_Behavior_Fact)
库存事实表(Inventory_Fact)

时间维度(Time_Dim)
产品维度(Product_Dim)
客户维度(Customer_Dim)
地理维度(Geography_Dim)
供应商维度(Supplier_Dim)
营销活动维度(Campaign_Dim)

-- 销售事实表
CREATE TABLE Sales_Fact (
    sale_id BIGINT PRIMARY KEY,
    order_date_key INT,
    customer_key INT,
    product_key INT,
    geography_key INT,
    campaign_key INT,
    quantity INT,
    unit_price DECIMAL(10,2),
    total_amount DECIMAL(10,2),
    discount_amount DECIMAL(10,2),
    net_amount DECIMAL(10,2)
);


-- 时间维度表
CREATE TABLE Time_Dim (
    date_key INT PRIMARY KEY,
    full_date DATE,
    year INT,
    quarter INT,
    month INT,
    week INT,
    day INT,
    is_weekend BOOLEAN,
    is_holiday BOOLEAN
);


-- 产品维度表
CREATE TABLE Product_Dim (
    product_key INT PRIMARY KEY,
    product_id VARCHAR(50),
    product_name VARCHAR(100),
    category VARCHAR(50),
    subcategory VARCHAR(50),
    brand VARCHAR(50),
    supplier_key INT,
    unit_cost DECIMAL(10,2)
);


-- 客户维度表
CREATE TABLE Customer_Dim (
    customer_key INT PRIMARY KEY,
    customer_id VARCHAR(50),
    first_name VARCHAR(50),
    last_name VARCHAR(50),
    email VARCHAR(100),
    phone VARCHAR(20),
    registration_date DATE,
    customer_segment VARCHAR(20)
);


-- 地理维度表
CREATE TABLE Geography_Dim (
    geography_key INT PRIMARY KEY,
    city VARCHAR(50),
    state VARCHAR(50),
    country VARCHAR(50),
    region VARCHAR(50),
    latitude DECIMAL(9,6),
    longitude DECIMAL(9,6)
);


-- 营销活动维度表
CREATE TABLE Campaign_Dim (
    campaign_key INT PRIMARY KEY,
    campaign_id VARCHAR(50),
    campaign_name VARCHAR(100),
    campaign_type VARCHAR(50),
    start_date DATE,
    end_date DATE,
    channel VARCHAR(50),
    budget DECIMAL(10,2)
);

import pandas as pd
from sqlalchemy import create_engine


# 连接到源数据库和目标数据仓库
source_engine = create_engine('postgresql://user:password@source_host/source_db')
dw_engine = create_engine('postgresql://user:password@dw_host/data_warehouse')


# 抽取源数据
orders_df = pd.read_sql("SELECT * FROM orders", source_engine)
products_df = pd.read_sql("SELECT * FROM products", source_engine)


# 转换数据
sales_fact_df = orders_df.merge(products_df, on='product_id')
sales_fact_df['order_date_key'] = pd.to_datetime(sales_fact_df['order_date']).dt.strftime('%Y%m%d').astype(int)
sales_fact_df['total_amount'] = sales_fact_df['quantity'] * sales_fact_df['unit_price']


# 加载数据到数据仓库
sales_fact_df.to_sql('Sales_Fact', dw_engine, if_exists='append', index=False)

def incremental_load(last_load_time):
    query = f"""
    SELECT * FROM orders
    WHERE order_date > '{last_load_time}'
    """
    new_orders_df = pd.read_sql(query, source_engine)
    # 处理新订单数据
    # ...
    
    # 更新最后加载时间
    update_last_load_time(datetime.now())


# 定期运行增量加载
schedule.every(1).hour.do(incremental_load, last_load_time=get_last_load_time())

def data_quality_check(df, table_name):
    # 检查空值
    null_counts = df.isnull().sum()
    if null_counts.any():
        log_error(f"发现空值在 {table_name}: {null_counts}")


    # 检查唯一性约束
    if df['sale_id'].nunique() != len(df):log_error(f"{table_name} 中的 sale_id 不是唯一的")


    # 检查数值范围
    if (df['quantity'] <= 0).any():
        log_error(f"{table_name} 中存在非正数量")


    # 检查日期有效性
    if (df['order_date'] > datetime.now()).any():
        log_error(f"{table_name} 中存在未来日期")


# 在数据加载前进行检查
data_quality_check(sales_fact_df, 'Sales_Fact')

-- 按日期范围分区的sales_fact表
CREATE TABLE sales_fact (
    sale_id BIGINT,
    order_date DATE,
    -- 其他列...
) PARTITION BY RANGE (order_date);


CREATE TABLE sales_fact_2023 PARTITION OF sales_fact
    FOR VALUES FROM ('2023-01-01') TO ('2024-01-01');


CREATE TABLE sales_fact_2024 PARTITION OF sales_fact
    FOR VALUES FROM ('2024-01-01') TO ('2025-01-01');

-- 在sales_fact表的外键上创建索引
CREATE INDEX idx_sales_fact_date ON sales_fact(order_date_key);
CREATE INDEX idx_sales_fact_product ON sales_fact(product_key);
CREATE INDEX idx_sales_fact_customer ON sales_fact(customer_key);


-- 在维度表的主键上创建索引（如果DBMS没有自动创建）
CREATE INDEX idx_time_dim_pk ON time_dim(date_key);
CREATE INDEX idx_product_dim_pk ON product_dim(product_key);
CREATE INDEX idx_customer_dim_pk ON customer_dim(customer_key);

-- 在PostgreSQL中使用ZSTD压缩
ALTER TABLE sales_fact SET (compression=zstd);

CREATE MATERIALIZED VIEW monthly_sales AS
SELECT 
    t.year,
    t.month,
    p.category,
    SUM(s.total_amount) as total_sales
FROM 
    sales_fact s
    JOIN time_dim t ON s.order_date_key = t.date_key
    JOIN product_dim p ON s.product_key = p.product_key
GROUP BY 
    t.year, t.month, p.category;


-- 创建索引以加快查询速度
CREATE INDEX idx_monthly_sales ON monthly_sales(year, month, category);


-- 定期刷新物化视图
REFRESH MATERIALIZED VIEW monthly_sales;

-- 优化前
SELECT c.customer_name, SUM(s.total_amount)
FROM sales_fact s
JOIN customer_dim c ON s.customer_key = c.customer_key
WHERE s.order_date_key IN (
    SELECT date_key 
    FROM time_dim 
    WHERE year = 2023 AND month = 12
)
GROUP BY c.customer_name;


-- 优化后
SELECT c.customer_name, SUM(s.total_amount)
FROM sales_fact s
JOIN customer_dim c ON s.customer_key = c.customer_key
JOIN time_dim t ON s.order_date_key = t.date_key
WHERE t.year = 2023 AND t.month = 12
GROUP BY c.customer_name;

-- 利用分区裁剪的查询
SELECT SUM(total_amount)
FROM sales_fact
WHERE order_date BETWEEN '2023-01-01' AND '2023-12-31';

-- 在PostgreSQL中设置并行查询
SET max_parallel_workers_per_gather = 4;

import redis
import json


redis_client = redis.Redis(host='localhost', port=6379, db=0)


def cached_query(query, cache_key, expire_time=3600):
    # 尝试从缓存获取结果
    cached_result = redis_client.get(cache_key)
    if cached_result:
        return json.loads(cached_result)


    # 如果缓存miss，执行查询
    result = execute_query(query)


    # 将结果存入缓存
    redis_client.setex(cache_key, expire_time, json.dumps(result))


    return result


# 使用缓存查询
monthly_sales = cached_query(
    "SELECT * FROM monthly_sales WHERE year = 2023",
    "monthly_sales_2023",
    3600  # 缓存1小时
)

-- 在PostgreSQL中创建列式表（使用cstore_fdw扩展）
CREATE FOREIGN TABLE sales_fact_columnar (
    sale_id BIGINT,
    order_date_key INTEGER,
    -- 其他列...
) SERVER cstore_server;

CREATE TABLE daily_sales_summary AS
SELECT 
    order_date_key,
    SUM(total_amount) as daily_total,
    COUNT(DISTINCT customer_key) as unique_customers
FROM 
    sales_fact
GROUP BY 
    order_date_key;


-- 创建索引以加快查询
CREATE INDEX idx_daily_sales_summary ON daily_sales_summary(order_date_key);

EXPLAIN ANALYZE
SELECT p.category, SUM(s.total_amount)
FROM sales_fact s
JOIN product_dim p ON s.product_key = p.product_key
WHERE s.order_date_key BETWEEN 20230101 AND 20231231
GROUP BY p.category;

# 创建一个新的分支来实施模型变更
git checkout -b add-new-dimension


# 添加新的维度表DDL
git add new_dimension.sql


# 提交变更
git commit -m "Add new dimension for customer loyalty program"


# 合并到主分支
git checkout main
git merge add-new-dimension

-- 增加新列到现有维度表
ALTER TABLE customer_dim ADD COLUMN loyalty_tier VARCHAR(20);


-- 更新现有数据
UPDATE customer_dim
SET loyalty_tier = 'Standard'
WHERE loyalty_tier IS NULL;


-- 为新列添加非空约束
ALTER TABLE customer_dim ALTER COLUMN loyalty_tier SET NOT NULL;

import psycopg2
import time


def monitor_query_performance(query):
    conn = psycopg2.connect("dbname=datawarehouse user=dw_user")
    cur = conn.cursor()


    start_time = time.time()
    cur.execute(query)
    end_time = time.time()


    execution_time = end_time - start_time
    print(f"Query execution time: {execution_time:.2f} seconds")


    cur.close()
    conn.close()


    return execution_time


# 监控关键查询的性能
daily_performance = monitor_query_performance("SELECT * FROM daily_sales_summary")
if daily_performance > 5:  # 如果查询时间超过5秒
    send_alert("Daily sales summary query is slow")

from great_expectations import DataContext


def run_data_quality_checks():
    context = DataContext("/path/to/great_expectations")
    suite = context.get_expectation_suite("sales_fact_suite")
    batch = context.get_batch({"path": "/data/sales_fact.csv"}, suite)
    results = context.run_validation(batch, expectation_suite=suite)
    
    if not results["success"]:
        send_alert("Data quality check failed for sales_fact")


# 定期运行数据质量检查
schedule.every().day.at("01:00").do(run_data_quality_checks)

# Sales Fact Table


## Description
This table contains all sales transactions at the order item level.


## Columns
- sale_id (BIGINT): Unique identifier for each sale item
- order_date_key (INT): Foreign key to Time_Dim table
- customer_key (INT): Foreign key to Customer_Dim table
- product_key (INT): Foreign key to Product_Dim table
- quantity (INT): Number of items sold
- unit_price (DECIMAL): Price per unit
- total_amount (DECIMAL): Total sale amount (quantity * unit_price)


## Common Queries
1. Total sales by date:
   ```sql
   SELECT t.full_date, SUM(s.total_amount)
   FROM sales_fact s
   JOIN time_dim t ON s.order_date_key = t.date_key
   GROUP BY t.full_date
   ORDER BY t.full_date

-- 创建通用的分类表
CREATE TABLE category_types (
    category_type_id SERIAL PRIMARY KEY,
    category_type_name VARCHAR(50) UNIQUE NOT NULL
);


CREATE TABLE categories (
    category_id SERIAL PRIMARY KEY,
    category_type_id INT REFERENCES category_types(category_type_id),
    category_name VARCHAR(50) NOT NULL,
    UNIQUE (category_type_id, category_name)
);


-- 插入示例数据
INSERT INTO category_types (category_type_name) VALUES ('Product Category'), ('Customer Segment');
INSERT INTO categories (category_type_id, category_name) 
VALUES 
    (1, 'Electronics'), (1, 'Clothing'), 
    (2, 'New'), (2, 'Returning');

-- 适度反规范化的客户维度表
CREATE TABLE customer_dim (
    customer_key INT PRIMARY KEY,
    customer_id VARCHAR(50),
    first_name VARCHAR(50),
    last_name VARCHAR(50),
    current_address VARCHAR(200),
    current_city VARCHAR(50),
    current_state VARCHAR(50),
    current_country VARCHAR(50),
    registration_date DATE
);

def monitor_data_growth():
    conn = create_database_connection()
    cursor = conn.cursor()
    
    cursor.execute("SELECT COUNT(*) FROM sales_fact")
    current_count = cursor.fetchone()[0]
    
    cursor.execute("SELECT COUNT(*) FROM sales_fact WHERE order_date >= DATE_TRUNC('month', CURRENT_DATE - INTERVAL '1 month')")
    last_month_count = cursor.fetchone()[0]
    
    growth_rate = (last_month_count  current_count) * 100
    
    if growth_rate > 10:  # 如果月if growth_rate > 10:  # 如果月增长率超过10%
        send_alert(f"Data growth rate is high: {growth_rate:.2f}%")


# 定期运行数据增长监控
schedule.every().day.do(monitor_data_growth)

-- 使用SCD Type 2的客户维度表
CREATE TABLE customer_dim (
    customer_key SERIAL PRIMARY KEY,
    customer_id VARCHAR(50),
    first_name VARCHAR(50),
    last_name VARCHAR(50),
    email VARCHAR(100),
    address VARCHAR(200),
    effective_date DATE,
    end_date DATE,
    is_current BOOLEAN
);


-- 更新客户信息的存储过程
CREATE OR REPLACE PROCEDURE update_customer_dim(
    p_customer_id VARCHAR(50),
    p_first_name VARCHAR(50),
    p_last_name VARCHAR(50),
    p_email VARCHAR(100),
    p_address VARCHAR(200)
)
LANGUAGE plpgsql
AS $$
BEGIN
    -- 结束当前记录
    UPDATE customer_dim
    SET end_date = CURRENT_DATE - INTERVAL '1 day',
        is_current = FALSE
    WHERE customer_id = p_customer_id AND is_current = TRUE;


    -- 插入新记录
    INSERT INTO customer_dim (customer_id, first_name, last_name, email, address, effective_date, end_date, is_current)
    VALUES (p_customer_id, p_first_name, p_last_name, p_email, p_address, CURRENT_DATE, '9999-12-31', TRUE);
END;
$$;

-- 创建多级聚合表
CREATE TABLE sales_summary (
    date_key INT,
    product_key INT,
    geography_key INT,
    total_sales DECIMAL(15,2),
    total_quantity INT,
    granularity VARCHAR(20),  -- 'daily', 'monthly', 'yearly'
    PRIMARY KEY (date_key, product_key, geography_key, granularity)
);


-- 填充聚合表
INSERT INTO sales_summary
SELECT 
    t.date_key,
    s.product_key,
    s.geography_key,
    SUM(s.total_amount) as total_sales,
    SUM(s.quantity) as total_quantity,
    'daily' as granularity
FROM 
    sales_fact s
    JOIN time_dim t ON s.order_date_key = t.date_key
GROUP BY 
    t.date_key, s.product_key, s.geography_key


UNION ALL


SELECT 
    DATE_TRUNC('month', t.full_date)::INT as date_key,
    s.product_key,
    s.geography_key,
    SUM(s.total_amount) as total_sales,
    SUM(s.quantity) as total_quantity,
    'monthly' as granularity
FROM 
    sales_fact s
    JOIN time_dim t ON s.order_date_key = t.date_key
GROUP BY 
    DATE_TRUNC('month', t.full_date), s.product_key, s.geography_key;

-- 创建角色和授予适当的权限
CREATE ROLE analyst;
GRANT SELECT ON sales_summary TO analyst;


-- 对敏感列进行加密
ALTER TABLE customer_dim
ALTER COLUMN email TYPE bytea 
USING PGP_SYM_ENCRYPT(email::text, 'AES_KEY')::bytea;


-- 创建视图来屏蔽敏感信息
CREATE VIEW customer_dim_masked AS
SELECT 
    customer_key,
    customer_id,
    first_name,
    last_name,
    CASE 
        WHEN LENGTH(email::text) > 5 THEN 
            LEFT(email::text, 2) || '***' || RIGHT(email::text, 2)
        ELSE '***'
    END as masked_email,
    address
FROM customer_dim;

-- 添加外键约束
ALTER TABLE sales_fact
ADD CONSTRAINT fk_sales_customer
FOREIGN KEY (customer_key) REFERENCES customer_dim(customer_key);


-- 创建触发器以确保一致性
CREATE OR REPLACE FUNCTION check_date_consistency()
RETURNS TRIGGER AS $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM time_dim WHERE date_key = NEW.order_date_key) THEN
        RAISE EXCEPTION 'Invalid order_date_key: %', NEW.order_date_key;
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;


CREATE TRIGGER sales_fact_date_consistency
BEFORE INSERT OR UPDATE ON sales_fact
FOR EACH ROW EXECUTE FUNCTION check_date_consistency();