本文目录导读：
我来详细介绍几种生成数据比对报告的方法,根据不同的使用场景和技术栈：
Python 脚本方案（最常用）
基础对比脚本
import pandas as pd
import numpy as np
from datetime import datetime
def compare_dataframes(df1, df2, key_columns, report_name="comparison_report"):
"""
比对两个DataFrame并生成报告
"""
report = []
report.append(f"=== 数据比对报告 - {datetime.now()} ===\n")
# 1. 基本信息
report.append(f"源数据行数: {len(df1)}")
report.append(f"目标数据行数: {len(df2)}")
# 2. 行数差异
only_in_df1 = df1[~df1[key_columns].isin(df2[key_columns])]
only_in_df2 = df2[~df2[key_columns].isin(df1[key_columns])]
report.append(f"\n仅在源数据中的记录: {len(only_in_df1)}")
report.append(f"仅在目标数据中的记录: {len(only_in_df2)}")
# 3. 字段值对比
common = pd.merge(df1, df2, on=key_columns, suffixes=('_source', '_target'))
diff_columns = [col for col in df1.columns if col not in key_columns]
for col in diff_columns:
source_col = f"{col}_source"
target_col = f"{col}_target"
if source_col in common.columns and target_col in common.columns:
differences = common[common[source_col] != common[target_col]]
if len(differences) > 0:
report.append(f"\n字段 '{col}' 差异数: {len(differences)}")
# 可以输出具体差异
return "\n".join(report)
# 使用示例
df1 = pd.read_csv('source.csv')
df2 = pd.read_csv('target.csv')
report_text = compare_dataframes(df1, df2, ['id'])
print(report_text)
生成 HTML 报告
def generate_html_report(df1, df2, key_columns):
"""生成可视化的HTML比对报告"""
html_parts = ["""
<html>
<head>
<style>
body { font-family: Arial; padding: 20px; }
.summary { background: #f0f0f0; padding: 10px; margin: 10px 0; }
table { border-collapse: collapse; width: 100%; }
th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
.match { background-color: #d4edda; }
.mismatch { background-color: #f8d7da; }
</style>
</head>
<body>
<h1>数据比对报告</h1>
"""]
# 添加统计信息
html_parts.append(f"""
<div class="summary">
<h3>统计摘要</h3>
<p>源数据: {len(df1)} 行</p>
<p>目标数据: {len(df2)} 行</p>
<p>匹配: {len(pd.merge(df1, df2, on=key_columns))} 行</p>
</div>
""")
html_parts.append("</body></html>")
return "\n".join(html_parts)
Excel 报告生成
import openpyxl
from openpyxl.styles import PatternFill, Font
from openpyxl.utils.dataframe import dataframe_to_rows
def create_excel_comparison_report(df1, df2, key_columns, output_file):
"""生成Excel格式的比对报告"""
wb = openpyxl.Workbook()
# Sheet 1: 汇总信息
ws_summary = wb.active
ws_summary.title = "汇总"
ws_summary['A1'] = "数据比对报告"
# Sheet 2: 差异详情
ws_diff = wb.create_sheet("差异详情")
# 添加数据
diff_df = find_differences(df1, df2, key_columns)
for r in dataframe_to_rows(diff_df, index=False, header=True):
ws_diff.append(r)
# 标记差异单元格
red_fill = PatternFill(start_color="FF0000", end_color="FF0000", fill_type="solid")
green_fill = PatternFill(start_color="00FF00", end_color="00FF00", fill_type="solid")
for row in ws_diff.iter_rows(min_row=2):
for cell in row:
if isinstance(cell.value, str) and "DIFF" in cell.value:
cell.fill = red_fill
wb.save(output_file)
Shell 脚本方案（Linux/Unix）
#!/bin/bash
# 数据比对脚本
generate_comparison_report() {
local file1=$1
local file2=$2
local report_file="comparison_report_$(date +%Y%m%d).txt"
echo "=== 数据比对报告 - $(date) ===" > $report_file
echo "" >> $report_file
# 行数对比
echo "文件1行数: $(wc -l < $file1)" >> $report_file
echo "文件2行数: $(wc -l < $file2)" >> $report_file
# 差异对比
echo "" >> $report_file
echo "=== 差异分析 ===" >> $report_file
diff $file1 $file2 > /tmp/diff_out.txt 2>&1
cat /tmp/diff_out.txt >> $report_file
echo "报告已生成: $report_file"
}
# 使用
generate_comparison_report "data_source.csv" "data_target.csv"
高级功能脚本
包含智能匹配和统计
import json
from typing import Dict, List, Any
class DataComparisonReport:
def __init__(self):
self.report = {
"summary": {},
"details": [],
"statistics": {}
}
def compare_json_objects(self, obj1: Dict, obj2: Dict, path: str = ""):
"""深度对比JSON对象"""
differences = []
all_keys = set(list(obj1.keys()) + list(obj2.keys()))
for key in all_keys:
current_path = f"{path}.{key}" if path else key
if key not in obj1:
differences.append({
"path": current_path,
"status": "missing_in_source",
"value": obj2[key]
})
elif key not in obj2:
differences.append({
"path": current_path,
"status": "missing_in_target",
"value": obj1[key]
})
elif obj1[key] != obj2[key]:
differences.append({
"path": current_path,
"status": "mismatch",
"source_value": obj1[key],
"target_value": obj2[key]
})
return differences
def export_report(self, format: str = "json"):
"""导出报告"""
if format == "json":
return json.dumps(self.report, indent=2, ensure_ascii=False)
elif format == "text":
return self._format_text_report()
def _format_text_report(self):
"""格式化为文本报告"""
lines = ["="*50, "数据比对报告", "="*50]
# 格式化逻辑
return "\n".join(lines)
实用示例：数据库表对比
import sqlite3
import mysql.connector
def compare_database_tables(db_config1, db_config2, table_name):
"""对比两个数据库中的表"""
# 连接数据库
conn1 = mysql.connector.connect(**db_config1)
conn2 = mysql.connector.connect(**db_config2)
# 读取数据
df1 = pd.read_sql(f"SELECT * FROM {table_name}", conn1)
df2 = pd.read_sql(f"SELECT * FROM {table_name}", conn2)
# 生成报告
report = compare_dataframes(df1, df2, ['id'])
# 保存报告
with open(f"db_comparison_{table_name}.txt", "w", encoding="utf-8") as f:
f.write(report)
conn1.close()
conn2.close()
使用建议
-
选择合适的工具：
- 简单对比 → Shell 脚本
- 数据分析 → Python pandas
- Web 展示 → HTML/JS 方案
- 企业级 → 商业工具如 Tableau
-
建议：
- 总览统计
- 差异明细
- 异常标记
- 时间戳和版本信息
-
自动化部署：
# crontab 定时任务 0 8 * * * /path/to/compare_script.py
需要我针对具体的使用场景,提供更详细的脚本实现吗？
