-- 销售趋势分析
SELECT 
    date_trunc('month', order_date) as month,
    SUM(order_amount) as monthly_sales,
    COUNT(DISTINCT customer_id) as active_customers,
    AVG(order_amount) as avg_order_value
FROM fact_orders
WHERE order_date >= '2023-01-01'
GROUP BY date_trunc('month', order_date)
ORDER BY month;

-- 用户行为漏斗分析
WITH user_funnel AS (
    SELECT 
        user_id,
        MAX(CASE WHEN event_type = 'visit' THEN 1 ELSE 0 END) as visited,
        MAX(CASE WHEN event_type = 'view_product' THEN 1 ELSE 0 END) as viewed,
        MAX(CASE WHEN event_type = 'add_cart' THEN 1 ELSE 0 END) as added_cart,
        MAX(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) as purchased
    FROM fact_user_events
    WHERE event_date = current_date
    GROUP BY user_id
)
SELECT 
    SUM(visited) as visitors,
    SUM(viewed) as product_viewers,
    SUM(added_cart) as cart_users,
    SUM(purchased) as buyers,
    SUM(viewed) * 100.0 / SUM(visited) as view_rate,
    SUM(purchased) * 100.0 / SUM(visited) as conversion_rate
FROM user_funnel;

-- Hub表：业务实体
CREATE TABLE hub_customer (
    customer_hk CHAR(32) PRIMARY KEY,  -- Hash Key
    customer_id VARCHAR(50),           -- 业务键
    load_date TIMESTAMP,               -- 加载时间
    record_source VARCHAR(50)          -- 数据源
);

-- Link表：关系
CREATE TABLE link_customer_order (
    customer_order_hk CHAR(32) PRIMARY KEY,
    customer_hk CHAR(32),
    order_hk CHAR(32),
    load_date TIMESTAMP,
    record_source VARCHAR(50),
    FOREIGN KEY (customer_hk) REFERENCES hub_customer(customer_hk),
    FOREIGN KEY (order_hk) REFERENCES hub_order(order_hk)
);

-- Satellite表：属性
CREATE TABLE sat_customer_details (
    customer_hk CHAR(32),
    load_date TIMESTAMP,
    customer_name VARCHAR(100),
    email VARCHAR(100),
    phone VARCHAR(20),
    hash_diff CHAR(32),               -- 变更检测
    PRIMARY KEY (customer_hk, load_date),
    FOREIGN KEY (customer_hk) REFERENCES hub_customer(customer_hk)
);

# 流处理作业 (Flink/Spark Streaming)
def process_real_time_orders():
    env = StreamExecutionEnvironment.get_execution_environment()
    
    # 从Kafka读取实时订单数据
    orders_stream = env.add_source(
        FlinkKafkaConsumer('orders_topic', SimpleStringSchema(), kafka_props)
    )
    
    # 实时聚合计算
    order_metrics = orders_stream \
        .map(parse_order) \
        .key_by(lambda x: x.customer_id) \
        .window(TumblingProcessingTimeWindows.of(Time.minutes(5))) \
        .aggregate(OrderAggregateFunction())
    
    # 写入速度视图存储
    order_metrics.add_sink(RedisSink())

# 批处理作业 (Spark)
def process_batch_orders():
    spark = SparkSession.builder.appName("BatchOrderProcessing").getOrCreate()
    
    # 读取历史订单数据
    orders_df = spark.read.parquet("hdfs://orders/")
    
    # 批量聚合计算
    order_summary = orders_df \
        .groupBy("customer_id", date_trunc("day", "order_date")) \
        .agg(
            sum("order_amount").alias("daily_amount"),
            count("*").alias("daily_orders")
        )
    
    # 写入批量视图存储
    order_summary.write.mode("overwrite").parquet("hdfs://order_summary/")

-- 创建Delta表（启用自动优化和变更数据捕获）
CREATE TABLE orders_delta (...) USING DELTA
PARTITIONED BY (order_date)
TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true');

-- MERGE操作实现Upsert（最常用的增量更新模式）
MERGE INTO target USING source ON target.id = source.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

-- 时间旅行：查询历史版本数据（用于数据审计、回溯分析）
SELECT * FROM orders_delta VERSION AS OF 5;
SELECT * FROM orders_delta TIMESTAMP AS OF '2023-12-25 10:00:00';

-- 订单事实表
CREATE TABLE fact_orders (
    -- 代理键 (Surrogate Key)
    order_sk BIGINT PRIMARY KEY,
    
    -- 维度外键
    customer_sk BIGINT NOT NULL,
    product_sk BIGINT NOT NULL,
    date_sk INT NOT NULL,
    store_sk BIGINT NOT NULL,
    
    -- 退化维度 (Degenerate Dimension)
    order_number VARCHAR(50) NOT NULL,
    order_line_number INT NOT NULL,
    
    -- 度量值 (Measures)
    quantity INT NOT NULL,
    unit_price DECIMAL(10,2) NOT NULL,
    discount_amount DECIMAL(10,2) DEFAULT 0,
    tax_amount DECIMAL(10,2) NOT NULL,
    total_amount DECIMAL(10,2) NOT NULL,
    
    -- 审计字段
    created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    -- 外键约束
    FOREIGN KEY (customer_sk) REFERENCES dim_customer(customer_sk),
    FOREIGN KEY (product_sk) REFERENCES dim_product(product_sk),
    FOREIGN KEY (date_sk) REFERENCES dim_date(date_sk),
    FOREIGN KEY (store_sk) REFERENCES dim_store(store_sk)
)
PARTITION BY RANGE (date_sk);

-- 订单生命周期事实表
CREATE TABLE fact_order_lifecycle (
    order_sk BIGINT PRIMARY KEY,
    customer_sk BIGINT NOT NULL,
    product_sk BIGINT NOT NULL,
    
    -- 各阶段日期维度
    order_date_sk INT,
    payment_date_sk INT,
    shipped_date_sk INT,
    delivered_date_sk INT,
    
    -- 各阶段持续时间（天）
    payment_lag_days INT,
    shipping_lag_days INT,
    delivery_lag_days INT,
    total_cycle_days INT,
    
    -- 当前状态
    current_status VARCHAR(20),
    
    -- 度量值
    order_amount DECIMAL(10,2),
    shipping_cost DECIMAL(10,2)
);

-- 客户维度表
CREATE TABLE dim_customer (
    -- 代理键
    customer_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    
    -- 自然键
    customer_id VARCHAR(50) NOT NULL,
    
    -- 基本信息
    customer_name VARCHAR(100) NOT NULL,
    customer_type VARCHAR(20) NOT NULL, -- 'individual', 'corporate'
    gender VARCHAR(10),
    birth_date DATE,
    age_range VARCHAR(20), -- '18-25', '26-35', etc.
    
    -- 联系信息
    email VARCHAR(100),
    phone VARCHAR(20),
    
    -- 地址信息 (层次结构)
    address_line1 VARCHAR(200),
    address_line2 VARCHAR(200),
    city VARCHAR(50),
    state_province VARCHAR(50),
    postal_code VARCHAR(20),
    country VARCHAR(50),
    region VARCHAR(50), -- 销售区域
    
    -- 分类属性
    customer_segment VARCHAR(30), -- 'premium', 'standard', 'basic'
    credit_rating VARCHAR(10), -- 'AAA', 'AA', 'A', etc.
    preferred_language VARCHAR(20),
    
    -- 状态信息
    is_active BOOLEAN DEFAULT TRUE,
    registration_date DATE,
    last_update_date DATE,
    
    -- SCD Type 2 字段
    effective_date DATE NOT NULL,
    expiry_date DATE DEFAULT '9999-12-31',
    is_current BOOLEAN DEFAULT TRUE,
    
    -- 审计字段
    created_by VARCHAR(50),
    created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_by VARCHAR(50),
    updated_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    -- 索引
    INDEX idx_customer_id (customer_id),
    INDEX idx_effective_date (effective_date),
    INDEX idx_is_current (is_current)
);

-- 产品维度表
CREATE TABLE dim_product (
    product_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    product_id VARCHAR(50) NOT NULL,
    
    -- 产品基本信息
    product_name VARCHAR(200) NOT NULL,
    product_description TEXT,
    brand VARCHAR(100),
    model VARCHAR(100),
    
    -- 产品层次结构
    category_level1 VARCHAR(50), -- 电器
    category_level2 VARCHAR(50), -- 手机
    category_level3 VARCHAR(50), -- 智能手机
    category_level4 VARCHAR(50), -- iPhone
    category_path VARCHAR(500),  -- 电器/手机/智能手机/iPhone
    
    -- 产品属性
    color VARCHAR(30),
    size VARCHAR(30),
    weight DECIMAL(8,2),
    material VARCHAR(100),
    
    -- 价格信息
    list_price DECIMAL(10,2),
    cost_price DECIMAL(10,2),
    margin_percent DECIMAL(5,2),
    
    -- 状态信息
    product_status VARCHAR(20), -- 'active', 'discontinued', 'coming_soon'
    launch_date DATE,
    discontinue_date DATE,
    
    -- 供应商信息
    supplier_name VARCHAR(100),
    supplier_country VARCHAR(50),
    
    effective_date DATE NOT NULL,
    expiry_date DATE DEFAULT '9999-12-31',
    is_current BOOLEAN DEFAULT TRUE
);

-- Type 1: 直接更新，不保留历史
UPDATE dim_customer 
SET 
    phone = '138-0000-0001',
    updated_date = CURRENT_TIMESTAMP,
    updated_by = 'system'
WHERE customer_id = 'C001';

-- Type 2: 保留历史，添加新记录
-- 1. 关闭当前记录
UPDATE dim_customer 
SET 
    expiry_date = CURRENT_DATE - 1,
    is_current = FALSE,
    updated_date = CURRENT_TIMESTAMP
WHERE customer_id = 'C001' AND is_current = TRUE;

-- 2. 插入新记录
INSERT INTO dim_customer (
    customer_id, customer_name, email, phone,
    effective_date, expiry_date, is_current,
    created_date, created_by
) VALUES (
    'C001', 'John Smith', '[email protected]', '138-0000-0001',
    CURRENT_DATE, '9999-12-31', TRUE,
    CURRENT_TIMESTAMP, 'system'
);

-- Type 3: 保留有限历史
ALTER TABLE dim_customer 
ADD COLUMN previous_address VARCHAR(200),
ADD COLUMN address_change_date DATE;

UPDATE dim_customer 
SET 
    previous_address = current_address,
    current_address = '新地址',
    address_change_date = CURRENT_DATE
WHERE customer_id = 'C001';

-- 当前维度表
CREATE TABLE dim_customer_current AS 
SELECT * FROM dim_customer WHERE is_current = TRUE;

-- 历史维度表
CREATE TABLE dim_customer_history AS 
SELECT * FROM dim_customer WHERE is_current = FALSE;

-- Mini维度表（只包含快变属性）
CREATE TABLE dim_customer_mini (
    customer_sk BIGINT,
    customer_segment VARCHAR(30),
    credit_rating VARCHAR(10),
    effective_date DATE,
    expiry_date DATE
);

-- 订单号作为退化维度直接存储在事实表中
CREATE TABLE fact_order_items (
    order_item_sk BIGINT PRIMARY KEY,
    
    -- 维度外键
    customer_sk BIGINT,
    product_sk BIGINT,
    date_sk INT,
    
    -- 退化维度（不需要单独的维度表）
    order_number VARCHAR(50) NOT NULL,
    invoice_number VARCHAR(50),
    tracking_number VARCHAR(50),
    
    -- 度量值
    quantity INT,
    unit_price DECIMAL(10,2),
    total_amount DECIMAL(10,2)
);

-- 查询时可以直接使用退化维度
SELECT 
    order_number,
    SUM(total_amount) as order_total
FROM fact_order_items
WHERE order_number = 'ORD-2023-001234'
GROUP BY order_number;

-- 共享的时间维度表
CREATE TABLE dim_date (
    date_sk INT PRIMARY KEY,
    full_date DATE NOT NULL,
    day_of_week INT,
    day_name VARCHAR(10),
    day_of_month INT,
    day_of_year INT,
    week_of_year INT,
    month_number INT,
    month_name VARCHAR(10),
    quarter_number INT,
    quarter_name VARCHAR(10),
    year_number INT,
    is_weekend BOOLEAN,
    is_holiday BOOLEAN,
    holiday_name VARCHAR(50),
    fiscal_year INT,
    fiscal_quarter INT,
    fiscal_month INT
);

-- 不同主题域都使用相同的时间维度
-- 销售事实表
CREATE TABLE fact_sales (
    sales_sk BIGINT PRIMARY KEY,
    date_sk INT,
    customer_sk BIGINT,
    product_sk BIGINT,
    sales_amount DECIMAL(10,2),
    FOREIGN KEY (date_sk) REFERENCES dim_date(date_sk)
);

-- 库存事实表
CREATE TABLE fact_inventory (
    inventory_sk BIGINT PRIMARY KEY,
    date_sk INT,
    product_sk BIGINT,
    warehouse_sk BIGINT,
    quantity_on_hand INT,
    FOREIGN KEY (date_sk) REFERENCES dim_date(date_sk)
);

-- 客户基本信息表
CREATE TABLE customer (
    customer_id VARCHAR(50) PRIMARY KEY,
    customer_name VARCHAR(100) NOT NULL,
    customer_type_id INT,
    birth_date DATE,
    gender VARCHAR(10),
    registration_date DATE,
    FOREIGN KEY (customer_type_id) REFERENCES customer_type(customer_type_id)
);

-- 客户类型表
CREATE TABLE customer_type (
    customer_type_id INT PRIMARY KEY,
    type_name VARCHAR(50) NOT NULL,
    type_description TEXT
);

-- 客户地址表
CREATE TABLE customer_address (
    address_id BIGINT PRIMARY KEY AUTO_INCREMENT,
    customer_id VARCHAR(50),
    address_type VARCHAR(20), -- 'billing', 'shipping', 'home'
    address_line1 VARCHAR(200),
    address_line2 VARCHAR(200),
    city VARCHAR(50),
    state_province VARCHAR(50),
    postal_code VARCHAR(20),
    country VARCHAR(50),
    is_primary BOOLEAN DEFAULT FALSE,
    effective_date DATE,
    expiry_date DATE,
    FOREIGN KEY (customer_id) REFERENCES customer(customer_id)
);

-- 客户联系方式表
CREATE TABLE customer_contact (
    contact_id BIGINT PRIMARY KEY AUTO_INCREMENT,
    customer_id VARCHAR(50),
    contact_type VARCHAR(20), -- 'email', 'phone', 'fax'
    contact_value VARCHAR(100),
    is_primary BOOLEAN DEFAULT FALSE,
    effective_date DATE,
    expiry_date DATE,
    FOREIGN KEY (customer_id) REFERENCES customer(customer_id)
);

-- 销售数据集市 - 星型模型
-- 中心事实表
CREATE TABLE mart_sales.fact_daily_sales (
    sale_date_sk INT,
    customer_sk BIGINT,
    product_sk BIGINT,
    store_sk BIGINT,
    promotion_sk BIGINT,
    
    -- 度量值
    sales_quantity INT,
    sales_amount DECIMAL(10,2),
    discount_amount DECIMAL(10,2),
    tax_amount DECIMAL(10,2),
    profit_amount DECIMAL(10,2),
    cost_amount DECIMAL(10,2),
    
    PRIMARY KEY (sale_date_sk, customer_sk, product_sk, store_sk)
);

-- 聚合事实表 - 月度汇总
CREATE TABLE mart_sales.fact_monthly_sales (
    sale_month_sk INT,
    customer_sk BIGINT,
    product_category_sk BIGINT,
    region_sk BIGINT,
    
    -- 聚合度量值
    total_sales_amount DECIMAL(15,2),
    total_sales_quantity INT,
    total_profit_amount DECIMAL(15,2),
    avg_sale_amount DECIMAL(10,2),
    customer_count INT,
    order_count INT,
    
    PRIMARY KEY (sale_month_sk, customer_sk, product_category_sk, region_sk)
);

-- 客户价值分析集市
CREATE TABLE mart_customer.fact_customer_value (
    customer_sk BIGINT PRIMARY KEY,
    analysis_date DATE,
    
    -- RFM分析指标
    recency_days INT,           -- 最近购买距今天数
    frequency_count INT,        -- 购买频次
    monetary_amount DECIMAL(12,2), -- 购买金额
    
    -- RFM评分
    recency_score INT,          -- 1-5分
    frequency_score INT,        -- 1-5分
    monetary_score INT,         -- 1-5分
    rfm_segment VARCHAR(20),    -- 'Champions', 'Loyal Customers', etc.
    
    -- 生命周期价值
    ltv_amount DECIMAL(15,2),
    predicted_ltv DECIMAL(15,2),
    
    -- 客户状态
    customer_status VARCHAR(20), -- 'active', 'at_risk', 'churned'
    churn_probability DECIMAL(5,4)
);

MERGE INTO target_table AS target
USING source_table AS source
ON target.id = source.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

-- ClickHouse实时表设计要点
CREATE TABLE realtime_orders (
    order_id String,
    customer_id String,
    order_amount Decimal(10, 2),
    order_datetime DateTime
) ENGINE = ReplacingMergeTree(order_datetime)  -- 去重引擎
  PARTITION BY toYYYYMM(order_datetime)         -- 按月分区
  ORDER BY (customer_id, order_datetime);       -- 排序键影响查询性能

# 数据治理成熟度评估框架
governance_maturity_model = {
    'level_1': {
        'name': '初始阶段',
        'characteristics': [
            '数据管理活动临时性、不一致',
            '缺乏数据标准和政策',
            '数据质量问题被动解决'
        ],
        'kpi': {
            'data_quality_issues': '高',
            'data_accessibility': '低',
            'governance_awareness': '很低'
        }
    },
    'level_2': {
        'name': '重复阶段',
        'characteristics': [
            '部分数据管理流程已建立',
            '基本数据标准已定义',
            '部分关键数据资产已识别'
        ],
        'kpi': {
            'data_quality_issues': '中高',
            'data_accessibility': '中低',
            'governance_awareness': '低'
        }
    },
    'level_3': {
        'name': '定义阶段',
        'characteristics': [
            '正式的数据治理组织已建立',
            '数据管理流程已标准化',
            '数据质量监控已实施'
        ],
        'kpi': {
            'data_quality_issues': '中',
            'data_accessibility': '中',
            'governance_awareness': '中'
        }
    },
    'level_4': {
        'name': '管理阶段',
        'characteristics': [
            '数据治理融入业务流程',
            '数据质量度量定量管理',
            '数据资产价值已量化'
        ],
        'kpi': {
            'data_quality_issues': '低',
            'data_accessibility': '高',
            'governance_awareness': '高'
        }
    },
    'level_5': {
        'name': '优化阶段',
        'characteristics': [
            '数据治理持续改进',
            '数据价值最大化',
            '数据创新驱动业务'
        ],
        'kpi': {
            'data_quality_issues': '很低',
            'data_accessibility': '很高',
            'governance_awareness': '很高'
        }
    }
}

-- 1. 完整性检查：检查关键字段是否为空
SELECT COUNT(*) as missing_count
FROM dwd_order.fact_order_detail
WHERE customer_sk IS NULL OR product_sk IS NULL OR order_date_sk IS NULL;

-- 2. 有效性检查：检查数字范围
SELECT COUNT(*) as invalid_amount
FROM dwd_order.fact_order_detail
WHERE quantity <= 0 OR total_amount <= 0;

-- 3. 一致性检查：子表与主表数量一致性
SELECT 'Orders with no items' as issue_type, 
       COUNT(DISTINCT a.order_id) as issue_count
FROM dwd_order.fact_order_detail a
LEFT JOIN dwd_order.fact_order_items b ON a.order_id = b.order_id
WHERE b.order_id IS NULL;

-- 4. 时效性检查：检查数据更新延迟
SELECT 
    DATEDIFF(CURRENT_DATE, MAX(dt)) as days_delay,
    CASE 
        WHEN DATEDIFF(CURRENT_DATE, MAX(dt)) > 1 THEN 'Alert'
        ELSE 'Normal'
    END as status
FROM dwd_order.fact_order_detail;

-- 5. 准确性检查：验证汇总值
SELECT 
    SUM(ABS(a.total_amount - b.total_amount)) / SUM(a.total_amount) as discrepancy_rate
FROM 
    dwd_order.fact_order_detail a
JOIN 
    dws_order.order_summary_daily b 
ON 
    a.order_date_sk = b.date_sk
    AND DATE(a.dt) = DATE(b.dt);

# 数据质量监控框架示例
class DataQualityMonitor:
    def __init__(self, config):
        self.config = config
        self.thresholds = config['thresholds']
        self.notification = config['notification']
        
    def run_quality_checks(self, table_name, dt):
        """运行数据质量检查"""
        results = {}
        
        # 运行完整性检查
        results['completeness'] = self._check_completeness(table_name, dt)
        
        # 运行准确性检查
        results['accuracy'] = self._check_accuracy(table_name, dt)
        
        # 运行一致性检查
        results['consistency'] = self._check_consistency(table_name, dt)
        
        # 评估质量分数
        quality_score = self._calculate_quality_score(results)
        
        # 记录质量结果
        self._log_quality_results(table_name, dt, results, quality_score)
        
        # 检查是否需要告警
        if self._should_alert(quality_score, results):
            self._send_alert(table_name, dt, results, quality_score)
            
        return {
            'table_name': table_name,
            'date': dt,
            'quality_score': quality_score,
            'details': results,
            'status': 'PASS' if quality_score > self.thresholds['overall'] else 'FAIL'
        }

-- 血缘关系示例查询
WITH RECURSIVE data_lineage AS (
    -- 选择起始点（如ADS层某表）
    SELECT 
        source_table,
        target_table,
        transform_logic,
        1 as level
    FROM metadata.table_lineage
    WHERE target_table = 'ads_user.user_tags'
    
    UNION ALL
    
    -- 递归查找上游依赖
    SELECT 
        l.source_table,
        l.target_table,
        l.transform_logic,
        dl.level + 1
    FROM metadata.table_lineage l
    JOIN data_lineage dl ON l.target_table = dl.source_table
    WHERE dl.level < 10  -- 限制递归深度
)
SELECT 
    LPAD(' ', (level-1)*2) || source_table as source,
    target_table as target,
    transform_logic,
    level
FROM data_lineage
ORDER BY level, source_table;

{
    "table_metadata": {
        "table_name": "fact_order_detail",
        "database_name": "dwd_order",
        "description": "订单明细事实表，包含订单交易的详细信息",
        "owner": "data_team",
        "create_time": "2023-01-15T10:30:00Z",
        "update_time": "2023-06-22T15:45:32Z",
        "table_type": "FACT",
        "layer": "DWD",
        "subject_area": "订单域",
        "tags": ["核心表", "订单", "交易"],
        "lifecycle": {
            "retention_period": "3 years",
            "archive_policy": "compress after 1 year"
        },
        "columns": [
            {
                "name": "order_detail_sk",
                "type": "BIGINT",
                "description": "订单明细代理键",
                "is_primary_key": true,
                "is_nullable": false,
                "business_name": "订单明细标识",
                "sensitivity": "normal"
            },
            {
                "name": "order_id",
                "type": "VARCHAR(50)",
                "description": "订单业务标识",
                "is_primary_key": false,
                "is_nullable": false,
                "business_name": "订单号",
                "sensitivity": "normal"
            }
        ],
        "lineage": {
            "upstream": ["ods_order.order_info"],
            "downstream": ["dws_order.order_summary_daily"]
        },
        "quality_rules": [
            {
                "rule_id": "R001",
                "rule_type": "completeness",
                "rule_definition": "order_id IS NOT NULL",
                "severity": "critical"
            }
        ]
    }
}

-- 数据脱敏SQL示例

-- 1. 掩码处理 - 信用卡号
SELECT 
    customer_id,
    CONCAT(SUBSTRING(credit_card, 1, 4), '-XXXX-XXXX-', SUBSTRING(credit_card, 13, 4)) AS masked_cc
FROM customer_data;

-- 2. 截断 - 邮箱地址
SELECT 
    customer_id,
    CONCAT(SUBSTRING(email, 1, 3), '***', SUBSTRING(email, INSTR(email, '@'))) AS truncated_email
FROM customer_data;

-- 3. 置换 - 手机号码
SELECT 
    customer_id,
    CONCAT(SUBSTRING(phone, 1, 3), '-', 
           MD5(CONCAT(phone, 'salt_key')), 
           SUBSTRING(phone, LENGTH(phone)-1, 2)) AS tokenized_phone
FROM customer_data;

-- 4. 数据泛化 - 年龄分组
SELECT 
    customer_id,
    CASE 
        WHEN age BETWEEN 18 AND 25 THEN '18-25'
        WHEN age BETWEEN 26 AND 35 THEN '26-35'
        WHEN age BETWEEN 36 AND 50 THEN '36-50'
        WHEN age > 50 THEN '50+'
        ELSE 'Unknown'
    END AS age_group
FROM customer_data;

# 审计日志分析框架
class DataAccessAudit:
    def __init__(self, config):
        self.config = config
        self.log_source = config['log_source']
        self.alert_thresholds = config['alert_thresholds']
        
    def analyze_access_patterns(self, time_window):
        """分析数据访问模式，检测异常"""
        # 获取审计日志
        logs = self._fetch_logs(time_window)
        
        # 计算用户访问统计
        user_stats = self._calculate_user_stats(logs)
        
        # 检测异常访问
        anomalies = self._detect_anomalies(user_stats)
        
        # 生成安全报告
        report = self._generate_report(user_stats, anomalies)
        
        # 发送警报
        if anomalies:
            self._send_alerts(anomalies)
        
        return report
    
    def _detect_anomalies(self, user_stats):
        """检测异常访问模式"""
        anomalies = []
        
        # 检查访问量异常
        for user, stats in user_stats.items():
            # 异常高频访问
            if stats['access_count'] > self.alert_thresholds['max_access_count']:
                anomalies.append({
                    'type': 'high_frequency_access',
                    'user': user,
                    'count': stats['access_count'],
                    'threshold': self.alert_thresholds['max_access_count']
                })
            
            # 异常时间访问
            if stats['off_hours_access'] > 0:
                anomalies.append({
                    'type': 'off_hours_access',
                    'user': user,
                    'count': stats['off_hours_access']
                })
            
            # 异常敏感数据访问
            if stats['sensitive_data_access'] > self.alert_thresholds['sensitive_data_access']:
                anomalies.append({
                    'type': 'sensitive_data_access',
                    'user': user,
                    'count': stats['sensitive_data_access'],
                    'threshold': self.alert_thresholds['sensitive_data_access']
                })
        
        return anomalies

-- 按日期范围分区
CREATE TABLE fact_orders (
    order_id STRING,
    customer_id STRING,
    order_date DATE,
    total_amount DECIMAL(10,2)
)
PARTITIONED BY (dt STRING)
STORED AS PARQUET;

-- 动态分区插入
INSERT OVERWRITE TABLE fact_orders PARTITION(dt)
SELECT 
    order_id,
    customer_id,
    order_date,
    total_amount,
    DATE_FORMAT(order_date, 'yyyy-MM-dd') as dt
FROM staging_orders;

-- 多级分区设计
CREATE TABLE fact_orders_multi (
    order_id STRING,
    customer_id STRING,
    order_date DATE,
    total_amount DECIMAL(10,2)
)
PARTITIONED BY (year INT, month INT, day INT)
STORED AS PARQUET;

-- 基于常用JOIN键分桶
CREATE TABLE fact_orders_bucketed (
    order_id STRING,
    customer_id STRING,
    order_date DATE,
    total_amount DECIMAL(10,2)
)
PARTITIONED BY (dt STRING)
CLUSTERED BY (customer_id) INTO 64 BUCKETS
STORED AS ORC;

-- 桶表JOIN优化
SELECT 
    o.order_id, c.customer_name, o.total_amount
FROM 
    fact_orders_bucketed o
JOIN 
    dim_customer_bucketed c ON o.customer_id = c.customer_id
WHERE 
    o.dt = '2023-12-25';

-- ORC压缩配置
CREATE TABLE compressed_table (
    id BIGINT,
    name STRING,
    value DOUBLE
)
STORED AS ORC
TBLPROPERTIES (
    "orc.compress" = "SNAPPY",
    "orc.create.index" = "true",
    "orc.bloom.filter.columns" = "id"
);

-- Parquet压缩配置
SET parquet.compression=SNAPPY;
SET parquet.page.size=1048576;
SET parquet.block.size=268435456;

-- 手动合并小文件
SET hive.merge.mapfiles = true;
SET hive.merge.mapredfiles = true;
SET hive.merge.smallfiles.avgsize = 256000000; -- 256MB
SET hive.merge.size.per.task = 256000000;

-- 定期合并小文件的存储过程
CREATE PROCEDURE merge_small_files(IN table_name STRING, IN partition_spec STRING)
BEGIN
    EXECUTE IMMEDIATE 'ALTER TABLE ' || table_name || ' PARTITION(' || partition_spec || ') CONCATENATE';
END;

-- 表设计时指定排序键
CREATE TABLE sorted_fact_orders (
    order_date DATE,
    customer_id STRING,
    product_id STRING,
    order_id STRING,
    quantity INT,
    unit_price DECIMAL(10,2),
    total_amount DECIMAL(10,2)
)
STORED AS PARQUET
PARTITIONED BY (dt STRING)
SORTED BY (customer_id, order_date);

-- Delta Lake Z-Order优化
OPTIMIZE sales.sorted_fact_orders
ZORDER BY (customer_id, order_date);

-- 避免使用SELECT *，只查询需要的列
-- 不优化版本
SELECT * FROM fact_orders WHERE order_date > '2023-01-01';

-- 优化版本
SELECT order_id, customer_id, order_date, total_amount 
FROM fact_orders 
WHERE order_date > '2023-01-01';

-- 使用适当的JOIN策略
-- 小表广播JOIN
SELECT /*+ BROADCAST(c) */ 
    o.order_id, c.customer_name, o.total_amount
FROM 
    fact_orders o
JOIN 
    dim_customer c ON o.customer_id = c.customer_id
WHERE 
    o.dt = '2023-12-25';

-- 谓词下推和列裁剪
SELECT 
    c.customer_name,
    SUM(o.total_amount) as total_sales
FROM 
    fact_orders o
JOIN 
    dim_customer c ON o.customer_id = c.customer_id
WHERE 
    o.order_date BETWEEN '2023-01-01' AND '2023-12-31'
GROUP BY 
    c.customer_name;

-- 使用窗口函数替代自连接
-- 不优化版本
SELECT 
    a.customer_id,
    a.order_id,
    a.order_date,
    a.total_amount,
    (SELECT SUM(total_amount) 
     FROM fact_orders b
     WHERE b.customer_id = a.customer_id 
     AND b.order_date <= a.order_date) as cumulative_amount
FROM fact_orders a;

-- 优化版本
SELECT 
    customer_id,
    order_id,
    order_date,
    total_amount,
    SUM(total_amount) OVER (
        PARTITION BY customer_id 
        ORDER BY order_date
        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) as cumulative_amount
FROM fact_orders;

-- 创建物化视图
CREATE MATERIALIZED VIEW mv_daily_sales
REFRESH COMPLETE ON DEMAND
ENABLE QUERY REWRITE
AS
SELECT 
    DATE(order_date) as sale_date,
    product_id,
    SUM(quantity) as total_quantity,
    SUM(total_amount) as total_sales
FROM fact_orders
GROUP BY DATE(order_date), product_id;

-- 使用物化视图
SELECT 
    sale_date, 
    SUM(total_sales) as daily_sales
FROM mv_daily_sales
WHERE sale_date BETWEEN '2023-01-01' AND '2023-01-31'
GROUP BY sale_date;

# Spark SQL参数优化示例
spark-sql \
  --conf spark.sql.adaptive.enabled=true \
  --conf spark.sql.adaptive.skewJoin.enabled=true \
  --conf spark.sql.adaptive.coalescePartitions.enabled=true \
  --conf spark.sql.shuffle.partitions=200 \
  --conf spark.executor.memory=8g \
  --conf spark.executor.cores=4 \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.sql.autoBroadcastJoinThreshold=100m \
  --conf spark.memory.fraction=0.8

-- Oracle提示符示例
SELECT /*+ INDEX(customers cust_idx) */ 
    customer_name, city
FROM 
    customers
WHERE 
    region_id = 3;

-- Spark SQL提示符示例
SELECT /*+ BROADCAST(dim_date) */ 
    d.date_desc, SUM(f.sales_amount) as total_sales
FROM 
    fact_sales f
JOIN 
    dim_date d ON f.date_id = d.date_id
GROUP BY 
    d.date_desc;

# YARN资源队列配置示例
yarn.scheduler.capacity.root.queues: default,production,development
yarn.scheduler.capacity.root.production.capacity: 60
yarn.scheduler.capacity.root.development.capacity: 20
yarn.scheduler.capacity.root.default.capacity: 20

# 资源管理器动态分配配置
spark.dynamicAllocation.enabled: true
spark.dynamicAllocation.minExecutors: 5
spark.dynamicAllocation.maxExecutors: 100
spark.dynamicAllocation.schedulerBacklogTimeout: 60s
spark.dynamicAllocation.executorIdleTimeout: 120s

class QueryRouter:
    def __init__(self):
        self.engines = {
            'interactive': {
                'engine': 'presto',
                'url': 'jdbc:presto://presto-coordinator:8080',
                'max_execution_time': 60,  # 60秒
                'max_concurrent_queries': 50
            },
            'reporting': {
                'engine': 'spark',
                'url': 'jdbc:hive2://spark-thriftserver:10000',
                'max_execution_time': 300,  # 5分钟
                'max_concurrent_queries': 20
            },
            'batch': {
                'engine': 'hive',
                'url': 'jdbc:hive2://hiveserver2:10000',
                'max_execution_time': 3600,  # 1小时
                'max_concurrent_queries': 10
            }
        }
        self.metrics = {
            'engine': {
                'presto': {'load': 0, 'availability': 1.0},
                'spark': {'load': 0, 'availability': 1.0},
                'hive': {'load': 0, 'availability': 1.0}
            }
        }
    
    def route_query(self, query, user, priority):
        # 分析查询复杂度
        complexity = self._analyze_query_complexity(query)
        
        # 根据查询类型、用户角色、优先级选择合适的引擎
        if complexity < 3 and user.role in ['analyst', 'dashboard']:
            engine_type = 'interactive'
        elif complexity >= 3 and complexity < 7:
            engine_type = 'reporting'
        else:
            engine_type = 'batch'
            
        # 考虑当前负载和可用性
        selected_engine = self.engines[engine_type]
        engine_name = selected_engine['engine']
        
        # 检查是否超出并发限制
        if self.metrics['engine'][engine_name]['load'] >= selected_engine['max_concurrent_queries']:
            # 尝试故障转移
            if engine_type == 'interactive' and self.metrics['engine']['spark']['load'] < self.engines['reporting']['max_concurrent_queries']:
                engine_type = 'reporting'
                selected_engine = self.engines[engine_type]
                engine_name = selected_engine['engine']
            else:
                # 放入队列等待
                return self._enqueue_query(query, engine_type, priority)
        
        # 增加负载计数
        self.metrics['engine'][engine_name]['load'] += 1
        
        # 返回引擎连接信息
        return {
            'engine': engine_name,
            'connection_url': selected_engine['url'],
            'timeout': selected_engine['max_execution_time'],
            'query_id': self._generate_query_id()
        }

# Kubernetes自动伸缩配置示例
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: presto-worker-autoscaler
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: presto-worker
  minReplicas: 5
  maxReplicas: 50
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 75

<!-- HDFS核心配置 -->
<configuration>
    <!-- 副本数 -->
    <property>
        <name>dfs.replication</name>
        <value>3</value>
    </property>
    
    <!-- 块大小 -->
    <property>
        <name>dfs.blocksize</name>
        <value>134217728</value> <!-- 128MB -->
    </property>
    
    <!-- 心跳间隔 -->
    <property>
        <name>dfs.heartbeat.interval</name>
        <value>3</value> <!-- 3秒 -->
    </property>
    
    <!-- 启用短路读取 -->
    <property>
        <name>dfs.client.read.shortcircuit</name>
        <value>true</value>
    </property>
    
    <!-- 启用HDFS缓存 -->
    <property>
        <name>dfs.datanode.max.locked.memory</name>
        <value>4294967296</value> <!-- 4GB -->
    </property>
</configuration>

# 使用boto3操作S3
import boto3

class DataLakeStorage:
    def __init__(self, bucket_name, region='us-east-1'):
        self.s3 = boto3.client('s3', region_name=region)
        self.bucket = bucket_name
        
    def upload_data(self, local_file, s3_key, storage_class='STANDARD'):
        """上传数据到S3"""
        extra_args = {
            'StorageClass': storage_class,
            'ContentType': self._get_content_type(local_file)
        }
        
        self.s3.upload_file(
            local_file, self.bucket, s3_key, 
            ExtraArgs=extra_args
        )
        
    def create_lifecycle_policy(self):
        """设置生命周期策略"""
        lifecycle_config = {
            'Rules': [
                {
                    'ID': 'archive-old-data',
                    'Status': 'Enabled',
                    'Prefix': 'raw-data/',
                    'Transitions': [
                        {
                            'Days': 30,
                            'StorageClass': 'STANDARD_IA'
                        },
                        {
                            'Days': 90,
                            'StorageClass': 'GLACIER'
                        }
                    ],
                    'Expiration': {
                        'Days': 365
                    }
                },
                {
                    'ID': 'delete-temp-data',
                    'Status': 'Enabled',
                    'Prefix': 'tmp/',
                    'Expiration': {
                        'Days': 1
                    }
                }
            ]
        }
        
        self.s3.put_bucket_lifecycle_configuration(
            Bucket=self.bucket,
            LifecycleConfiguration=lifecycle_config
        )

-- PostgreSQL数据仓库优化配置
ALTER SYSTEM SET 
    -- 内存配置
    shared_buffers = '8GB',            -- 服务器内存的25%
    work_mem = '256MB',                -- 复杂排序和哈希所需内存
    maintenance_work_mem = '1GB',      -- 维护任务内存
    
    -- 查询优化
    random_page_cost = 1.1,            -- SSD存储假设随机访问成本低
    effective_cache_size = '24GB',     -- 假设系统缓存可用大小
    
    -- 并行查询
    max_worker_processes = 16,         -- 并行进程数上限
    max_parallel_workers_per_gather = 4, -- 每个查询的并行工作进程
    
    -- 写入性能
    wal_buffers = '16MB',              -- WAL缓冲区大小
    checkpoint_timeout = '15min',      -- 检查点间隔时间
    
    -- 分区表设置
    enable_partition_pruning = on,     -- 启用分区裁剪
    constraint_exclusion = on;         -- 启用约束排除

// Spark SQL数据仓库ETL示例
import org.apache.spark.sql.{SparkSession, SaveMode, DataFrame}
import org.apache.spark.sql.functions._

object DataWarehouseETL {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("DataWarehouseETL")
      .config("spark.sql.warehouse.dir", "s3://data-warehouse/")
      .config("spark.sql.adaptive.enabled", "true")
      .config("spark.sql.adaptive.skewJoin.enabled", "true")
      .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
      .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
      .enableHiveSupport()
      .getOrCreate()
    
    // 提取源数据
    val ordersDf = extractOrderData(spark)
    
    // 数据转换
    val transformedDf = transformOrderData(ordersDf)
    
    // 加载到目标表
    loadData(transformedDf, "dwd_order.fact_order_detail")
  }
  
  def extractOrderData(spark: SparkSession): DataFrame = {
    // 从ODS层读取数据
    spark.sql("""
      SELECT 
        order_id, customer_id, order_date, order_status,
        payment_method, shipping_address, order_items, total_amount
      FROM ods_order.order_info
      WHERE dt = date_sub(current_date, 1)
    """)
  }
  
  def transformOrderData(ordersDf: DataFrame): DataFrame = {
    // 数据清洗与转换
    import spark.implicits._
    
    ordersDf
      .withColumn("order_date_sk", 
                  date_format(col("order_date"), "yyyyMMdd").cast("int"))
      .withColumn("order_item", explode(col("order_items")))
      .withColumn("product_id", col("order_item.product_id"))
      .withColumn("quantity", col("order_item.quantity"))
      .withColumn("unit_price", col("order_item.price"))
      .withColumn("subtotal_amount", col("quantity") * col("unit_price"))
      .drop("order_items", "order_item")
  }
  
  def loadData(df: DataFrame, targetTable: String): Unit = {
    // 写入目标表
    df.write
      .format("delta")
      .mode(SaveMode.Append)
      .partitionBy("order_date_sk")
      .saveAsTable(targetTable)
  }
}

// Flink实时数仓处理示例
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;

public class RealtimeDataWarehouse {
    public static void main(String[] args) throws Exception {
        // 设置流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60000); // 每分钟做一次Checkpoint
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
        
        // 创建Table API环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
            .inStreamingMode()
            .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        
        // 定义Kafka源表
        tableEnv.executeSql("""
            CREATE TABLE orders_source (
                order_id STRING,
                customer_id STRING,
                order_time TIMESTAMP(3),
                order_amount DECIMAL(10, 2),
                products ARRAY<ROW(product_id STRING, quantity INT, price DECIMAL(10,2))>,
                order_status STRING,
                proctime AS PROCTIME()
            ) WITH (
                'connector' = 'kafka',
                'topic' = 'orders',
                'properties.bootstrap.servers' = 'kafka:9092',
                'properties.group.id' = 'realtime-dw',
                'format' = 'json',
                'scan.startup.mode' = 'latest-offset'
            )
        """);
        
        // 定义维度表（通过CDC保持更新）
        tableEnv.executeSql("""
            CREATE TABLE dim_customer (
                customer_id STRING,
                customer_name STRING,
                customer_type STRING,
                city STRING,
                region STRING,
                PRIMARY KEY (customer_id) NOT ENFORCED
            ) WITH (
                'connector' = 'jdbc',
                'url' = 'jdbc:mysql://mysql:3306/dw',
                'table-name' = 'dim_customer',
                'username' = 'dw_user',
                'password' = 'dw_password',
                'lookup.cache.max-rows' = '10000',
                'lookup.cache.ttl' = '1 hour'
            )
        """);
        
        // 定义目标表
        tableEnv.executeSql("""
            CREATE TABLE enriched_orders (
                order_id STRING,
                order_time TIMESTAMP(3),
                customer_id STRING,
                customer_name STRING,
                customer_type STRING,
                city STRING,
                region STRING,
                product_id STRING,
                quantity INT,
                unit_price DECIMAL(10,2),
                product_amount DECIMAL(10,2),
                PRIMARY KEY (order_id, product_id) NOT ENFORCED
            ) WITH (
                'connector' = 'upsert-kafka',
                'topic' = 'enriched_orders',
                'properties.bootstrap.servers' = 'kafka:9092',
                'key.format' = 'json',
                'value.format' = 'json'
            )
        """);
        
        // 执行实时ETL
        tableEnv.executeSql("""
            INSERT INTO enriched_orders
            SELECT
                o.order_id,
                o.order_time,
                o.customer_id,
                c.customer_name,
                c.customer_type,
                c.city,
                c.region,
                p.product_id,
                p.quantity,
                p.price as unit_price,
                p.quantity * p.price as product_amount
            FROM orders_source o,
                 LATERAL TABLE(o.products) AS p,
                 dim_customer FOR SYSTEM_TIME AS OF o.proctime as c
            WHERE o.customer_id = c.customer_id
        """);
        
        env.execute("Realtime Data Warehouse");
    }
}