如何用Python对比Azure SQL开发测试生产三环境数据并生成校验报告

核心疑问解答 #

• 数据存储问题：临时对比数据直接存在pandas DataFrame中即可，无需额外中间存储。如果需要留存历史校验记录，可以每次跑完将原始行数、对比结果追加存储到本地CSV文件，或专门用于数据校验的数据库表中。
• 对比逻辑选择：直接一次性合并三个环境的行数数据做对比即可，无需两两对比。既匹配你需要的报告格式，后续扩展环境时也不需要修改对比逻辑，效率更高。

整体拆分为4个低耦合的模块即可，满足后续表新增、环境新增的扩展需求：

1. 配置管理模块：统一维护三个环境的连接配置，避免硬编码在业务逻辑中
2. 通用查询模块：封装单环境表行数查询逻辑，复用代码，避免重复写三次连接查询逻辑
3. 数据合并对比模块：将三个环境的查询结果合并为宽表，批量判断行数一致性
4. 报告生成模块：输出符合要求格式的校验报告，支持扩展不同格式（CSV/Excel/HTML等）

首先安装依赖：

pip install pyodbc pandas

完整可运行代码：

import pyodbc
import pandas as pd

# --------------- 配置模块：修改为自己的环境配置即可 ---------------
ENV_CONFIG = {
    "Dev": {
        "server": "dev_sql_server地址",
        "database": "数据库名",
        "driver": "{ODBC Driver 17 for SQL Server}", # 也可以用你当前的{SQL Server}
        "trusted_connection": "yes"
        # 如果是账密登录，新增uid、pwd参数即可
    },
    "Test": {
        "server": "test_sql_server地址",
        "database": "数据库名",
        "driver": "{ODBC Driver 17 for SQL Server}",
        "trusted_connection": "yes"
    },
    "Prod": {
        "server": "prod_sql_server地址",
        "database": "数据库名",
        "driver": "{ODBC Driver 17 for SQL Server}",
        "trusted_connection": "yes"
    }
}

# 行数查询通用SQL
COUNT_SQL = """
SELECT (SCHEMA_NAME(A.schema_id) + '.' + A.Name) AS TableName,
SUM(B.rows) AS RecordCount
FROM sys.objects A
INNER JOIN sys.partitions B ON A.object_id = B.object_id
WHERE A.type = 'U'
GROUP BY A.schema_id, A.Name
"""

# --------------- 通用查询模块 ---------------
def get_env_row_count(env_config: dict) -> pd.DataFrame:
    """获取单个环境的所有用户表行数"""
    conn_str = (
        f'Driver={env_config["driver"]};'
        f'Server={env_config["server"]};'
        f'Database={env_config["database"]};'
        f'Trusted_Connection={env_config["trusted_connection"]};'
    )
    # 异常捕获避免单环境连接失败导致整个脚本终止
    try:
        with pyodbc.connect(conn_str, timeout=10) as conn:
            df = pd.read_sql_query(COUNT_SQL, conn)
        return df
    except Exception as e:
        print(f"连接{env_config.get('server')}失败，错误信息：{str(e)}")
        return pd.DataFrame(columns=["TableName", "RecordCount"])

# --------------- 合并对比模块 ---------------
if __name__ == "__main__":
    # 批量获取所有环境的行数数据
    env_data = {}
    for env_name, config in ENV_CONFIG.items():
        df = get_env_row_count(config)
        env_data[env_name] = df.rename(columns={"RecordCount": env_name})

    # 合并所有环境的数据为一张宽表，outer join可以识别某个环境缺失的表
    merge_df = None
    for env_name, df in env_data.items():
        if merge_df is None:
            merge_df = df
        else:
            merge_df = pd.merge(merge_df, df, on="TableName", how="outer")
    
    # 空值填充为0（代表该环境无此表或行数为0）
    merge_df = merge_df.fillna(0)
    # 新增一致性校验列
    merge_df["是否一致"] = merge_df.apply(lambda row: row["Dev"] == row["Test"] == row["Prod"], axis=1)

    # --------------- 报告生成模块 ---------------
    # 输出你需要的CSV格式，quoting=1代表所有字段加双引号，完全匹配你的示例
    report_df = merge_df[["TableName", "Dev", "Test", "Prod"]]
    report_df.to_csv("表行数校验报告.csv", index=False, quoting=1, encoding="utf-8-sig")

    # 控制台打印差异统计
    total_table = len(merge_df)
    inconsistent_table = len(merge_df[merge_df["是否一致"] == False])
    print(f"校验完成，总表数：{total_table}，不一致表数：{inconsistent_table}")
    if inconsistent_table > 0:
        print("不一致的表清单：")
        print(merge_df[merge_df["是否一致"] == False][["TableName", "Dev", "Test", "Prod"]])

• 如果需要留存历史校验记录，可在合并后的merge_df中新增校验时间列，每次跑完追加写入到历史校验CSV或者专门的校验数据库表中
• 可以将输出格式改为带高亮的Excel报告，用pandas.Styler给不一致的行标红，更直观
• 如果后续表数量增长到上千级别，可以把对比逻辑改为numpy向量化运算，比apply效率更高

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