本文目录导读：
对于Python脚本校验全量同步数据完整性，有几种常见且可靠的方法，我会从简单到复杂逐步介绍,你可以根据数据量和业务场景选择。
基于计数和校验和的通用方案
这是最常用且高效的方法,适用于大多数场景。
import hashlib
import pandas as pd
from typing import List, Dict
class DataIntegrityValidator:
"""数据完整性校验器"""
def __init__(self, source_data, target_data):
"""
初始化
:param source_data: 源数据 (DataFrame 或 List)
:param target_data: 目标数据
"""
self.source = source_data
self.target = target_data
def count_check(self) -> Dict:
"""行数校验"""
source_count = len(self.source)
target_count = len(self.target)
return {
"source_count": source_count,
"target_count": target_count,
"match": source_count == target_count,
"diff": source_count - target_count
}
def hash_check(self, columns: List[str] = None) -> Dict:
"""
哈希值校验（逐行或整体）
:param columns: 需要校验的列，None表示所有列
"""
source_hash = self._compute_hash(self.source, columns)
target_hash = self._compute_hash(self.target, columns)
return {
"source_hash": source_hash,
"target_hash": target_hash,
"match": source_hash == target_hash
}
def _compute_hash(self, data, columns=None):
"""计算数据的哈希值"""
if isinstance(data, pd.DataFrame):
df = data[columns] if columns else data
# 先排序以确保一致性
df_sorted = df.sort_values(by=df.columns.tolist())
# 将数据序列化后计算MD5
data_str = df_sorted.to_string(index=False)
return hashlib.md5(data_str.encode()).hexdigest()
else:
# 处理列表数据
sorted_data = sorted(str(item) for item in data)
return hashlib.md5(''.join(sorted_data).encode()).hexdigest()
def checksum_aggregate(self, numeric_columns: List[str]) -> Dict:
"""
聚合校验（对数值列求和）
:param numeric_columns: 需要校验的数值列
"""
results = {}
for col in numeric_columns:
source_sum = self.source[col].sum()
target_sum = self.target[col].sum()
results[col] = {
"source_sum": float(source_sum),
"target_sum": float(target_sum),
"match": abs(source_sum - target_sum) < 0.001, # 浮点数比较
"diff": float(source_sum - target_sum)
}
return results
基于主键的逐行比对（精确校验）
适用于有明确主键的场景,可以定位到具体哪些记录不一致。
class RowLevelValidator:
"""逐行数据校验"""
def __init__(self, source_df: pd.DataFrame, target_df: pd.DataFrame,
primary_key: str):
"""
:param source_df: 源数据
:param target_df: 目标数据
:param primary_key: 主键列名（唯一标识）
"""
self.source = source_df.set_index(primary_key)
self.target = target_df.set_index(primary_key)
self.pk = primary_key
def compare_all_records(self) -> Dict:
"""逐行比较所有记录"""
results = {
"total": len(self.source),
"matched": 0,
"mismatched": [],
"missing_in_target": [],
"extra_in_target": []
}
# 获取主键集合
source_keys = set(self.source.index)
target_keys = set(self.target.index)
# 找出缺失和多出的记录
results["missing_in_target"] = list(source_keys - target_keys)
results["extra_in_target"] = list(target_keys - source_keys)
# 比较共同存在的记录
common_keys = source_keys & target_keys
for key in common_keys:
source_row = self.source.loc[key]
target_row = self.target.loc[key]
if source_row.equals(target_row):
results["matched"] += 1
else:
# 记录差异详情
diff_columns = []
for col in source_row.index:
if source_row[col] != target_row[col]:
diff_columns.append({
"column": col,
"source_value": source_row[col],
"target_value": target_row[col]
})
results["mismatched"].append({
"key": key,
"differences": diff_columns
})
return results
def summary_report(self) -> str:
"""生成校验报告摘要"""
results = self.compare_all_records()
report = f"""
===== 数据完整性校验报告 =====
总记录数: {results['total']}
匹配成功: {results['matched']}
匹配失败: {len(results['mismatched'])}
目标缺失: {len(results['missing_in_target'])}
目标多余: {len(results['extra_in_target'])}
校验结果: {'✅ 通过' if results['matched'] == results['total'] else '❌ 失败'}
"""
return report
高效批量校验（大文件场景）
对于超大文件,使用分块读取和增量校验。
import mmap
class LargeFileValidator:
"""大文件高效校验"""
@staticmethod
def validate_large_csv(source_path: str, target_path: str,
chunk_size: int = 10000) -> Dict:
"""
逐块校验大CSV文件
"""
import csv
results = {
"source_lines": 0,
"target_lines": 0,
"block_hashes": [],
"total_match": True
}
with open(source_path, 'r') as sf, open(target_path, 'r') as tf:
source_reader = csv.reader(sf)
target_reader = csv.reader(tf)
# 比较表头
source_header = next(source_reader)
target_header = next(target_reader)
results["header_match"] = source_header == target_header
while True:
source_chunk = list(itertools.islice(source_reader, chunk_size))
target_chunk = list(itertools.islice(target_reader, chunk_size))
if not source_chunk and not target_chunk:
break
# 记录行数
results["source_lines"] += len(source_chunk)
results["target_lines"] += len(target_chunk)
# 计算块哈希
source_hash = hashlib.md5(str(source_chunk).encode()).hexdigest()
target_hash = hashlib.md5(str(target_chunk).encode()).hexdigest()
chunk_match = source_hash == target_hash
results["block_hashes"].append({
"block": len(results["block_hashes"]) + 1,
"match": chunk_match
})
if not chunk_match:
results["total_match"] = False
return results
@staticmethod
def fast_file_hash(file_path: str) -> str:
"""快速计算文件整体哈希（适合中小文件）"""
hash_md5 = hashlib.md5()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hash_md5.update(chunk)
return hash_md5.hexdigest()
实际使用示例
import pandas as pd
def validate_data_sync():
"""完整的数据同步校验示例"""
# 1. 准备测试数据
source_df = pd.DataFrame({
'id': [1, 2, 3, 4],
'name': ['A', 'B', 'C', 'D'],
'amount': [100.0, 200.0, 300.0, 400.0]
})
target_df = pd.DataFrame({
'id': [1, 2, 3, 4, 5], # 多了一条
'name': ['A', 'B', 'C', 'E'], # D变成E（错误）
'amount': [100.0, 200.0, 300.0, 400.0]
})
# 2. 执行多重校验
validator = DataIntegrityValidator(source_df, target_df)
# 行数校验
count_result = validator.count_check()
print(f"行数校验: {count_result}")
# 哈希校验
hash_result = validator.hash_check()
print(f"哈希校验: {hash_result}")
# 聚合校验
agg_result = validator.checksum_aggregate(['amount'])
print(f"聚合校验: {agg_result}")
# 3. 逐行精确校验（如果有主键）
row_validator = RowLevelValidator(source_df, target_df, 'id')
compare_result = row_validator.compare_all_records()
print(f"逐行比对: {compare_result}")
# 4. 生成报告
report = row_validator.summary_report()
print(report)
# 执行校验
validate_data_sync()
最佳实践建议
选择策略
- 小型数据（<10万行）：使用逐行比对，完整校验
- 中型数据（10万-100万行）：哈希校验 + 抽样校验
- 大型数据（>100万行）：分块校验 + 聚合校验
推荐的多层校验策略
def comprehensive_validation(source, target, primary_key=None):
"""
综合校验策略：多维度验证
"""
report = {
"timing": {"start": None, "end": None},
"checks": {}
}
# 1. 快速校验层
report["checks"]["row_count"] = check_row_count(source, target)
report["checks"]["columns"] = check_columns(source, target)
# 2. 聚合校验层
report["checks"]["sum_checks"] = check_numeric_sums(source, target)
# 3. 精确校验层（可选，根据数据量）
if len(source) < 100000:
report["checks"]["full_comparison"] = full_compare(source, target, primary_key)
else:
# 大文件采用抽样校验
report["checks"]["sample_check"] = sample_check(source, target, 0.1)
return report
通过组合使用这些方法，你可以实现高效且可靠的数据完整性校验，建议根据实际情况选择合适的策略,必要时可以组合多层校验来确保数据同步的准确性。
