Python脚本如何校验全量同步数据完整

wen python案例 2026-06-26 242

本文目录导读：

基于计数和校验和的通用方案
基于主键的逐行比对（精确校验）
高效批量校验（大文件场景）
实际使用示例
最佳实践建议

对于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()