开发Python工具验证RDBMS到MongoDB迁移数据的完整性与准确性

Hey there! 我之前做过类似的RDBMS到MongoDB的迁移验证工具，结合你的需求（30+表、百万级记录），给你分享一些实用的思路和代码片段，帮你快速上手。

一、核心校验思路选择 #

面对百万级数据，全量逐字段对比效率太低，推荐两种方法结合，兼顾速度与准确性：

• 统计量校验：先快速验证表级的记录数、字段聚合值（比如数值型字段的sum/avg、字符串的distinct count），能快速定位差异较大的表，缩小排查范围
• MD5校验和：对记录级做哈希校验，但要重点解决RDBMS与MongoDB的数据类型转换问题（比如日期、小数），必须统一格式后再计算MD5

二、工具开发步骤分解 #

1. 连接源端与目标端 #

用通用库适配多种数据库，避免重复造轮子：

• RDBMS：用sqlalchemy支持MySQL、PostgreSQL、Oracle等多种数据库，比单一驱动更灵活
• MongoDB：用官方pymongo驱动，注意配置连接池提升批量操作性能

示例代码：

from sqlalchemy import create_engine
from pymongo import MongoClient
import datetime

# 初始化RDBMS连接（以MySQL为例，其他数据库只需修改连接字符串）
rdbms_engine = create_engine("mysql+pymysql://username:password@host:port/db_name")

# 初始化MongoDB连接
mongo_client = MongoClient("mongodb://username:password@host:port/")
mongo_db = mongo_client["target_database"]

2. 批量处理数据，避免内存溢出 #

百万级数据不能一次性加载到内存，必须分页/游标迭代：

• RDBMS：用sqlalchemy的分页查询，比如每次取1000条记录；先通过inspect获取表字段，避免硬编码字段名
• MongoDB：用游标迭代或者skip()+limit()分页，配合batch_size参数优化查询效率

3. 记录级MD5校验的关键细节 #

要保证两端MD5一致，必须统一数据格式：

• 日期类型：统一转为ISO格式字符串（如YYYY-MM-DD HH:MM:SS）
• 小数/浮点数：统一保留相同精度后转字符串，避免存储格式差异导致哈希不同
• 空值：RDBMS的NULL与MongoDB的None统一转为相同标识（比如"NULL"）
• 字段顺序：MongoDB是无序文档，必须按固定字段顺序拼接内容，防止字段顺序不同导致哈希差异

示例生成MD5的函数：

import hashlib
import json

def generate_record_md5(record, field_order):
    sorted_values = []
    for field in field_order:
        value = record.get(field)
        # 处理日期类型
        if isinstance(value, (datetime.date, datetime.datetime)):
            sorted_values.append(value.isoformat())
        # 处理空值
        elif value is None or str(value).upper() == "NULL":
            sorted_values.append("NULL")
        # 处理数值类型，统一精度
        elif isinstance(value, (int, float)):
            sorted_values.append(f"{value:.6f}")
        # 其他类型直接转字符串
        else:
            sorted_values.append(str(value))
    # 拼接后计算MD5
    content = "|".join(sorted_values).encode('utf-8')
    return hashlib.md5(content).hexdigest()

4. 性能优化技巧 #

• 并行处理：用concurrent.futures.ThreadPoolExecutor同时处理多个表，注意数据库连接池大小要匹配线程数
• 增量校验：如果是分批迁移，记录已验证的表/记录范围，下次只校验新增部分
• 抽样校验：全量MD5耗时太长时，可先做统计量校验，再随机抽取1%-5%的记录做MD5校验，平衡速度与准确性

三、验证结果输出 #

生成清晰的可视化报告，方便排查问题：

• 每个表的校验状态（通过/失败）
• 记录数差异（如果有）
• 统计量差异（sum/avg等）
• 抽样失败的记录详情

示例报告生成代码：

def generate_validation_report(results):
    report_content = ["# 数据迁移完整性验证报告"]
    for table_name, result in results.items():
        report_content.append(f"\n## {table_name}")
        if result["status"] == "PASS":
            report_content.append("- 状态：✅ 通过")
            report_content.append(f"- 源端记录数：{result['source_count']} | 目标端记录数：{result['target_count']}")
        else:
            report_content.append("- 状态：❌ 失败")
            report_content.append(f"- 记录数差异：源端{result['source_count']} vs 目标端{result['target_count']}")
            if result.get("sample_failures"):
                report_content.append("- 抽样失败记录示例：")
                for idx, fail in enumerate(result["sample_failures"][:5], 1):
                    report_content.append(f"  {idx}. 源端MD5: {fail['source_md5']} | 目标端MD5: {fail['target_md5']}")
    # 保存为Markdown报告
    with open("validation_report.md", "w", encoding="utf-8") as f:
        f.write("\n".join(report_content))

四、踩过的坑提醒 #

• 数据类型转换：比如MySQL的DECIMAL到MongoDB的NumberDecimal，Python读取时要统一处理，避免浮点数精度丢失
• MongoDB的_id字段：如果迁移时自动生成了_id，不要将其加入MD5计算，除非源端有对应唯一键
• 大字段处理：RDBMS的TEXT/BLOB与MongoDB的String/Binary，读取时要确保内容一致，比如避免编码差异
• 超时问题：批量查询时设置合理的超时时间，避免数据库连接超时

内容的提问来源于stack exchange，提问作者user3058477