-- MySQL 8.0 行为日志表设计CREATE TABLE user_behavior_mysql ( id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, user_id INT UNSIGNED NOT NULL, action_type ENUM('click', 'view', 'purchase') NOT NULL, action_data JSON NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, KEY idx_user_id (user_id), KEY idx_created_at (created_at)) ENGINE=InnoDB;-- 查询：获取每个用户最新的3个行为（MySQL版本）SELECT ub.user_id, JSON_ARRAYAGG( JSON_OBJECT('type', ub.action_type, 'data', ub.action_data) ORDER BY ub.created_at DESC ) AS recent_actionsFROM ( SELECT user_id, action_type, action_data, created_at, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY created_at DESC) as rn FROM user_behavior_mysql WHERE created_at >= '2025-01-01') ubWHERE ub.rn <= 3GROUP BY ub.user_id;
-- PostgreSQL 15 行为日志表设计（利用数组类型）CREATE TABLE user_behavior_pg ( user_id INTEGER NOT NULL, action_type VARCHAR(50) NOT NULL, action_data JSONB NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW(), event_id BIGSERIAL) PARTITION BY RANGE (created_at);-- 创建分区（按月份）CREATE TABLE user_behavior_2025_01 PARTITION OF user_behavior_pg FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');-- 创建BRIN索引（适用于时序数据）CREATE INDEX idx_user_behavior_brin ON user_behavior_pg USING BRIN (user_id, created_at);-- 查询：获取每个用户最新的3个行为（PostgreSQL优化版本）WITH ranked_actions AS ( SELECT user_id, action_type, action_data, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY created_at DESC) as rn FROM user_behavior_pg WHERE created_at >= '2025-01-01'::timestamptz)SELECT user_id, ARRAY_AGG(action_data ORDER BY rn) FILTER (WHERE rn <= 3) AS recent_actionsFROM ranked_actionsGROUP BY user_id;-- 更高效的数组聚合方案SELECT user_id, (ARRAY_AGG(action_data ORDER BY created_at DESC))[1:3] AS recent_actionsFROM user_behavior_pgWHERE created_at >= '2025-01-01'::timestamptzGROUP BY user_id;
-- 特征工程查询（涉及7张表JOIN）SELECT u.user_id, p.profile_features, b.behavior_score, i.item_embeddings, c.context_featuresFROM users uJOIN user_profiles p ON u.user_id = p.user_idJOIN behavior_summary b ON u.user_id = b.user_idJOIN user_interests i ON u.user_id = i.user_idJOIN context_data c ON u.last_login_ip = c.ip_addressJOIN geographic_info g ON u.location_id = g.idJOIN device_info d ON u.device_id = d.idWHERE u.active_status = 'active' AND b.last_update > NOW() - INTERVAL '7 days' AND g.region IN ('NA', 'EU');
-- 创建并行安全函数CREATE OR REPLACE FUNCTION calculate_tfidf( user_doc TEXT, corpus_doc TEXT[], OUT tfidf_vector REAL[]) AS $$DECLARE total_docs INT := array_length(corpus_doc, 1); word_count HSTORE; doc_freq HSTORE;BEGIN -- 分词并计算TF SELECT hstore_agg(word => count::TEXT) INTO word_count FROM regexp_split_to_table(lower(user_doc), '\s+') AS word GROUP BY word; -- 获取文档频率（预计算） SELECT hstore_agg(word => df::TEXT) INTO doc_freq FROM word_document_frequency WHERE word = ANY (akeys(word_count)); -- 计算TF-IDF SELECT ARRAY_AGG( (COALESCE(word_count[word], '0')::REAL / total_docs) * log(total_docs / (COALESCE(doc_freq[word], '1')::REAL) ) INTO tfidf_vector FROM (SELECT unnest(akeys(word_count)) AS word) sub; RETURN;END;$$ LANGUAGE plpgsql PARALLEL SAFE;-- 并行执行特征计算SET max_parallel_workers_per_gather = 8;SET parallel_tuple_cost = 0.1;WITH user_documents AS ( SELECT user_id, STRING_AGG(action_description, ' ') AS doc FROM user_actions WHERE created_at >= CURRENT_DATE - 30 GROUP BY user_id)SELECT user_id, calculate_tfidf(doc, (SELECT ARRAY_AGG(doc) FROM user_documents))FROM user_documents;-- 监控并行执行EXPLAIN (ANALYZE, VERBOSE, BUFFERS)SELECT /* 查询语句 */;
# 1. MySQL schema转换（使用pgloader）cat > mysql_to_pg.load <
方案二：物化视图重构（推荐算法场景）
在迁移过程中，重构数据模型以利用PostgreSQL特性：
-- 将MySQL中的多表JOIN预计算转为物化视图CREATE MATERIALIZED VIEW user_feature_vectors ASWITH behavior_stats AS ( SELECT user_id, COUNT(*) AS total_actions, COUNT(DISTINCT action_type) AS action_diversity, ARRAY_AGG(action_type ORDER BY created_at DESC) AS action_sequence FROM user_behavior_pg WHERE created_at >= NOW() - INTERVAL '30 days' GROUP BY user_id),profile_enhanced AS ( SELECT p.user_id, p.profile_vector, bs.total_actions, bs.action_diversity, bs.action_sequence FROM user_profiles p LEFT JOIN behavior_stats bs ON p.user_id = bs.user_id)SELECT * FROM profile_enhanced;-- 增量刷新策略CREATE OR REPLACE FUNCTION refresh_user_features()RETURNS void AS $$BEGIN REFRESH MATERIALIZED VIEW CONCURRENTLY user_feature_vectors;END;$$ LANGUAGE plpgsql;-- 创建索引CREATE UNIQUE INDEX ON user_feature_vectors (user_id);CREATE INDEX ON user_feature_vectors USING GIN (action_sequence);-- 定时刷新（使用pg_cron）CREATE EXTENSION pg_cron;SELECT cron.schedule('refresh-features', '*/15 * * * *', 'SELECT refresh_user_features()');
迁移验证脚本：
#!/usr/bin/env python3"""迁移数据一致性验证脚本"""import mysql.connectorimport psycopg2import hashlibfrom concurrent.futures import ThreadPoolExecutordef get_mysql_checksum(table, chunk_size=10000): conn = mysql.connector.connect(**MYSQL_CONFIG) cursor = conn.cursor() cursor.execute(f"SELECT COUNT(*) FROM {table}") total = cursor.fetchone()[0] checksums = [] for offset in range(0, total, chunk_size): cursor.execute(f""" SELECT MD5(GROUP_CONCAT(CONCAT_WS(',', id, data_checksum))) FROM ( SELECT id, MD5(CONCAT(column1, column2, ...)) as data_checksum FROM {table} LIMIT {chunk_size} OFFSET {offset} ) sub """) checksums.append(cursor.fetchone()[0]) return hashlib.md5(''.join(checksums).encode()).hexdigest()def get_pg_checksum(table, chunk_size=10000): conn = psycopg2.connect(**PG_CONFIG) cursor = conn.cursor() cursor.execute(f"SELECT COUNT(*) FROM {table}") total = cursor.fetchone()[0] checksums = [] for offset in range(0, total, chunk_size): cursor.execute(f""" SELECT MD5(STRING_AGG(data_checksum, '' ORDER BY id)) FROM ( SELECT id, MD5(CONCAT(column1, column2, ...)) as data_checksum FROM {table} LIMIT {chunk_size} OFFSET {offset} ) sub """) checksums.append(cursor.fetchone()[0]) return hashlib.md5(''.join(checksums).encode()).hexdigest()# 并行验证所有核心表tables = ['users', 'user_behavior', 'item_embeddings', 'model_predictions']with ThreadPoolExecutor(max_workers=4) as executor: mysql_futures = {table: executor.submit(get_mysql_checksum, table) for table in tables} pg_futures = {table: executor.submit(get_pg_checksum, f"migration.{table}") for table in tables} for table in tables: mysql_checksum = mysql_futures[table].result() pg_checksum = pg_futures[table].result() status = "✓ PASS" if mysql_checksum == pg_checksum else "✗ FAIL" print(f"{table}: MySQL={mysql_checksum} PG={pg_checksum} {status}")
3生产环境切换与回滚预案
蓝绿部署策略：
-- 创建蓝绿环境切换函数CREATE OR REPLACE PROCEDURE switch_environment(target_env TEXT)LANGUAGE plpgsql AS $$BEGIN IF target_env = 'green' THEN -- 切换读流量到新集群 ALTER DATABASE ml_platform rename TO ml_platform_blue; ALTER DATABASE ml_platform_green rename TO ml_platform; -- 更新连接池配置 PERFORM pg_reload_conf(); -- 监控切换后性能 INSERT INTO deployment_log (action, timestamp, status) VALUES ('switch_to_green', NOW(), 'completed'); END IF;END;$$;-- 准备回滚点CREATE SNAPSHOT before_switch_YYYYMMDD;
回滚检查清单：
检查项
确认命令
阈值
回滚条件
主从延迟
SELECT pg_stat_replication;
活跃连接数
SELECT count(*) FROM pg_stat_activity;
错误率
SELECT * FROM pg_stat_statements;
平均查询时间
同上
V. 性能基准测试与真实案例分析
1标准化基准测试工具部署
pbench测试框架配置：
# pbench_config.pyimport psycopg2import mysql.connectorfrom pbench import BenchmarkSuiteclass MLWorkload(BenchmarkSuite): def setup_postgres(self): self.pg_conn = psycopg2.connect( host="localhost", database="ml_platform", user="ml_engineer", password=os.getenv("PG_PASSWORD") ) self.pg_conn.set_session(autocommit=True) def setup_mysql(self): self.mysql_conn = mysql.connector.connect( host="mysql-host", database="source_db", user="root", password=os.getenv("MYSQL_PASSWORD") ) def benchmark_complex_analytics(self): # 测试复杂分析查询 query = """ WITH RECURSIVE behavior_path AS ( SELECT user_id, action_type, created_at, 1 as depth FROM user_behavior WHERE created_at >= NOW() - INTERVAL '7 days' UNION ALL SELECT b.user_id, b.action_type, b.created_at, p.depth + 1 FROM user_behavior b JOIN behavior_path p ON b.user_id = p.user_id WHERE b.created_at > p.created_at AND p.depth < 5 ) SELECT user_id, COUNT(DISTINCT action_sequence) as unique_paths FROM ( SELECT user_id, STRING_AGG(action_type, '->' ORDER BY created_at) as action_sequence FROM behavior_path GROUP BY user_id, depth ) sub GROUP BY user_id; """ # PG执行 with self.pg_conn.cursor() as cur: cur.execute("EXPLAIN (ANALYZE, BUFFERS) " + query) pg_plan = cur.fetchall() # MySQL执行 with self.mysql_conn.cursor() as cur: cur.execute("EXPLAIN ANALYZE " + query.replace('INTERVAL', 'INTERVAL').replace('NOW()', 'NOW()')) mysql_plan = cur.fetchall() return { 'postgres': self.extract_execution_time(pg_plan), 'mysql': self.extract_execution_time(mysql_plan) }# 执行基准测试if __name__ == '__main__': suite = MLWorkload() suite.run_benchmarks(duration_seconds=300) suite.generate_report(output_format='markdown')
-- MySQL 8.0原方案：特征表设计CREATE TABLE user_features ( user_id INT PRIMARY KEY, feature_json JSON NOT NULL, -- 存储5000维稀疏特征 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP) ENGINE=InnoDB;-- 问题1：特征更新慢UPDATE user_features SET feature_json = JSON_SET(feature_json, '$.feature_1234', 0.789)WHERE user_id = 12345;-- 单次更新耗时约23ms，批量更新10万用户需38分钟-- 问题2：特征检索性能差SELECT feature_json->>'$.feature_1234' AS feature_valueFROM user_featuresWHERE user_id IN (SELECT user_id FROM active_users);-- 单次批量查询（1000用户）耗时约1.8秒
PostgreSQL重构方案：
-- 方案：HSTORE+分区表CREATE TABLE user_features_pg ( user_id INTEGER NOT NULL, feature_date DATE NOT NULL, features HSTORE NOT NULL, -- 键值对存储，查询效率高 version INTEGER DEFAULT 1, PRIMARY KEY (user_id, feature_date)) PARTITION BY RANGE (feature_date);-- 创建月分区CREATE TABLE user_features_2025_01 PARTITION OF user_features_pg FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');-- 创建GIN索引加速稀疏特征查询CREATE INDEX idx_features_gin ON user_features_pg USING GIN (features);-- 优化后的特征更新（批量操作）CREATE OR REPLACE FUNCTION batch_update_features( user_ids INTEGER[], feature_key TEXT, feature_values REAL[]) RETURNS INTEGER AS $$DECLARE updated_count INTEGER;BEGIN UPDATE user_features_pg f SET features = f.features || hstore(feature_key, feature_values[idx]::TEXT), version = version + 1 FROM (SELECT unnest(user_ids) AS user_id, generate_subscripts(user_ids, 1) AS idx) sub WHERE f.user_id = sub.user_id AND f.feature_date = CURRENT_DATE; GET DIAGNOSTICS updated_count = ROW_COUNT; RETURN updated_count;END;$$ LANGUAGE plpgsql;-- 调用批量更新（10万用户）SELECT batch_update_features( ARRAY(SELECT user_id FROM staging_updates LIMIT 100000), 'feature_1234', ARRAY(SELECT random() FROM generate_series(1, 100000)));-- 执行时间：4.3秒（提升530倍）
性能对比结果（生产环境数据）：
指标
MySQL 8.0
PostgreSQL 15
提升倍数
单特征更新
23ms
0.04ms
575x
批量更新10万
38分钟
4.3秒
530x
特征查询（1000用户）
1.8秒
12ms
150x
存储空间
1.2TB
680GB
1.8x（压缩）
3 A/B测试框架的SQL化改造
MySQL方案（应用层逻辑）：
# Python代码实现测试分组def assign_test_group(user_id): cursor.execute("SELECT user_id FROM experiment_groups WHERE user_id = %s", (user_id,)) if cursor.fetchone(): return cursor.fetchone()['group_name'] group = 'A' if hash(user_id) % 100 < 50 else 'B' cursor.execute("INSERT INTO experiment_groups VALUES (%s, %s)", (user_id, group)) return group
PostgreSQL方案（数据库层函数）：
-- 创建确定性哈希函数CREATE OR REPLACE FUNCTION stable_hash(input TEXT)RETURNS INTEGER AS $$ SELECT ('x' || substr(md5(input), 1, 8))::bit(32)::int;$$ LANGUAGE SQL IMMUTABLE PARALLEL SAFE;-- 测试分组函数CREATE OR REPLACE FUNCTION get_experiment_group( p_user_id INTEGER, p_experiment_id TEXT, p_allocation JSONB -- {'A': 50, 'B': 30, 'C': 20}) RETURNS TEXT AS $$DECLARE user_hash INTEGER; cumulative INTEGER := 0; group_name TEXT; group_percent INTEGER;BEGIN user_hash := abs(stable_hash(p_experiment_id || p_user_id::TEXT)) % 100; FOR group_name, group_percent IN SELECT key, value::INTEGER FROM jsonb_each_text(p_allocation) LOOP cumulative := cumulative + group_percent; IF user_hash < cumulative THEN RETURN group_name; END IF; END LOOP; RETURN 'control';END;$$ LANGUAGE plpgsql IMMUTABLE;-- 使用示例SELECT user_id, get_experiment_group(user_id, 'homepage_redesign', '{"A": 50, "B": 25, "C": 25}') AS test_groupFROM active_usersWHERE user_id BETWEEN 1 AND 10000;-- 索引支持CREATE INDEX idx_experiment_lookup ON active_users USING btree (get_experiment_group(user_id, 'homepage_redesign', '{"A": 50, "B": 25, "C": 25}'));
We and selected third parties use cookies or similar technologies for technical purposes and, with your consent, for other purposes as specified in the cookie policy. You can consent to the use of such technologies by closing this notice, by interacting with any link or button outside of this notice or by continuing to browse otherwise.
NULL, action_type ENUM('click', 'view', 'purchase') NOT NULL, action_data JSON NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, KEY idx_user_id (user_id), KEY idx_created_at (created_at)) ENGINE=InnoDB;-- 查询：获取每个用户最新的3个行为（MySQL版本）SELECT ub.user_id, JSON_ARRAYAGG( JSON_OBJECT('type', ub.action_type, 'data', ub.action_data) ORDER BY ub.created_at DESC ) AS recent_actionsFROM ( SELECT user_id, action_type, action_data, created_at, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY created_at DESC) as rn FROM user_behavior_mysql WHERE created_at >= '2025-01-01') ubWHERE ub.rn <= 3GROUP BY ub.user_id;
NULL, action_type ENUM('click', 'view', 'purchase') NOT NULL, action_data JSON NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, KEY idx_user_id (user_id), KEY idx_created_at (created_at)) ENGINE=InnoDB;-- 查询：获取每个用户最新的3个行为（MySQL版本）SELECT ub.user_id, JSON_ARRAYAGG( JSON_OBJECT('type', ub.action_type, 'data', ub.action_data) ORDER BY ub.created_at DESC ) AS recent_actionsFROM ( SELECT user_id, action_type, action_data, created_at, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY created_at DESC) as rn FROM user_behavior_mysql WHERE created_at >= '2025-01-01') ubWHERE ub.rn <= 3GROUP BY ub.user_id;