数据迁移脚本怎样保证数据完整性

wen 实用脚本 2026-07-25 2

从校验到容错的实战指南

目录导读

为什么数据完整性是迁移的核心命脉？
数据完整性面临的三大风险
保证完整性的六大关键策略
- 1 预迁移数据快照与校验
- 2 事务性批量写入机制
- 3 行级与字段级校验算法
- 4 断点续传与幂等性设计
- 5 增量同步与冲突检测
- 6 全链路日志与回滚预案
完整的数据迁移脚本模板（伪代码）
常见问答
让迁移脚本成为守护数据的“忠诚哨兵”

为什么数据完整性是迁移的核心命脉？

数据迁移（如从MySQL迁移到PostgreSQL、从本地存储迁移到云平台）是系统升级、架构重构中的高频操作。但一个不可挽回的错误——比如丢失一条订单记录、一个用户账户余额字段被截断——就可能导致业务损失、客户投诉甚至合规风险。

数据迁移脚本怎样保证数据完整性

数据完整性是指：

准确性：迁移后的数据与源数据完全一致（无增、无减、无修改）
一致性：关联表之间的逻辑关系不受破坏（如外键约束、主键唯一性）
原子性：迁移过程中不会出现“部分成功”的脏数据状态

核心观点：数据迁移脚本不是简单的“读取-写入”，而是一套包含“校验+容错+可回溯”的工程系统。

数据完整性面临的三大风险

网络中断或数据库连接超时：在批量写入时，如果中途断连，可能导致部分数据写入成功,部分失败。
数据类型不兼容：源库与目标库的字段长度、字符集、默认值规则不同（如源库varchar(255)到目标库varchar(100)导致截断）。
并发写入干扰：迁移过程中源库仍在接受新数据写入,导致增量数据未同步。
脚本逻辑错误：比如WHERE条件写错、JOIN多表导致数据重复或遗漏。

保证完整性的六大关键策略

1 预迁移数据快照与校验

操作：

在迁移开始前，对源库执行一次“数据快照”（例如导出CHECKSUM TABLE或计算目标表的总行数与每行哈希值）。
迁移完成后，重新计算目标库的哈希值,对比是否一致。

示例（伪代码）：

# 源库快照
source_checksum = "SELECT GROUP_CONCAT(MD5(CONCAT(id, name, amount))) FROM orders"
# 目标库快照
target_checksum = "SELECT GROUP_CONCAT(MD5(CONCAT(id, name, amount))) FROM orders_target"
if source_checksum != target_checksum:
    raise Exception("数据完整性校验失败")

2 事务性批量写入机制

关键点：

将迁移任务拆分为“一批数据”为一个事务单元（比如每500行一个事务）。
如果该批次写入失败，整个批次回滚,不会产生脏数据。

BEGIN;
INSERT INTO target_table (id, name) VALUES (1,'A'), (2,'B');
COMMIT;  -- 如果COMMIT失败，自动ROLLBACK

3 行级与字段级校验算法

常用手段：

行数校验：源库和目标库统计COUNT(*)一致。
字段哈希校验：对每一行数据生成MD5/SHA256哈希值,比对源与目标。
范围校验：检查金额字段的值是否在预期范围内（如金额>0）。

脚本示例：

def verify_row(source_row, target_row):
    for col in source_row.keys():
        if source_row[col] != target_row[col]:
            log_error(f"字段 {col} 不一致: {source_row[col]} vs {target_row[col]}")
            return False
    return True

4 断点续传与幂等性设计

场景：迁移进行到50%时，脚本崩溃，重启后如何不重复、不遗漏？实现：

记录偏移量：将已成功迁移的批次ID/主键最大值保存到一个状态表（如migration_progress）。
幂等写入：目标表的主键唯一约束 + INSERT ... ON DUPLICATE KEY UPDATE（MySQL）或UPSERT（PostgreSQL）。

INSERT INTO target (id, data) VALUES (101, 'value')
ON DUPLICATE KEY UPDATE data = VALUES(data);
-- 如果id已存在，则更新（不会重复插入）

5 增量同步与冲突检测

注意：如果迁移过程中源库仍在写入，需要先做全量迁移，再通过时间戳或变更数据捕获（CDC） 同步增量。

逻辑：记录迁移开始时间戳，之后源库的修改（INSERT/UPDATE/DELETE）单独同步。
冲突处理：如果目标库已存在相同主键，使用“后写入优先”或“人工确认”原则。

6 全链路日志与回滚预案

必要组件：

迁移日志：记录每批次的开始时间、结束时间、成功/失败点数、错误详情。
预生产流量回放：在正式迁移前,用生产环境子集模拟迁移并验证结果。
一键回滚脚本：如果发现数据错误,能立刻从备份恢复或从目标库反向同步回源库。

完整的数据迁移脚本模板（伪代码）

import hashlib
import logging
class DataMigrator:
    def __init__(self, source_conn, target_conn, batch_size=1000):
        self.source = source_conn
        self.target = target_conn
        self.batch_size = batch_size
        self.progress_table = "migration_progress"
    def migrate_table(self, table_name):
        # 1. 预校验：检查源表与目标表结构兼容性
        self.validate_schema_compatibility(table_name)
        # 2. 获取上次迁移进度（支持断点续传）
        last_id = self.get_last_migrated_id(table_name)
        offset = last_id if last_id else 0
        # 3. 循环分批读取源数据
        while True:
            rows = self.source.fetch(f"SELECT * FROM {table_name} WHERE id > {offset} ORDER BY id LIMIT {self.batch_size}")
            if not rows:
                break
            # 4. 事务批量写入目标库
            batch_hashes = []
            try:
                with self.target.transaction():
                    for row in rows:
                        # 生成该行数据的哈希（用于校验）
                        row_hash = hashlib.md5(str(row).encode()).hexdigest()
                        batch_hashes.append(row_hash)
                        # 写入目标表（使用UPSERT实现幂等）
                        self.target.execute(f"INSERT INTO {table_name} (id, col1, col2) VALUES (%s, %s, %s) ON CONFLICT (id) DO UPDATE", row)
                    # 记录批次哈希、完成偏移量
                    self.record_progress(table_name, max(row['id'] for row in rows), hashlib.sha256(''.join(batch_hashes).encode()).hexdigest())
            except Exception as e:
                logging.error(f"Batch failed at offset {offset}, error: {e}")
                # 自动回滚事务；记录失败日志
                self.record_failed_batch(table_name, offset, str(e))
                # 可选择重试或终止
                raise
            offset = max(row['id'] for row in rows)
        # 5. 最终全量校验
        self.final_verification(table_name)