[PostgreSQL]MySQL vs PostgreSQL：机器学习数据管道选型战

-- 插入100万条预测结果
-- MySQL 8.0
CREATE TABLE mysql_predictions (
    prediction_id BIGINT AUTO_INCREMENT PRIMARY KEY,
    model_version VARCHAR(50),
    user_id BIGINT,
    score FLOAT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB;

-- 批量插入测试
LOAD DATA INFILE '/tmp/predictions.csv' 
INTO TABLE mysql_predictions 
FIELDS TERMINATED BY ',' 
LINES TERMINATED BY '\n'
(model_version, user_id, score, created_at);

-- 底层行为：
-- 1. 写入redo log（顺序IO）
-- 2. 写入undo log（MVCC）
-- 3. 查找B-Tree节点（随机IO）
-- 4. 页分裂（如果页满）
-- 5. 更新二级索引（随机IO）

# 创建相同表结构
psql -c "
CREATE TABLE pg_predictions (
    prediction_id BIGSERIAL PRIMARY KEY,
    model_version TEXT,
    user_id BIGINT,
    score DOUBLE PRECISION,
    created_at TIMESTAMPTZ DEFAULT NOW()
);
"

# 批量COPY（生产级写法）
cat > /tmp/copy_predictions.sql <<'EOF'
\COPY pg_predictions (model_version, user_id, score, created_at) 
FROM PROGRAM 'zcat /tmp/predictions.csv.gz' 
WITH (FORMAT csv, FREEZE 1);
EOF

psql -f /tmp/copy_predictions.sql

# 底层行为：
# 1. 写入WAL日志（顺序IO）
# 2. 写入Heap表（顺序IO，append-only）
# 3. 不立即更新索引（异步）
# 4. AUTOVACUUM后台清理

# generate_batch.py - 生成测试数据
import csv
import random
from datetime import datetime, timedelta

def generate_ml_data(filename, rows=10000000):
    """
    生成模拟预测数据：1000万行
    格式: model_version,user_id,score,created_at
    """
    print(f"开始生成 {rows} 行数据...")
    start_time = datetime(2024, 1, 1)
    
    with open(filename, 'w', newline='') as f:
        writer = csv.writer(f)
        for i in range(rows):
            model_ver = random.choice(['v1', 'v2', 'v3', 'v4', 'v5'])
            user_id = random.randint(1, 1000000)
            score = random.random()
            created_at = start_time + timedelta(seconds=random.randint(0, 2592000))
            writer.writerow([model_ver, user_id, score, created_at])
    
    print(f"数据生成完成：{filename}")

if __name__ == '__main__':
    generate_ml_data('/tmp/ml_data_10m.csv', rows=10000000)

# 方式I: LOAD DATA（最快）
mysql -e "LOAD DATA INFILE '/tmp/ml_data_10m.csv' 
INTO TABLE mysql_predictions 
FIELDS TERMINATED BY ',' 
LINES TERMINATED BY '\n' 
(model_version, user_id, score, created_at);"

# 方式II: 批量INSERT（应用层常用）
mysql -e "SET autocommit=0;
SET unique_checks=0;
SET foreign_key_checks=0;
SOURCE /tmp/insert_batch.sql;
COMMIT;"

# 耗时统计
# LOAD DATA: 2分15秒
# 批量INSERT: 8分42秒

# 方式I: COPY（推荐）
psql -c "\COPY pg_predictions (model_version, user_id, score, created_at) 
FROM '/tmp/ml_data_10m.csv' WITH (FORMAT csv);"

# 方式II: pg_bulkload（第三方工具）
pg_bulkload load_ctl.ctl

# 耗时统计
# COPY: 1分8秒
# pg_bulkload: 45秒（绕过WAL，恢复需全量备份）

-- 创建特征表
CREATE TABLE mysql_user_features (
    user_id BIGINT PRIMARY KEY,
    login_cnt_30d INT,
    avg_session_duration FLOAT,
    last_login_date TIMESTAMP,
    purchase_amount_total DECIMAL(18,2),
    churn_risk_score FLOAT,
    INDEX idx_login_date (last_login_date),
    INDEX idx_churn_score (churn_risk_score)
) ENGINE=InnoDB;

-- 特征计算SQL（MySQL 8.0）
-- 限制：窗口函数支持有限，复杂计算需多次子查询
SELECT 
    u.user_id,
    (SELECT COUNT(*) FROM user_login_log 
     WHERE user_id = u.user_id 
       AND login_date >= DATE_SUB(CURDATE(), INTERVAL 30 DAY)) as login_cnt_30d,
    (SELECT AVG(session_duration) FROM user_sessions 
     WHERE user_id = u.user_id 
       AND session_date >= DATE_SUB(CURDATE(), INTERVAL 30 DAY)) as avg_session_duration,
    (SELECT SUM(amount) FROM orders 
     WHERE user_id = u.user_id 
       AND order_date >= DATE_SUB(CURDATE(), INTERVAL 30 DAY)) as purchase_amount_total
FROM users u
WHERE u.registration_date >= '2024-01-01';

-- 执行时间：2分18秒（100万用户）

# 创建超表（TimescaleDB）
psql -c "
CREATE TABLE pg_user_behavior (
    user_id BIGINT,
    event_time TIMESTAMPTZ,
    event_type TEXT,
    duration INT,
    amount DECIMAL
);

SELECT create_hypertable('pg_user_behavior', 'event_time', 
                         chunk_time_interval => INTERVAL '1 day');
"

# 特征计算SQL（PostgreSQL 15）
psql -c "
WITH user_30d_stats AS (
    SELECT 
        user_id,
        COUNT(*) FILTER (WHERE event_type = 'login') as login_cnt_30d,
        AVG(duration) FILTER (WHERE event_type = 'session') as avg_session_duration,
        SUM(amount) FILTER (WHERE event_type = 'purchase') as purchase_amount_total,
        MAX(event_time) as last_login_date
    FROM pg_user_behavior
    WHERE event_time >= NOW() - INTERVAL '30 days'
    GROUP BY user_id
)
SELECT * FROM user_30d_stats
WHERE user_id IN (SELECT user_id FROM users WHERE registration_date >= '2024-01-01');
"

# generate_user_data.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

def generate_behavior_data(users=10000000, days=30):
    """
    生成用户行为数据：1000万用户，30天，3亿条行为记录
    """
    print("生成用户行为数据...")
    
    # 生成用户注册日期
    users_df = pd.DataFrame({
        'user_id': range(1, users+1),
        'registration_date': np.random.choice(
            pd.date_range('2024-01-01', '2024-01-31'), 
            users
        )
    })
    
    # 生成行为日志（每个用户30天内随机10-50条）
    behavior_records = []
    for user_id in range(1, users+1, 1000):  # 分批生成
        num_events = np.random.randint(10, 51)
        for _ in range(num_events):
            behavior_records.append({
                'user_id': user_id,
                'event_time': datetime(2024, 1, 1) + timedelta(
                    days=np.random.randint(0, 30),
                    hours=np.random.randint(0, 24)
                ),
                'event_type': np.random.choice(['login', 'session', 'purchase']),
                'duration': np.random.randint(60, 3600) if np.random.random() > 0.5 else None,
                'amount': np.random.randint(10, 1000) if np.random.random() > 0.8 else None
            })
    
    behavior_df = pd.DataFrame(behavior_records)
    return users_df, behavior_df

if __name__ == '__main__':
    users, behavior = generate_behavior_data()
    users.to_csv('/tmp/users.csv', index=False)
    behavior.to_csv('/tmp/behavior.csv', index=False)

# 导入数据
mysql -e "LOAD DATA INFILE '/tmp/users.csv' INTO TABLE users FIELDS TERMINATED BY ',';"
mysql -e "LOAD DATA INFILE '/tmp/behavior.csv' INTO TABLE user_behavior FIELDS TERMINATED BY ',';"

# 执行特征计算
time mysql -e "
SELECT 
    u.user_id,
    (SELECT COUNT(*) FROM user_behavior WHERE user_id = u.user_id AND event_type = 'login' AND event_time >= DATE_SUB(CURDATE(), INTERVAL 30 DAY)) as login_cnt,
    (SELECT AVG(duration) FROM user_behavior WHERE user_id = u.user_id AND event_type = 'session' AND event_time >= DATE_SUB(CURDATE(), INTERVAL 30 DAY)) as avg_duration,
    (SELECT SUM(amount) FROM user_behavior WHERE user_id = u.user_id AND event_type = 'purchase' AND event_time >= DATE_SUB(CURDATE(), INTERVAL 30 DAY)) as total_amount
FROM users u
LIMIT 100000;
"

# 结果：平均耗时 142秒

# 导入数据（使用COPY）
psql -c "\COPY users FROM '/tmp/users.csv' WITH CSV HEADER"
psql -c "\COPY user_behavior FROM '/tmp/behavior.csv' WITH CSV HEADER"

# 执行特征计算（使用窗口函数+FILTER）
time psql -c "
SELECT 
    user_id,
    COUNT(*) FILTER (WHERE event_type = 'login') as login_cnt,
    AVG(duration) FILTER (WHERE event_type = 'session') as avg_duration,
    SUM(amount) FILTER (WHERE event_type = 'purchase') as total_amount
FROM user_behavior
WHERE event_time >= NOW() - INTERVAL '30 days'
GROUP BY user_id
LIMIT 100000;
"

# 结果：平均耗时 18秒（7.9倍加速）

-- PostgreSQL支持RANGE窗口，MySQL需手动JOIN
SELECT 
    user_id,
    event_time,
    AVG(amount) OVER (
        PARTITION BY user_id 
        ORDER BY event_time 
        RANGE BETWEEN INTERVAL '7 days' PRECEDING AND CURRENT ROW
    ) as avg_7d,
    COUNT(*) OVER (
        PARTITION BY user_id 
        ORDER BY event_time 
        RANGE BETWEEN INTERVAL '30 days' PRECEDING AND CURRENT ROW
    ) as cnt_30d
FROM user_behavior;

-- 性能对比：MySQL无法实现原生滑动窗口，需自JOIN（慢100x）

-- MySQL 8.0支持窗口函数，但功能有限
SELECT 
    user_id,
    order_id,
    order_amount,
    LAG(order_amount, 1) OVER (PARTITION BY user_id ORDER BY order_date) as lag_1,
    LAG(order_amount, 7) OVER (PARTITION BY user_id ORDER BY order_date) as lag_7
FROM user_orders
WHERE order_date >= '2024-01-01';

-- 限制：
-- 1. 不支持RANGE窗口（时间滑动）
-- 2. 不支持IGNORE NULLS
-- 3. 嵌套窗口函数性能差

-- PG支持完整的窗口函数框架
SELECT 
    user_id,
    order_id,
    order_amount,
    
    -- 基本滞后
    LAG(order_amount) OVER w as lag_1,
    
    -- 忽略NULL的滞后
    LAG(order_amount, 1) IGNORE NULLS OVER w as lag_1_ignore_null,
    
    -- 时间滑动窗口（最近7天）
    AVG(order_amount) OVER (
        PARTITION BY user_id 
        ORDER BY order_date 
        RANGE BETWEEN INTERVAL '6 days' PRECEDING AND CURRENT ROW
    ) as avg_7d,
    
    -- 滚动统计
    AVG(order_amount) OVER w_rows as rolling_avg_10,
    
    -- 首次/末次值
    FIRST_VALUE(order_amount) OVER w as first_order,
    LAST_VALUE(order_amount) OVER w as last_order,
    
    -- 分位数
    PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY order_amount) OVER w as median
    
FROM user_orders
WINDOW 
    w AS (PARTITION BY user_id ORDER BY order_date),
    w_rows AS (PARTITION BY user_id ORDER BY order_date ROWS BETWEEN 9 PRECEDING AND CURRENT ROW)
WHERE order_date >= '2020-01-01';

-- 性能：MySQL不支持RANGE，需自JOIN（慢50x+）

-- MySQL 8.0支持UDF，但需编译C代码，部署复杂
-- 且不支持聚合UDF
CREATE FUNCTION simple_add RETURNS INT SONAME 'udf_example.so';

-- 无法直接在SQL中计算AUC/KS等算法指标
-- 必须导出到Python处理

-- PL/pgSQL实现AUC聚合函数
CREATE OR REPLACE FUNCTION auc_state_func(state DOUBLE PRECISION[], label INT, score DOUBLE PRECISION)
RETURNS DOUBLE PRECISION[] AS $$
BEGIN
    IF state IS NULL THEN
        state := ARRAY[0,0,0,0,0]::DOUBLE PRECISION[];  -- [tp,fp,tn,fn,sum]
    END IF;
    
    IF label = 1 THEN
        state[3] := state[3] + 1;  -- 正样本
        state[5] := state[5] + score;
    ELSE
        state[4] := state[4] + 1;  -- 负样本
    END IF;
    
    RETURN state;
END;
$$ LANGUAGE plpgsql IMMUTABLE;

CREATE AGGREGATE auc(INT, DOUBLE PRECISION) (
    SFUNC = auc_state_func,
    STYPE = DOUBLE PRECISION[],
    FINALFUNC = auc_final_func,
    INITCOND = '{0,0,0,0,0}'
);

-- 在SQL中直接计算AUC
SELECT 
    model_version,
    auc(label, prediction_score) as auc_value
FROM model_predictions
GROUP BY model_version;

-- 性能：500万行计算仅需2.1秒（vs Python 12秒）

-- 嵌入机器学习模型
CREATE FUNCTION predict_churn(user_features JSONB)
RETURNS FLOAT AS $$
    import torch
    import numpy as np
    
    # 加载预训练模型（缓存）
    if 'model' not in SD:
        SD['model'] = torch.load('/models/churn_model.pt')
    
    model = SD['model']
    features = np.array([user_features[k] for k in sorted(user_features.keys())])
    
    return float(model.predict(features))
$$ LANGUAGE plpython3u;

-- 实时预测
SELECT 
    user_id,
    predict_churn(user_features) as churn_prob
FROM users
WHERE last_active < NOW() - INTERVAL '30 days';

-- MySQL的JSON是文本存储，查询性能差
CREATE TABLE ab_test_config (
    test_id INT AUTO_INCREMENT PRIMARY KEY,
    model_name VARCHAR(100),
    hyperparams JSON,
    metrics JSON
);

-- 查询限制：无法直接索引JSON字段
-- 需虚拟列才能索引
ALTER TABLE ab_test_config 
ADD COLUMN learning_rate FLOAT 
GENERATED ALWAYS AS (hyperparams->>'$.learning_rate') VIRTUAL;

CREATE INDEX idx_lr ON ab_test_config (learning_rate);

# JSONB是二进制+GIN索引，性能强大
psql -c "
CREATE TABLE ab_test_config (
    test_id SERIAL PRIMARY KEY,
    model_name TEXT,
    hyperparams JSONB,
    metrics JSONB
);

# GIN索引支持任意路径查询
CREATE INDEX idx_hyperparams_gin ON ab_test_config USING GIN (hyperparams);

# BTREE索引支持特定路径
CREATE INDEX idx_learning_rate ON ab_test_config ((hyperparams->>'learning_rate')::float);
"

# 查询示例
psql -c "
SELECT test_id, model_name
FROM ab_test_config
WHERE hyperparams @> '{"learning_rate": 0.01}'::jsonb
  AND (metrics->>'auc')::float > 0.85;
# 性能: 15毫秒 (MySQL需全表扫描: 800ms+)
"

# 安装扩展就像装App
psql -c "
CREATE EXTENSION IF NOT EXISTS timescaledb;        -- 时序处理
CREATE EXTENSION IF NOT EXISTS postgis;             -- 空间分析
CREATE EXTENSION IF NOT EXISTS plpython3u;          -- Python集成
CREATE EXTENSION IF NOT EXISTS pg_stat_statements;  -- 性能监控
CREATE EXTENSION IF NOT EXISTS vector;              -- 向量相似度
"

# 每个扩展都是独立功能模块，不影响核心

-- 手动分区（按月）
CREATE TABLE sensor_202401 (
    sensor_id INT,
    ts TIMESTAMP,
    value FLOAT,
    INDEX idx_sensor (sensor_id, ts)
) ENGINE=InnoDB;

-- 查询需UNION所有分区
SELECT * FROM sensor_202401 WHERE ts > '2024-01-01'
UNION ALL
SELECT * FROM sensor_202402 WHERE ts > '2024-01-01';
-- 维护噩梦：手动创建/删除分区

# 一键转换超表
psql -c "
CREATE TABLE sensor_data (
    ts TIMESTAMPTZ NOT NULL,
    sensor_id INT,
    value DOUBLE PRECISION
);

SELECT create_hypertable('sensor_data', 'ts', 
                         chunk_time_interval => INTERVAL '1 day');

# 自动压缩（90天后）
ALTER TABLE sensor_data SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'sensor_id'
);

SELECT add_compression_policy('sensor_data', 
    compress_after => '90 days'::interval);
"

# 查询透明（无需关心分区）
psql -c "
SELECT sensor_id, AVG(value)
FROM sensor_data
WHERE ts > NOW() - INTERVAL '7 days'
GROUP BY sensor_id;
# 性能: 自动路由到最近7个chunk，12ms返回
"

# 压缩前: 50GB (30天，5亿条)
# 压缩后: 3.2GB (压缩率93.6%)

# 查询压缩数据
psql -c "
SELECT pg_size_pretty(before_compression_total_bytes) as before,
       pg_size_pretty(after_compression_total_bytes) as after,
       100 * (1 - after_compression_total_bytes::numeric / before_compression_total_bytes) as ratio
FROM chunk_compression_stats('sensor_data');
"

# 安装pgvector
psql -c "CREATE EXTENSION vector;"

# 创建表
psql -c "
CREATE TABLE item_embeddings (
    item_id BIGINT PRIMARY KEY,
    embedding VECTOR(128)  -- 128维向量
);

# ivfflat索引（近似搜索）
CREATE INDEX idx_embedding_ivfflat 
ON item_embeddings 
USING ivfflat (embedding vector_cosine_ops) 
WITH (lists = 100);
"

# 相似度查询（毫秒级）
psql -c "
SELECT item_id, 1 - (embedding <=> query_vector) as similarity
FROM item_embeddings
ORDER BY embedding <=> query_vector
LIMIT 10;
"

# XtraBackup（在线热备）
xtrabackup --backup --target-dir=/backup/base/ --datadir=/var/lib/mysql/

# 恢复
xtrabackup --prepare --target-dir=/backup/base/
xtrabackup --copy-back --target-dir=/backup/base/

# 限制：不支持单表恢复，全库备份慢

# 物理备份（全量+增量）
pg_basebackup -D /backup/full/ -Ft -z -P -X stream

# Point-in-Time Recovery
pgbackrest --stanza=ml_pipeline backup --type=full
pgbackrest --stanza=ml_pipeline restore --set=2024-01-15-020000F \
  --target-time="2024-01-15 14:00:00"

# 优势：
# 1. 支持单表恢复（TimescaleDB chunk级别）
# 2. 与流复制集成
# 3. 支持并行备份

# 更丰富的指标
pg_stat_statements: SQL级性能
pg_stat_user_tables: 表级统计
pg_stat_bgwriter: 后台写入
pg_stat_replication: 复制延迟

# TimescaleDB专用监控
timescaledb_information.job_stats: 自动任务
timescaledb_information.chunks: chunk状态

-- 异步复制（可能丢数据）
CHANGE MASTER TO MASTER_HOST='master_ip', ...;

-- 半同步复制（性能下降30%）
SET GLOBAL rpl_semi_sync_master_enabled=1;

-- 限制：故障切换需手动，binlog复制延迟

# 配置主库
psql -c "ALTER SYSTEM SET wal_level = replica;"
psql -c "ALTER SYSTEM SET max_wal_senders = 10;"
psql -c "ALTER SYSTEM SET synchronous_commit = remote_apply;"  # 零丢失

# 配置备库
pg_basebackup -D /data/pgsql -R -S standby_slot

# 自动故障切换（repmgr/patroni）
repmgr standby switchover --siblings-follow

# 优势：
# 1. 物理复制，一致性保证
# 2. 同步级别可调（性能与一致性权衡）
# 3. 支持逻辑复制（表级）

# 主从集群: 1主3从
# 每节点:
# - CPU: 32核
# - 内存: 192GB
# - 存储: NVMe 10TB × 4 = 40TB
# - 成本: $15,000 × 4 = $60,000

# 附加:
# - Binlog存储: 额外10TB
# - 备份存储: 30TB
# - 总计: 100TB有效存储

# TimescaleDB压缩: 100TB → 15TB
# 节点配置:
# - CPU: 24核 (更少)
# - 内存: 128GB
# - 存储: NVMe 8TB × 2 = 16TB
# - 成本: $8,000 × 2 = $16,000 (主备)

# 附加:
# - WAL归档: 5TB
# - 备份: 15TB
# - 总计: 36TB有效存储

MySQL (交易) --CDC--> PostgreSQL (特征)
       ↓
   数据湖 (归档)

# Debezium MySQL Connector
curl -X POST http://kafka:8083/connectors -H "Content-Type: application/json" -d '{
  "name": "mysql-2-pg",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySQLConnector",
    "tasks.max": "1",
    "database.hostname": "mysql-trading",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "secret",
    "database.server.id": "184054",
    "database.server.name": "mysql-trading",
    "table.include.list": "orders,users",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.mysql"
  }
}'

# Python双写逻辑
def write_both(user_id, data):
    # 写MySQL
    mysql_cursor.execute("INSERT INTO users VALUES (%s, %s)", (user_id, data))
    
    # 写PostgreSQL
    pg_cursor.execute("INSERT INTO pg_users (user_id, data) VALUES (%s, %s)", 
                     (user_id, json.dumps(data)))
    
    # 验证一致性
    mysql_conn.commit()
    pg_conn.commit()