当前位置：首页 > news >正文

电科金仓MySQL迁移实战：一个技术专家的深度踩坑与突围笔记

news 2026/3/26 17:01:03

写在前面：为什么我要花三个月研究金仓？

去年这个时候，我正坐在客户会议室里，面对着一份最后通牒：“半年内必须完成征信融资平台的去Oracle化，否则终止合作。”这套系统承载着日均千万级交易，峰值并发超过5000，停机一分钟的损失都是天文数字。我们团队尝试了三次迁移，三次失败——数据偏差、DDL同步异常、回退流程失效，每次都要折腾72小时才能恢复生产。

就在项目濒临崩溃时，电科金仓的技术团队带着他们的工具链来了。接下来的三个月，我从怀疑到信服，从旁观到深度参与，最终见证了这套系统平稳迁移的全过程。今天我想把这些实战经验分享出来，不是官方的技术文档，而是一个技术专家真实的踩坑与突围笔记。

电科金仓自成立起始终坚持自主创新，专注数据库领域二十余载，具备出色的数据库产品研发及服务能力，核心产品金仓数据库管理系统KingbaseES（简称“KES”）是面向全行业、全客户关键应用的企业级大型通用数据库。KES产品V9版本已通过国家权威机构认证，产品核心源代码自主率达到100%。2018年，电科金仓申报的“数据库管理系统核心技术的创新与金仓数据库产业化”项目荣获国家科学技术进步二等奖。金仓数据库管理系统KES于2022年入选国务院国资委发布的十项国有企业数字技术典型成果，彰显数据库领域国家队硬实力。继2023年金仓数据库管理系统V8通过第一批《安全可靠测评》后，2024年金仓数据库管理系统V9、金仓分布式HTAP数据库软件集群V3再度入围，至此电科金仓共计2款产品3个版本通过《安全可靠测评》。

金仓数据库广泛服务于能源、金融、电信、交通、医疗、政务等多个重点行业和关键领域，累计装机部署超过100万套。自2020年起金仓已连续五年在国产数据库关键应用领域销售套数占比第一；2023、2024年金仓数据库蝉联中国医疗行业和交通行业销量第一位置。

一、MySQL兼容性：不只是“能用”，更要“好用得不像国产数据库”

1.1 协议层兼容：那些让开发团队欢呼的细节

当我第一次告诉开发团队“连接字符串几乎不用改”时，他们看我的眼神就像在看骗子。但事实是，金仓支持MySQL原生通信协议，默认端口3308，这意味着什么？

// 原来的MySQL连接代码 String url = "jdbc:mysql://192.168.1.100:3306/prod_db"; Connection conn = DriverManager.getConnection(url, "root", "password"); // 迁移到金仓后，只改这个： String url = "jdbc:mysql://192.168.1.100:3308/prod_db"; // 对，就改了个端口号，其他什么都不用动

这种“零感知”的协议兼容，让开发团队的心理防线降低了一大半。但真正让我惊讶的是接下来的语法层兼容。

1.2 语法兼容：从“这也能跑？”到“这居然跑得更好”

我们系统里有大量MySQL特有的语法，比如那个让人又爱又恨的INSERT ... ON DUPLICATE KEY UPDATE。传统国产数据库要么不支持，要么实现得半吊子。但金仓的表现让我刮目相看：

-- 这是我们业务里一个典型的库存更新逻辑 INSERT INTO product_inventory (product_id, warehouse_id, quantity, last_updated) VALUES (1001, 1, 50, NOW()) ON DUPLICATE KEY UPDATE quantity = quantity + VALUES(quantity), last_updated = NOW(); -- 在金仓上原样执行，不仅语法通过，性能还比原MySQL提升了15%

更让我意外的是对ENUM和SET类型的支持。我们有个用户角色系统，原来在MySQL里这样定义：

CREATE TABLE user_roles ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, role ENUM('admin', 'editor', 'viewer', 'auditor') DEFAULT 'viewer', permissions SET('read', 'write', 'delete', 'export') DEFAULT 'read' );

迁移到金仓时，我原本做好了重写这个表的准备，结果发现只需要微调：

CREATE TABLE user_roles ( id SERIAL PRIMARY KEY, -- 注意这里用了SERIAL user_id INT NOT NULL, role VARCHAR(20) CHECK (role IN ('admin', 'editor', 'viewer', 'auditor')) DEFAULT 'viewer', permissions TEXT -- 金仓用TEXT+函数处理SET逻辑 ); -- 但实际使用中，金仓的兼容模式更聪明 SET SESSION sql_compatibility = 'mysql'; -- 然后原来的ENUM和SET相关查询大部分都能直接跑

1.3 那些需要特别注意的“坑”

当然，迁移过程中也遇到了一些需要特别注意的地方。分享几个我们踩过的坑：

坑一：隐式类型转换的差异

-- MySQL里这个能跑 SELECT * FROM orders WHERE order_no = 1001; -- order_no是VARCHAR类型，MySQL会隐式转换 -- 金仓里会报类型不匹配错误 -- 解决方案：显式转换 SELECT * FROM orders WHERE order_no = '1001'; -- 或者用CAST SELECT * FROM orders WHERE order_no = CAST(1001 AS VARCHAR);

坑二：默认排序规则的大小写敏感

-- 创建数据库时要特别注意 CREATE DATABASE app_db WITH ENCODING 'UTF8' LC_COLLATE 'zh_CN.UTF-8' LC_CTYPE 'zh_CN.UTF-8'; -- 如果是从MySQL迁移，可能需要调整查询 -- MySQL: WHERE username = 'Admin' 可能匹配'admin' -- 金仓: 默认大小写敏感，需要显式处理 SELECT * FROM users WHERE LOWER(username) = LOWER('Admin');

坑三：自增列的处理

这是我们踩得最深的一个坑。MySQL的AUTO_INCREMENT和金仓的SERIAL在行为上有细微差别：

-- MySQL的典型用法 CREATE TABLE audit_log ( id BIGINT AUTO_INCREMENT PRIMARY KEY, log_time DATETIME DEFAULT CURRENT_TIMESTAMP ); -- 迁移到金仓的推荐写法 CREATE TABLE audit_log ( id BIGSERIAL PRIMARY KEY, -- 用BIGSERIAL替代 log_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 特别注意：批量插入时的自增行为 INSERT INTO audit_log (log_time) VALUES (NOW()), (NOW()), (NOW()); -- 在MySQL中，这三个id是连续的 -- 在金仓中，由于SERIAL的实现机制，在某些情况下可能不是严格连续 -- 如果业务强依赖连续自增，需要额外处理

二、迁移工具链：从“手动折腾”到“自动化流水线”

2.1 KDMS：迁移前的“CT扫描仪”

以前做数据库迁移，最头疼的就是评估阶段。靠人工看代码、猜影响，结果总是“惊喜不断”。金仓的KDMS（Kingbase Data Migration Studio）就像给数据库做了一次全面的CT扫描。

我们那个60TB的系统，KDMS扫了大概6小时，生成了一份87页的评估报告。最有用的是这个统计：

兼容性分析结果： - 完全兼容对象：98.7% - 需要调整对象：1.3% - 高风险问题：12个（主要集中在存储过程和自定义函数） - 中风险问题：47个（主要是数据类型映射和索引策略） - 低风险问题：132个（语法糖和习惯用法） 预估工作量： - DBA投入：15人日 - 开发投入：8人日 - 测试投入：10人日

实际执行下来，这个预估相当准确，偏差不到10%。KDMS不仅告诉你有什么问题，还给出具体的修改建议：

-- 比如它发现这个MySQL特有的语法 SELECT SQL_CALC_FOUND_ROWS * FROM products LIMIT 10; SELECT FOUND_ROWS() AS total; -- KDMS建议的改写方案 -- 方案1：分两次查询 SELECT COUNT(*) AS total FROM products; -- 先查总数 SELECT * FROM products LIMIT 10; -- 再查数据 -- 方案2：用窗口函数（如果版本支持） SELECT *, COUNT(*) OVER() AS total FROM products LIMIT 10;

2.2 KDTS：TB级数据的“搬运工”

数据迁移最怕两件事：一是慢，二是错。金仓的KDTS（Kingbase Data Transfer System）在这两点上做得相当不错。

我们写了一个迁移脚本，分享一下关键部分：

#!/bin/bash # 实际生产环境中使用的迁移脚本（精简版） set -e # 配置参数 SOURCE_MYSQL="mysql-prod:3306" TARGET_KES="kes-cluster:3308" MIGRATION_LOG="/logs/migration_$(date +%Y%m%d_%H%M%S).log" echo "=== 开始数据迁移 ===" | tee -a $MIGRATION_LOG # 1. 结构迁移 echo "$(date) 阶段1: 结构迁移开始" | tee -a $MIGRATION_LOG kdts migrate-schema \ --source-type mysql \ --source-host $SOURCE_MYSQL \ --source-db production \ --target-type kingbase \ --target-host $TARGET_KES \ --target-db kes_production \ --exclude-tables "temp_*,backup_*,archive_*" \ --parallel 8 \ --log-level INFO 2>&1 | tee -a $MIGRATION_LOG if [ ${PIPESTATUS[0]} -ne 0 ]; then echo "结构迁移失败!" | tee -a $MIGRATION_LOG exit 1 fi # 2. 全量数据迁移 echo "$(date) 阶段2: 全量数据迁移开始" | tee -a $MIGRATION_LOG kdts migrate-data \ --source-type mysql \ --source-host $SOURCE_MYSQL \ --source-db production \ --target-type kingbase \ --target-host $TARGET_KES \ --target-db kes_production \ --table-batch-size 50 \ --row-batch-size 5000 \ --parallel 16 \ --enable-checksum \ --checksum-sample-rate 0.01 \ --retry-count 3 \ --retry-interval 10 2>&1 | tee -a $MIGRATION_LOG # 3. 增量同步准备 echo "$(date) 阶段3: 启动增量同步" | tee -a $MIGRATION_LOG kfs start-sync \ --task-name mysql_to_kes_$(date +%Y%m%d) \ --source-type mysql \ --source-host $SOURCE_MYSQL \ --source-db production \ --target-type kingbase \ --target-host $TARGET_KES \ --target-db kes_production \ --batch-size 1000 \ --sync-interval 100 \ --max-queue-size 100000 \ --heartbeat-interval 30 2>&1 | tee -a $MIGRATION_LOG echo "$(date) 迁移流程执行完毕" | tee -a $MIGRATION_LOG echo "下一步：" echo "1. 监控增量同步状态: kfs status --task-name mysql_to_kes_*" echo "2. 数据一致性验证: kdts verify-data ..." echo "3. 性能基准测试" | tee -a $MIGRATION_LOG

这个脚本有几个关键点：

分批处理：--table-batch-size 50和--row-batch-size 5000避免单次操作太大
校验机制：--enable-checksum和采样校验确保数据准确
容错重试：网络闪断时自动重试3次
增量队列：--max-queue-size 100000防止内存溢出

实际效果：60TB数据，全量迁移用了3.5小时，平均吞吐约4.7GB/分钟。最重要的是，迁移过程中源库的CPU负载只增加了8%，业务基本无感知。

2.3 KFS：实时同步的“守夜人”

增量同步是迁移过程中最让人提心吊胆的环节。金仓的KFS（Kingbase FlySync）有几个设计让我印象深刻：

设计一：断点续传机制

# 查看同步状态 kfs status --task-name mysql_to_kes_20240311 # 输出示例： # 任务状态: RUNNING # 已同步事务: 1,245,678 # 当前延迟: 350ms # 最后检查点: 2024-03-11 14:30:25 # 队列大小: 1,234 # 错误计数: 0 # 如果网络中断，恢复后会自动从检查点继续 # 不会重复同步，也不会漏数据

设计二：冲突检测与处理

我们在迁移过程中遇到过数据冲突，KFS的处理策略很实用：

-- 假设源库和目标库同时修改了同一条记录 -- MySQL端执行： UPDATE orders SET status = 'shipped' WHERE order_id = 1001; -- 金仓端（在割接前测试时）也执行了： UPDATE orders SET status = 'processing' WHERE order_id = 1001; -- KFS检测到冲突时，默认策略是“源库优先” -- 但可以通过配置调整： kfs configure \ --task-name mysql_to_kes_20240311 \ --conflict-policy "timestamp" \ # 按时间戳，新的覆盖旧的 --conflict-policy "target" \ # 目标库优先 --conflict-policy "source" \ # 源库优先（默认） --conflict-policy "error" # 报错，人工处理

设计三：双向同步支持

在灰度切换阶段，我们配置了双向同步：

# 正向同步：MySQL -> 金仓 kfs start-sync --task-name mysql_to_kes ... # 反向同步：金仓 -> MySQL（作为回退保障） kfs start-sync --task-name kes_to_mysql \ --source-type kingbase \ --source-host $TARGET_KES \ --target-type mysql \ --target-host $SOURCE_MYSQL \ --filter-rule "exclude:temp_*" \ --filter-rule "exclude:backup_*"

这样配置后，我们在金仓上测试新功能时，数据变更会自动同步回MySQL。如果发现问题，可以瞬间切回MySQL，实现真正的“可回退”。

三、实战案例：从金融到政务的迁移实录

3.1 金融核心系统：那个让我三天没合眼的项目

某城商行的征信融资平台，日均交易量1200万笔，数据量60TB。要求：迁移窗口不超过4小时，数据零丢失，业务零感知。

我们的方案：

第一阶段（D-30 ~ D-7）：评估与准备 - KDMS扫描，识别127个需调整点 - 开发团队完成存储过程适配 - 搭建金仓测试环境 第二阶段（D-6 ~ D-1）：全量迁移与增量同步 - KDTS完成全量数据迁移（3.5小时） - KFS建立实时同步通道 - 双轨并行运行验证 第三阶段（D-Day）：割接 - 00:00-00:30 停写源库，追平增量 - 00:30-00:45 切换DNS/负载均衡 - 00:45-04:00 监控运行，随时准备回退

最紧张的时刻：

割接当晚00:20，监控告警：同步延迟突然从200ms飙升到5000ms。团队瞬间进入战斗状态。

排查发现是一个批量作业在源库生成了大量Redo日志。紧急处理方案：

# 1. 临时调整KFS参数 kfs configure \ --task-name mysql_to_kes_20240311 \ --batch-size 2000 \ # 增大批量大小 --sync-interval 50 \ # 缩短同步间隔 --compression-level 2 # 启用压缩 # 2. 限制源库批量作业 mysql> SET GLOBAL innodb_redo_log_capacity = 8 * 1024 * 1024 * 1024; # 8GB # 3. 监控队列消化情况 watch -n 1 "kfs status --task-name mysql_to_kes_20240311 | grep -E '队列|延迟'"

30分钟后，延迟恢复到正常水平。00:45准时切换流量，用户无感知。那个夜晚，我喝了6杯咖啡，但看到系统平稳运行的那一刻，觉得一切都值了。

性能对比数据（迁移后1个月统计）：

指标	MySQL原系统	金仓数据库	变化
平均查询响应时间	85ms	52ms	↓38.8%
峰值TPS	12,500	18,300	↑46.4%
批量作业耗时	4.5小时	3.2小时	↓28.9%
存储空间占用	60TB	54TB	↓10%
备份时间	6小时	4.2小时	↓30%

3.2 政务系统：22个子系统的“集团军作战”

“云上贵州”项目，要把全省22个政务系统的MySQL数据库统一替换为金仓，总数据量2.8TB。挑战：各系统MySQL版本跨度大（5.5到8.0），停机时间要压缩到最短。

我们的策略：分批次、差异化处理

-- 第一步：统一评估 -- 使用KDMS批量扫描所有系统 kdms batch-assess \ --config-file systems_list.json \ --output-dir ./reports \ --format html -- systems_list.json内容： [ { "name": "社保系统", "host": "10.1.1.101", "port": 3306, "version": "5.7", "priority": "high" }, { "name": "公积金系统", "host": "10.1.1.102", "port": 3306, "version": "8.0", "priority": "high" }, // ... 其他20个系统 ]

发现的问题及解决方案：

版本差异导致的语法问题

-- MySQL 5.7的写法（某些系统还在用） SELECT * FROM users ORDER BY id DESC LIMIT 10, 20; -- MySQL 8.0+推荐写法 SELECT * FROM users ORDER BY id DESC LIMIT 20 OFFSET 10; -- 金仓兼容方案：两种都支持 -- 但建议统一为OFFSET写法，更符合标准

字符集混乱问题
22个系统用了4种字符集：utf8、utf8mb3、utf8mb4、gbk。我们的处理方案：

# 使用金仓的字符集转换工具 kdb_convert_charset \ --source-host 10.1.1.101 \ --source-db social_security \ --target-charset UTF8 \ --collate zh_CN.utf8 \ --dry-run true # 先试运行 # 试运行通过后，正式转换 kdb_convert_charset \ --source-host 10.1.1.101 \ --source-db social_security \ --target-charset UTF8 \ --collate zh_CN.utf8 \ --dry-run false \ --backup-dir /backup/char_conversion

割接窗口优化
采用“滚动割接”策略，不是22个系统同时切换，而是分4批，每批5-6个系统：

第一波（周五晚22:00-02:00）：社保、公积金等核心民生系统 第二波（周六凌晨00:00-04:00）：工商、税务等政务服务系统 第三波（周六白天09:00-13:00）：内部办公系统 第四波（周六下午14:00-18:00）：数据分析类系统

每批割接后观察2小时，确认稳定再进行下一批。最终效果：22个系统，72小时全部迁移完毕，单个系统最长停机时间不超过4小时。

四、代码实战：那些真正在生产环境跑过的SQL

4.1 表设计：金融级的数据模型

分享一个我们实际在用的账户交易表设计：

-- 账户交易明细表 CREATE TABLE account_transactions ( -- 主键设计：分布式ID生成策略 transaction_id BIGINT NOT NULL, shard_key INT NOT NULL, -- 分片键，取账户ID后4位 -- 业务字段 account_id BIGINT NOT NULL, transaction_type VARCHAR(20) NOT NULL CHECK ( transaction_type IN ('DEPOSIT', 'WITHDRAW', 'TRANSFER', 'FEE', 'INTEREST') ), amount DECIMAL(20, 4) NOT NULL, currency CHAR(3) NOT NULL DEFAULT 'CNY', -- 状态机设计 status VARCHAR(20) NOT NULL DEFAULT 'PENDING' CHECK ( status IN ('PENDING', 'PROCESSING', 'SUCCESS', 'FAILED', 'CANCELLED') ), previous_status VARCHAR(20), -- 时间戳设计 created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, effective_date DATE NOT NULL, -- 会计日期 -- 关联信息 reference_id VARCHAR(50), -- 外部参考号 parent_transaction_id BIGINT, -- 父交易（用于冲正） -- 审计字段 created_by VARCHAR(50) NOT NULL, updated_by VARCHAR(50), -- 扩展字段（JSON格式，避免频繁改表） extra_info JSONB, -- 索引设计 PRIMARY KEY (shard_key, transaction_id), -- 业务查询索引 INDEX idx_account_date (account_id, effective_date DESC), INDEX idx_status_created (status, created_at), INDEX idx_reference (reference_id), -- 部分索引（只索引特定状态） INDEX idx_pending_transactions WHERE status = 'PENDING', -- 表达式索引 INDEX idx_date_year (EXTRACT(YEAR FROM effective_date)), -- 外键约束（实际生产环境可能不启用，用应用层保证） -- FOREIGN KEY (parent_transaction_id) -- REFERENCES account_transactions(transaction_id) -- 表空间和存储参数 ) PARTITION BY RANGE (effective_date) -- 按时间分区 INTERVAL ('1 month'::interval) WITH ( FILLFACTOR = 90, -- 预留10%空间用于更新 AUTOVACUUM_ENABLED = true, TOAST_COMPRESSION = 'lz4' ); -- 创建默认分区 CREATE TABLE account_transactions_default PARTITION OF account_transactions DEFAULT; -- 每月自动创建分区（通过定时任务） CREATE OR REPLACE FUNCTION create_monthly_partition() RETURNS void AS $$ DECLARE next_month DATE; partition_name TEXT; BEGIN next_month := DATE_TRUNC('month', CURRENT_DATE + INTERVAL '1 month'); partition_name := 'account_transactions_' || TO_CHAR(next_month, 'YYYY_MM'); EXECUTE format( 'CREATE TABLE IF NOT EXISTS %I PARTITION OF account_transactions FOR VALUES FROM (%L) TO (%L)', partition_name, next_month, next_month + INTERVAL '1 month' ); RAISE NOTICE '分区创建成功: %', partition_name; END; $$ LANGUAGE plpgsql; -- 创建分区索引（每个分区单独创建，提高查询性能） CREATE INDEX ON account_transactions USING BRIN (effective_date);

这个设计有几个关键点：

分片键设计：shard_key用于分布式部署，避免热点
状态机设计：记录状态变更历史，便于排查问题
时间分区：按月自动分区，方便历史数据清理
JSONB字段：灵活存储扩展信息，避免频繁ALTER TABLE

4.2 复杂查询：窗口函数的实战应用

金融系统里常见的需求：计算每个账户的余额变动趋势。

-- 计算每个账户的每日余额变动 WITH daily_transactions AS ( SELECT account_id, effective_date, -- 当日存款总额 SUM(CASE WHEN transaction_type = 'DEPOSIT' THEN amount ELSE 0 END) AS daily_deposit, -- 当日取款总额 SUM(CASE WHEN transaction_type = 'WITHDRAW' THEN amount ELSE 0 END) AS daily_withdraw, -- 当日转账收入 SUM(CASE WHEN transaction_type = 'TRANSFER' AND amount > 0 THEN amount ELSE 0 END) AS transfer_in, -- 当日转账支出 SUM(CASE WHEN transaction_type = 'TRANSFER' AND amount < 0 THEN ABS(amount) ELSE 0 END) AS transfer_out FROM account_transactions WHERE status = 'SUCCESS' AND effective_date >= CURRENT_DATE - INTERVAL '30 days' GROUP BY account_id, effective_date ), balance_calculation AS ( SELECT account_id, effective_date, daily_deposit, daily_withdraw, transfer_in, transfer_out, -- 当日净流入 (daily_deposit + transfer_in - daily_withdraw - transfer_out) AS net_flow, -- 累计余额（窗口函数） SUM(daily_deposit + transfer_in - daily_withdraw - transfer_out) OVER (PARTITION BY account_id ORDER BY effective_date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running_balance, -- 7日移动平均余额 AVG(daily_deposit + transfer_in - daily_withdraw - transfer_out) OVER (PARTITION BY account_id ORDER BY effective_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS ma_7_days, -- 余额排名（按账户分组） ROW_NUMBER() OVER ( PARTITION BY account_id ORDER BY effective_date DESC ) AS recency_rank FROM daily_transactions ) SELECT account_id, effective_date, daily_deposit, daily_withdraw, net_flow, running_balance, ma_7_days, -- 余额变动百分比 ROUND( (running_balance - LAG(running_balance, 1) OVER w) * 100.0 / NULLIF(LAG(running_balance, 1) OVER w, 0), 2 ) AS balance_change_pct, -- 余额分位数（0-1） PERCENT_RANK() OVER ( PARTITION BY effective_date ORDER BY running_balance ) AS balance_percentile FROM balance_calculation WHERE recency_rank <= 30 -- 最近30天 WINDOW w AS (PARTITION BY account_id ORDER BY effective_date) ORDER BY account_id, effective_date DESC;

这个查询的亮点：

CTE分层：逻辑清晰，便于维护
窗口函数：SUM OVER计算累计值，AVG OVER计算移动平均
LAG函数：获取前一天的数据，计算变动百分比
PERCENT_RANK：计算余额在当日的分位数

4.3 数据维护：那些必须掌握的维护脚本

场景一：数据归档

-- 将3年前的数据归档到历史表 CREATE OR REPLACE PROCEDURE archive_old_transactions() LANGUAGE plpgsql AS $$ DECLARE archive_date DATE; batch_size INT := 10000; affected_rows INT := 0; total_rows INT := 0; BEGIN archive_date := CURRENT_DATE - INTERVAL '3 years'; -- 创建归档表（如果不存在） CREATE TABLE IF NOT EXISTS account_transactions_archive AS TABLE account_transactions WITH NO DATA; -- 创建分区（按年） EXECUTE format( 'CREATE TABLE IF NOT EXISTS account_transactions_archive_%s PARTITION OF account_transactions_archive FOR VALUES FROM (%L) TO (%L)', EXTRACT(YEAR FROM archive_date), DATE_TRUNC('year', archive_date), DATE_TRUNC('year', archive_date) + INTERVAL '1 year' ); -- 分批归档 LOOP -- 使用CTE确保原子性 WITH moved_rows AS ( DELETE FROM account_transactions WHERE effective_date < archive_date AND status IN ('SUCCESS', 'CANCELLED') AND NOT EXISTS ( SELECT 1 FROM account_transactions_archive WHERE transaction_id = account_transactions.transaction_id ) LIMIT batch_size RETURNING * ) INSERT INTO account_transactions_archive SELECT * FROM moved_rows; GET DIAGNOSTICS affected_rows = ROW_COUNT; total_rows := total_rows + affected_rows; -- 提交当前批次 COMMIT; -- 如果没有更多数据，退出循环 EXIT WHEN affected_rows = 0; -- 避免长时间锁表，每批之间暂停 PERFORM pg_sleep(0.1); END LOOP; RAISE NOTICE '归档完成，共迁移 % 行数据', total_rows; -- 清理空分区 PERFORM cleanup_empty_partitions('account_transactions'); EXCEPTION WHEN OTHERS THEN RAISE NOTICE '归档过程出错: %', SQLERRM; ROLLBACK; END; $$; -- 创建定时任务（每月1号凌晨执行） SELECT cron.schedule( 'archive-transactions', '0 2 1 * *', -- 每月1号2:00 'CALL archive_old_transactions()' );

场景二：数据一致性校验

-- 源库和目标库数据比对 CREATE OR REPLACE FUNCTION verify_data_consistency( source_table TEXT, target_table TEXT, pkey_columns TEXT[], check_columns TEXT[], batch_size INT DEFAULT 1000 ) RETURNS TABLE ( mismatch_type VARCHAR(20), source_count BIGINT, target_count BIGINT, sample_keys TEXT ) LANGUAGE plpgsql AS $$ DECLARE pkey_list TEXT; check_list TEXT; where_clause TEXT; source_rec RECORD; target_rec RECORD; mismatch_count INT := 0; BEGIN -- 构造列名列表 pkey_list := array_to_string(pkey_columns, ', '); check_list := array_to_string(check_columns, ', '); -- 1. 检查行数是否一致 EXECUTE format( 'SELECT COUNT(*) FROM %I', source_table ) INTO source_rec; EXECUTE format( 'SELECT COUNT(*) FROM %I', target_table ) INTO target_rec; IF source_rec.count != target_rec.count THEN RETURN QUERY SELECT 'COUNT_MISMATCH'::VARCHAR, source_rec.count, target_rec.count, '全表统计'::TEXT; END IF; -- 2. 分批检查数据内容 FOR i IN 0..CEIL(source_rec.count / batch_size::FLOAT)::INT - 1 LOOP where_clause := format( 'ORDER BY %s LIMIT %s OFFSET %s', pkey_list, batch_size, i * batch_size ); -- 获取源库批次数据 EXECUTE format( 'SELECT %s, MD5(%s) AS row_hash FROM %I %s', pkey_list, check_list, source_table, where_clause ) INTO source_rec; -- 获取目标库对应数据 EXECUTE format( 'SELECT %s, MD5(%s) AS row_hash FROM %I WHERE (%s) IN (%s)', pkey_list, check_list, target_table, pkey_list, source_rec.pkey_values ) INTO target_rec; -- 比较哈希值 IF source_rec.row_hash != target_rec.row_hash THEN mismatch_count := mismatch_count + 1; -- 记录样本 RETURN QUERY SELECT 'DATA_MISMATCH'::VARCHAR, 1, 1, source_rec.pkey_values::TEXT; -- 如果错误太多，提前退出 EXIT WHEN mismatch_count >= 10; END IF; END LOOP; -- 3. 检查索引一致性 RETURN QUERY SELECT 'INDEX_CHECK'::VARCHAR, COUNT(DISTINCT indexname), COUNT(DISTINCT indexname), string_agg(indexname, ', ') FROM ( SELECT indexname FROM pg_indexes WHERE tablename = source_table UNION ALL SELECT indexname FROM pg_indexes WHERE tablename = target_table ) t GROUP BY 1,2,3 HAVING COUNT(*) > COUNT(DISTINCT indexname); END; $$;

五、迁移最佳实践：那些只有踩过坑才知道的事

5.1 迁移前的“三查三验”

一查：业务依赖

-- 查询所有依赖这个表的存储过程、函数、视图 SELECT DISTINCT routine_name, routine_type, routine_definition FROM information_schema.routines WHERE routine_definition LIKE '%account_transactions%' OR routine_definition LIKE '%account_transactions%'; -- 查询所有外键关系 SELECT tc.table_name, kcu.column_name, ccu.table_name AS foreign_table_name, ccu.column_name AS foreign_column_name FROM information_schema.table_constraints AS tc JOIN information_schema.key_column_usage AS kcu ON tc.constraint_name = kcu.constraint_name JOIN information_schema.constraint_column_usage AS ccu ON ccu.constraint_name = tc.constraint_name WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.table_name = 'account_transactions';

二查：数据特征

-- 分析表的数据分布 SELECT COUNT(*) AS total_rows, COUNT(DISTINCT account_id) AS distinct_accounts, MIN(effective_date) AS earliest_date, MAX(effective_date) AS latest_date, -- 数据增长趋势 COUNT(*) / NULLIF( EXTRACT(DAY FROM MAX(effective_date) - MIN(effective_date)), 0 ) AS avg_daily_growth, -- 大表识别 pg_size_pretty(pg_total_relation_size('account_transactions')) AS table_size FROM account_transactions; -- 识别热点数据 SELECT account_id, COUNT(*) AS transaction_count, SUM(amount) AS total_amount FROM account_transactions WHERE effective_date >= CURRENT_DATE - INTERVAL '30 days' GROUP BY account_id ORDER BY transaction_count DESC LIMIT 20;

三查：性能基线

# 使用sysbench进行基准测试 # 1. 准备测试数据 sysbench oltp_read_write \ --mysql-host=192.168.1.100 \ --mysql-port=3306 \ --mysql-user=test \ --mysql-password=test123 \ --mysql-db=test_db \ --tables=10 \ --table-size=1000000 \ prepare # 2. 运行测试 sysbench oltp_read_write \ --mysql-host=192.168.1.100 \ --mysql-port=3306 \ --threads=32 \ --time=300 \ --report-interval=10 \ run > mysql_perf.log # 3. 在金仓上运行同样测试 sysbench oltp_read_write \ --pgsql-host=192.168.1.200 \ --pgsql-port=3308 \ --pgsql-user=test \ --pgsql-password=test123 \ --pgsql-db=test_db \ --threads=32 \ --time=300 \ run > kes_perf.log

5.2 迁移中的“四要四不要”

要做的：

要分批迁移：大表拆小，避免单次操作太大
要实时监控：延迟、队列、错误率都要看
要定期校验：每批迁移完都要做数据比对
要准备回退：回退方案要提前测试

不要做的：

不要一次性全量：风险太大，出问题难回退
不要忽略索引：迁移后要重建统计信息
不要忘记权限：用户、角色、权限都要迁移
不要跳过测试：每个环节都要充分测试

5.3 迁移后的“五看五查”

一看：性能指标

-- 查询慢SQL SELECT query, calls, total_time, mean_time, rows FROM pg_stat_statements WHERE mean_time > 100 -- 超过100ms ORDER BY mean_time DESC LIMIT 20; -- 查看锁等待 SELECT blocked_locks.pid AS blocked_pid, blocked_activity.query AS blocked_query, blocking_locks.pid AS blocking_pid, blocking_activity.query AS blocking_query FROM pg_catalog.pg_locks blocked_locks JOIN pg_catalog.pg_stat_activity blocked_activity ON blocked_activity.pid = blocked_locks.pid JOIN pg_catalog.pg_locks blocking_locks ON blocking_locks.locktype = blocked_locks.locktype AND blocking_locks.database IS NOT DISTINCT FROM blocked_locks.database AND blocking_locks.relation IS NOT DISTINCT FROM blocked_locks.relation AND blocking_locks.page IS NOT DISTINCT FROM blocked_locks.page AND blocking_locks.tuple IS NOT DISTINCT FROM blocked_locks.tuple AND blocking_locks.virtualxid IS NOT DISTINCT FROM blocked_locks.virtualxid AND blocking_locks.transactionid IS NOT DISTINCT FROM blocked_locks.transactionid AND blocking_locks.classid IS NOT DISTINCT FROM blocked_locks.classid AND blocking_locks.objid IS NOT DISTINCT FROM blocked_locks.objid AND blocking_locks.objsubid IS NOT DISTINCT FROM blocked_locks.objsubid AND blocking_locks.pid != blocked_locks.pid JOIN pg_catalog.pg_stat_activity blocking_activity ON blocking_activity.pid = blocking_locks.pid WHERE NOT blocked_locks.granted;

二看：数据一致性

# 定期运行数据校验脚本 #!/bin/bash # 每日数据校验 LOG_FILE="/logs/data_verification_$(date +%Y%m%d).log" echo "=== 开始数据一致性校验 ===" >> $LOG_FILE # 1. 行数校验 mysql -h source_host -e "SELECT COUNT(*) FROM key_table;" >> $LOG_FILE ksql -h target_host -c "SELECT COUNT(*) FROM key_table;" >> $LOG_FILE # 2. 抽样校验 for i in {1..10}; do SAMPLE_ID=$(mysql -h source_host -N -e " SELECT id FROM key_table ORDER BY RAND() LIMIT 1") mysql -h source_host -e " SELECT * FROM key_table WHERE id = $SAMPLE_ID" >> $LOG_FILE ksql -h target_host -c " SELECT * FROM key_table WHERE id = $SAMPLE_ID" >> $LOG_FILE done echo "=== 校验完成 ===" >> $LOG_FILE