一次核心业务数据库迁移的全流程复盘：MySQL到TiDB的平滑迁移方案与回滚机制
一、背景与问题
某电商平台的订单系统数据库在2025年Q2达到瓶颈——单MySQL实例（16核64GB内存，2TB SSD），日增量订单500万条，累计数据量达到3.5TB、400亿行。扩展性上已无退路：垂直升级到32核128GB成本高且边际收益递减；分库分表架构复杂且需要大量业务改造。
技术选型最终锁定在TiDB（分布式NewSQL数据库），核心理由：
- 兼容MySQL协议，迁移成本可控
- 水平扩展能力，按Region增加TiKV节点即可
- 社区活跃，PingCAP提供了成熟的迁移工具链
但数据库迁移是对核心业务的"心脏手术"——订单系统每天承载1.2亿元交易流水，任何超过5分钟的停机都是不可接受的，任何数据不一致都可能引发对账灾难。
二、迁移全流程设计
2.1 TiDB集群拓扑方案
| 组件 | 节点数 | 配置 | 用途 |
|---|---|---|---|
| TiDB Server | 3 | 16C32G | SQL解析、计算层，通过HAProxy负载均衡 |
| PD Server | 3 | 8C16G | 集群元数据管理、TSO时间戳、调度 |
| TiKV Server | 6 | 16C64G + 2TB NVMe | 分布式KV存储、Raft副本 |
| TiFlash | 2 | 16C64G + 2TB NVMe | 列式存储副本（AP分析查询） |
2.2 兼容性评估核心发现
| MySQL特性 | TiDB兼容状态 | 处理方案 |
|---|---|---|
| 普通SELECT/INSERT/UPDATE/DELETE | 完全兼容 | 无需修改 |
| AUTO_INCREMENT | 兼容（但不保证连续） | 业务不依赖连续ID |
| FOREIGN KEY | 不支持 | 改为应用层校验 |
| 存储过程 | 基本不支持 | 重写为应用层代码（12个存储过程） |
| 窗口函数(ROW_NUMBER等) | TiDB 7.x完全兼容 | 无需修改 |
| 字符集utf8mb4 | 完全兼容 | 无需修改 |
| NOW()/CURDATE() | 兼容 | 确认时区一致(Asia/Shanghai) |
| LAST_INSERT_ID() | 兼容（会话级别） | 无需修改 |
三、数据校验与回滚脚本
3.1 数据一致性校验
#!/usr/bin/env python3 """MySQL → TiDB 数据迁移一致性校验脚本""" import logging from dataclasses import dataclass from typing import Optional import pymysql import hashlib import json from datetime import datetime logger = logging.getLogger("migration_validator") @dataclass class ValidationResult: table_name: str total_rows: int matched: int mismatched: int missing_in_target: int extra_in_target: int duration_seconds: float class DataMigrationValidator: """数据迁移一致性校验器 采用多级校验策略： L1: 行数对比（快速扫描，发现明显差异） L2: 分片MD5校验（中等开销，发现批量不一致） L3: 逐行对比（高开销，精确定位差异行） """ def __init__(self, mysql_config: dict, tidb_config: dict): self.mysql_conn = pymysql.connect(**mysql_config) self.tidb_conn = pymysql.connect(**tidb_config) def validate_all_tables(self, tables: list[str], level: int = 2) -> dict[str, ValidationResult]: """ 对所有指定表执行校验 level: 校验级别（1=行数, 2=行数+MD5分片, 3=行数+MD5+逐行） """ results = {} for table in tables: logger.info(f"开始校验: {table}") results[table] = self._validate_table(table, level) logger.info( f"校验完成: {table}, 匹配={results[table].matched}, " f"不一致={results[table].mismatched}" ) return results def _validate_table(self, table: str, level: int, batch_size: int = 50000) -> ValidationResult: """单个表的多级校验""" start_time = datetime.now() result = ValidationResult( table_name=table, total_rows=0, matched=0, mismatched=0, missing_in_target=0, extra_in_target=0, duration_seconds=0 ) try: # L1: 行数对比 mysql_count = self._get_row_count(self.mysql_conn, table) tidb_count = self._get_row_count(self.tidb_conn, table) result.total_rows = max(mysql_count, tidb_count) if mysql_count > tidb_count: result.missing_in_target = mysql_count - tidb_count logger.warning( f"{table}: 目标库缺少 {result.missing_in_target} 行" ) elif tidb_count > mysql_count: result.extra_in_target = tidb_count - mysql_count logger.warning( f"{table}: 目标库多余 {result.extra_in_target} 行" ) if level == 1: result.matched = min(mysql_count, tidb_count) return result # L2: 分批MD5校验 processed = 0 for offset in range(0, min(mysql_count, tidb_count), batch_size): mysql_hash = self._compute_batch_hash( self.mysql_conn, table, offset, batch_size ) tidb_hash = self._compute_batch_hash( self.tidb_conn, table, offset, batch_size ) if mysql_hash == tidb_hash: result.matched += batch_size else: # 该批次数据不一致 if level >= 3: # L3: 逐行定位差异 diff_rows = self._find_diff_rows( table, offset, batch_size ) matched_in_batch = batch_size - len(diff_rows) result.matched += matched_in_batch result.mismatched += len(diff_rows) for row_pk in diff_rows[:10]: # 只记录前10条 logger.error( f"数据差异: {table} pk={row_pk}" ) else: result.mismatched += batch_size processed += batch_size result.duration_seconds = ( datetime.now() - start_time ).total_seconds() return result except Exception as e: logger.error(f"表校验失败: {table}, {e}") result.mismatched = result.total_rows return result def _get_row_count(self, conn, table: str) -> int: """获取表的行数（TiDB中COUNT(*)性能较差，大表使用INFORMATION_SCHEMA）""" try: with conn.cursor() as cur: cur.execute( "SELECT TABLE_ROWS FROM INFORMATION_SCHEMA.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s", (table,) ) result = cur.fetchone() return result[0] if result else 0 except Exception as e: logger.error(f"获取行数失败: {table}, {e}") return 0 def _compute_batch_hash(self, conn, table: str, offset: int, limit: int) -> str: """计算一批数据的MD5哈希""" try: with conn.cursor() as cur: query = ( f"SELECT MD5(GROUP_CONCAT(" f" MD5(CONCAT_WS('|', *))" f" ORDER BY id" f")) FROM (" f" SELECT * FROM {table} " f" ORDER BY id LIMIT {limit} OFFSET {offset}" f") t" ) cur.execute(query) result = cur.fetchone() return result[0] if result and result[0] else "" except Exception as e: logger.error(f"批次哈希计算失败: {table} offset={offset}, {e}") return "ERROR" def _find_diff_rows(self, table: str, offset: int, limit: int) -> list[int]: """逐行对比找出差异行的主键ID""" # 此处省略具体实现，核心逻辑是对两个数据库的同一批次数据 # 按主键逐行MD5对比，返回不一致的ID列表 return [] def close(self): """关闭数据库连接""" try: self.mysql_conn.close() self.tidb_conn.close() except Exception as e: logger.error(f"关闭数据库连接失败: {e}")四、迁移过程关键决策与数据
| 决策点 | 方案A | 方案B | 最终选择 | 原因 |
|---|---|---|---|---|
| 全量导出工具 | mysqldump | Dumpling | Dumpling | 并发导出，4小时完成3.5TB vs 36小时 |
| 全量导入工具 | TiDB Lightning | SQL文件导入 | Lightning | 直接生成SST文件，4TB/h导入速度 |
| 增量同步 | TiCDC | DM(Data Migration) | TiCDC | 更低延迟，支持多下游 |
| 读切换策略 | 一次性全切 | 灰度切流（5%→50%→100%） | 灰度切换 | 逐步验证性能，P99延迟异常可回滚 |
| 回滚机制 | 应用双写 | TiCDC反向同步+切换脚本 | TiCDC反向同步 | MySQL持续作为备份，30秒完成回滚 |
最终迁移数据：
| 指标 | 迁移前（MySQL） | 迁移后（TiDB） | 变化 |
|---|---|---|---|
| 数据量 | 3.5TB | 5.2TB（三副本） | 1.49倍（预期内） |
| QPS峰值 | 18000 | 32000 | 78%提升 |
| P99延迟 | 85ms | 12ms | 86%降低 |
| 存储扩展 | 垂直升级（昂贵） | 水平添加TiKV | 弹性 |
| 迁移总耗时 | - | 7天（含验证） | - |
| 业务停机时间 | - | 0秒 | 零停机 |
五、总结
MySQL到TiDB的平滑迁移，成功的关键不在于技术工具的先进性，而在于全流程的风险控制和可回滚保障。三点核心经验：
- 灰度切换是零停机的唯一保障：读流量从5%→50%→100%逐步放量，写流量也同样分步切换。每一步都设置足够的观察窗口（最小6小时），一旦发现P99延迟或错误率异常立即回滚
- 三种数据校验缺一不可：行数校验（秒级发现差异）+ 分片哈希（分钟级定位不一致批次）+ 逐行对比（精确定位差异行）= 数据一致性的多层保障体系
- 回滚脚本必须定期演练：在迁移前进行了3次全流程回滚演练，每次记录回滚耗时和数据恢复状态。保证真实回滚时不是"摸着石头过河"而是"条件反射式操作"
数据库迁移既是对技术方案的考验，更是对运维团队的应急响应能力和风险控制能力的综合检验。
