📚 课程目录第4章数据一致性修复

4. 数据一致性修复：数据校验方法与修复脚本

轮换失败了，数据到底对不对？

这是我最怕听到的问题。说实话，没有校验就谈修复，那就是盲人摸象。我经历过一次惨痛的教训：某次轮换后，业务方反馈数据对不上，我们直接跑了全量同步，结果把原本正确的数据也覆盖了。嗯，从那以后，我养成了一个习惯——先校验，再修复。

4.1 数据校验方法

校验不是走过场。你得用工具、用算法，把两边的数据掰开揉碎了对比。我个人习惯用三种方法，按场景选。

4.1.1 Checksum 校验

Checksum 说白了就是给数据算个“指纹”。两边指纹一样，数据大概率一致。常用的算法有 MD5、SHA256、CRC32。

算法	输出长度	性能	适用场景
MD5	128 bit	快	小文件、非安全场景
SHA256	256 bit	中等	重要数据、安全校验
CRC32	32 bit	极快	大文件、网络传输

我的习惯：线上环境优先用 CRC32。为什么？因为它快，对 CPU 消耗小。SHA256 虽然更安全，但全量跑一遍能把机器打满。你想想看，轮换失败后系统本来就脆弱，别再雪上加霜了。

举个例子，用 cksum 命令对比两个文件：

# 源端
cksum /data/source/user_table.parquet > source.checksum

# 目标端
cksum /data/target/user_table.parquet > target.checksum

# 对比
diff source.checksum target.checksum

如果 diff 没输出，说明文件一致。有输出？那就要看具体差异了。

4.1.2 对比工具

Checksum 只能告诉你“对不对”，但没法告诉你“哪里不对”。这时候得上对比工具。

我常用的工具就三个：

diff / cmp：Linux 原生，适合小文件。我一般用它做快速验证。
rsync -c：边对比边同步，适合大文件。它会把文件分块计算 checksum，只传差异部分。
pt-table-checksum：MySQL 专用，能在线校验主从数据一致性。我在做数据库轮换时必用。

避坑指南：我曾经用 diff 对比两个 50GB 的文件，结果跑了两个小时还没出结果。后来才发现，diff 会把整个文件加载到内存里。大文件一定要用 rsync -c 或者分块校验，别硬扛。

数据库层面的对比，我一般这么写：

# MySQL 主从校验
pt-table-checksum --databases=user_db --tables=user_info \
  --host=192.168.1.10 --user=checksum --password=******

这个工具会逐行计算 checksum，最后输出差异行数。如果差异为 0，恭喜你，数据一致。

4.2 数据同步策略

校验完了，发现确实有差异。怎么办？同步。但同步不是无脑全量覆盖，你得根据差异量选策略。

4.2.1 增量同步

差异数据量小（比如小于总数据量的 5%），用增量同步。只同步变化的部分，速度快，对系统影响小。

增量同步的核心是“找到变化”。常见方法：

时间戳对比：比较 update_time 字段，只同步最近变更的数据。
Binlog 解析：通过解析数据库的 binlog，回放变更操作。
Checksum 差异列表：先算 checksum，找出不一致的行，再逐行同步。

我写过的一个增量同步脚本片段：

#!/bin/bash
# 增量同步：基于时间戳
SOURCE_DB="user_db"
TARGET_DB="user_db"
LAST_SYNC="2024-01-15 03:00:00"

# 导出增量数据
mysql -h source_host -e "
  SELECT * FROM user_info 
  WHERE update_time > '$LAST_SYNC'
  INTO OUTFILE '/tmp/incremental_data.csv'
"

# 导入目标库
mysql -h target_host -e "
  LOAD DATA LOCAL INFILE '/tmp/incremental_data.csv'
  REPLACE INTO TABLE user_info
"

注意：增量同步依赖时间戳，但时间戳可能不准。比如业务方手动修改了系统时间，或者主从时钟不同步。我曾经遇到过因为 NTP 服务挂了，导致增量同步漏了 3 个小时的数据。所以，增量同步完成后，一定要再做一次小范围的 checksum 验证。

4.2.2 全量同步

差异数据量大（超过 30%），或者数据模型发生了变更，别犹豫，直接全量同步。

全量同步的步骤：

锁表或快照：保证导出时数据一致。MySQL 用 FLUSH TABLES WITH READ LOCK，或者用 mysqldump --single-transaction。
导出数据：用 mysqldump、pg_dump 或直接导出 CSV。
传输数据：用 scp、rsync 或云存储中转。
导入数据：先 truncate 目标表，再导入。
校验：全量同步后必须做全量 checksum。

一个典型的全量同步命令：

# 导出
mysqldump -h source_host --single-transaction \
  --databases user_db --tables user_info > full_dump.sql

# 传输
scp full_dump.sql target_host:/tmp/

# 导入
mysql -h target_host user_db < /tmp/full_dump.sql

我的建议：全量同步时，先把目标表重命名（比如 user_info_old），再创建新表导入。这样如果导入失败，还能快速回滚。别直接 truncate，给自己留条后路。

4.3 数据修复脚本编写

校验和同步是“术”，修复脚本是“道”。一个好的修复脚本，应该能自动发现差异、自动修复、自动验证。

我写修复脚本的模板：

#!/usr/bin/env python3
"""
数据修复脚本模板
功能：对比源端和目标端数据，自动修复差异
"""

import hashlib
import pymysql

# 配置
SOURCE_CONFIG = {'host': 'source_host', 'user': 'root', 'password': '****'}
TARGET_CONFIG = {'host': 'target_host', 'user': 'root', 'password': '****'}
TABLE = 'user_info'
KEY_COLUMN = 'id'

def get_checksum(conn, table, key_value):
    """获取单行数据的 checksum"""
    cursor = conn.cursor()
    cursor.execute(f"SELECT * FROM {table} WHERE {KEY_COLUMN}=%s", (key_value,))
    row = cursor.fetchone()
    if not row:
        return None
    # 将行数据序列化后计算 MD5
    row_str = str(row).encode('utf-8')
    return hashlib.md5(row_str).hexdigest()

def repair():
    """主修复逻辑"""
    source_conn = pymysql.connect(**SOURCE_CONFIG)
    target_conn = pymysql.connect(**TARGET_CONFIG)
    
    # 1. 获取所有主键
    cursor = source_conn.cursor()
    cursor.execute(f"SELECT {KEY_COLUMN} FROM {TABLE}")
    all_keys = [row[0] for row in cursor.fetchall()]
    
    # 2. 逐行校验
    for key in all_keys:
        source_checksum = get_checksum(source_conn, TABLE, key)
        target_checksum = get_checksum(target_conn, TABLE, key)
        
        if source_checksum != target_checksum:
            print(f"[修复] 行 {key} 不一致，开始修复...")
            # 3. 从源端获取完整行数据
            cursor.execute(f"SELECT * FROM {TABLE} WHERE {KEY_COLUMN}=%s", (key,))
            row_data = cursor.fetchone()
            
            # 4. 修复目标端
            target_cursor = target_conn.cursor()
            # 先删除旧数据
            target_cursor.execute(f"DELETE FROM {TABLE} WHERE {KEY_COLUMN}=%s", (key,))
            # 再插入新数据（简化处理，实际需要拼装 INSERT 语句）
            placeholders = ','.join(['%s'] * len(row_data))
            columns = ','.join([desc[0] for desc in cursor.description])
            target_cursor.execute(
                f"INSERT INTO {TABLE} ({columns}) VALUES ({placeholders})",
                row_data
            )
            target_conn.commit()
            print(f"[完成] 行 {key} 修复成功")
    
    source_conn.close()
    target_conn.close()
    print("[结束] 所有数据修复完成")

if __name__ == '__main__':
    repair()

核心要点：

修复脚本必须支持幂等性：同一个脚本跑多次，结果一样。不会因为重复执行导致数据错乱。
必须记录操作日志：修复了哪些行、修复前后的值、修复时间。方便事后审计。
必须支持回滚：修复前备份被修改的行。万一修错了，能快速恢复。

我曾经写过一个修复脚本，上线前忘了加日志。结果半夜跑完，业务方说数据还是不对。我盯着屏幕，完全不知道脚本到底修了哪些行。嗯，那次之后，我的脚本里日志模块永远是第一个写的。

4.4 知识体系总览

下面这张图，是我对数据一致性修复的总结。你写修复方案时，照着这个流程走，基本不会漏。

这张图我贴在了团队的 Wiki 上，每次做数据修复，大家都照着这个流程走。说白了，流程定好了，人就不容易犯错。