一次核心业务数据库迁移的全流程复盘：MySQL到TiDB的平滑迁移方案与回滚机制

一次核心业务数据库迁移的全流程复盘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 Server316C32GSQL解析、计算层通过HAProxy负载均衡PD Server38C16G集群元数据管理、TSO时间戳、调度TiKV Server616C64G 2TB NVMe分布式KV存储、Raft副本TiFlash216C64G 2TB NVMe列式存储副本AP分析查询2.2 兼容性评估核心发现MySQL特性TiDB兼容状态处理方案普通SELECT/INSERT/UPDATE/DELETE完全兼容无需修改AUTO_INCREMENT兼容但不保证连续业务不依赖连续IDFOREIGN 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_nametable, total_rows0, matched0, mismatched0, missing_in_target0, extra_in_target0, duration_seconds0 ) 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 ( fSELECT 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最终选择原因全量导出工具mysqldumpDumplingDumpling并发导出4小时完成3.5TB vs 36小时全量导入工具TiDB LightningSQL文件导入Lightning直接生成SST文件4TB/h导入速度增量同步TiCDCDM(Data Migration)TiCDC更低延迟支持多下游读切换策略一次性全切灰度切流5%→50%→100%灰度切换逐步验证性能P99延迟异常可回滚回滚机制应用双写TiCDC反向同步切换脚本TiCDC反向同步MySQL持续作为备份30秒完成回滚最终迁移数据指标迁移前MySQL迁移后TiDB变化数据量3.5TB5.2TB三副本1.49倍预期内QPS峰值180003200078%提升P99延迟85ms12ms86%降低存储扩展垂直升级昂贵水平添加TiKV弹性迁移总耗时-7天含验证-业务停机时间-0秒零停机五、总结MySQL到TiDB的平滑迁移成功的关键不在于技术工具的先进性而在于全流程的风险控制和可回滚保障。三点核心经验灰度切换是零停机的唯一保障读流量从5%→50%→100%逐步放量写流量也同样分步切换。每一步都设置足够的观察窗口最小6小时一旦发现P99延迟或错误率异常立即回滚三种数据校验缺一不可行数校验秒级发现差异 分片哈希分钟级定位不一致批次 逐行对比精确定位差异行 数据一致性的多层保障体系回滚脚本必须定期演练在迁移前进行了3次全流程回滚演练每次记录回滚耗时和数据恢复状态。保证真实回滚时不是摸着石头过河而是条件反射式操作数据库迁移既是对技术方案的考验更是对运维团队的应急响应能力和风险控制能力的综合检验。