如何编写自动比对数据脚本

本文目录导读：

1. 使用 Python + Pandas（最常用）
2. 数据库数据比对
3. 命令行脚本实现
4. 实时监控比对脚本
5. 使用建议
6. 使用示例

我来介绍几种常见的数据比对脚本编写方法,从简单到复杂：

使用 Python + Pandas（最常用）

Excel/CSV文件比对

import pandas as pd
import numpy as np
def compare_files(file1, file2, key_columns, output_file='差异结果.xlsx'):
    """
    比对两个文件的数据差异
    """
    # 读取文件
    df1 = pd.read_excel(file1)  # 或 pd.read_csv(file1)
    df2 = pd.read_excel(file2)
    # 按主键合并
    merged = df1.merge(df2, on=key_columns, how='outer', 
                       suffixes=('_左表', '_右表'), indicator=True)
    # 找出差异记录
    diff_records = merged[merged['_merge'] != 'both']
    # 找出字段值不同的记录
    value_diffs = []
    if len(diff_records) == 0:
        # 对比共有记录中的字段差异
        common = merged[merged['_merge'] == 'both']
        for i, row in common.iterrows():
            for col in df1.columns:
                if col not in key_columns:
                    col_left = f'{col}_左表'
                    col_right = f'{col}_右表'
                    if row[col_left] != row[col_right]:
                        value_diffs.append(row)
    # 输出结果
    with pd.ExcelWriter(output_file) as writer:
        if len(diff_records) > 0:
            diff_records.to_excel(writer, sheet_name='记录差异', index=False)
        if len(value_diffs) > 0:
            pd.DataFrame(value_diffs).to_excel(writer, sheet_name='字段差异', index=False)
    print(f"比对完成，发现 {len(diff_records)} 条记录差异")
    return output_file

数据库数据比对

MySQL数据库比对

import pymysql
import pandas as pd
class DatabaseComparer:
    def __init__(self, connection_params):
        self.conn_params = connection_params
    def query_to_df(self, sql):
        """执行SQL并返回DataFrame"""
        conn = pymysql.connect(**self.conn_params)
        try:
            return pd.read_sql(sql, conn)
        finally:
            conn.close()
    def compare_tables(self, sql1, sql2, key_columns):
        """比对两个查询结果"""
        df1 = self.query_to_df(sql1)
        df2 = self.query_to_df(sql2)
        # 转换数据类型以避免精度问题
        df1 = df1.astype(str)
        df2 = df2.astype(str)
        # 使用merge进行比对
        merged = df1.merge(df2, on=key_columns, how='outer',
                           suffixes=('_库1', '_库2'), indicator=True)
        # 结果统计
        results = {
            '仅在库1': merged[merged['_merge'] == 'left_only'],
            '仅在库2': merged[merged['_merge'] == 'right_only'],
            '字段差异': self.find_value_diffs(merged, key_columns)
        }
        return results
    def find_value_diffs(self, merged_df, key_columns):
        """找出字段值不同的记录"""
        diffs = []
        for _, row in merged_df[merged_df['_merge'] == 'both'].iterrows():
            for col in merged_df.columns:
                if col not in key_columns and col.endswith('_库1'):
                    base_col = col.replace('_库1', '')
                    col2 = f'{base_col}_库2'
                    if str(row[col]) != str(row[col2]):
                        diffs.append({
                            **{k: row[k] for k in key_columns},
                            '字段': base_col,
                            '库1值': row[col],
                            '库2值': row[col2]
                        })
        return pd.DataFrame(diffs)

命令行脚本实现

通用文件比对脚本

#!/usr/bin/env python3
import argparse
import hashlib
import sys
from pathlib import Path
def file_hash(filepath, chunk_size=8192):
    """计算文件MD5哈希"""
    md5 = hashlib.md5()
    with open(filepath, 'rb') as f:
        while chunk := f.read(chunk_size):
            md5.update(chunk)
    return md5.hexdigest()
def compare_directories(dir1, dir2):
    """比对两个目录中的所有文件"""
    differences = []
    path1 = Path(dir1)
    path2 = Path(dir2)
    files1 = {f.relative_to(path1): f for f in path1.rglob('*') if f.is_file()}
    files2 = {f.relative_to(path2): f for f in path2.rglob('*') if f.is_file()}
    # 检查所有文件和目录
    all_paths = set(files1.keys()) | set(files2.keys())
    for rel_path in all_paths:
        if rel_path not in files1:
            differences.append(f"仅在 {dir2}: {rel_path}")
        elif rel_path not in files2:
            differences.append(f"仅在 {dir1}: {rel_path}")
        else:
            # 比较文件内容
            hash1 = file_hash(files1[rel_path])
            hash2 = file_hash(files2[rel_path])
            if hash1 != hash2:
                differences.append(f"内容不同: {rel_path}")
    return differences
def main():
    parser = argparse.ArgumentParser(description='数据比对工具')
    parser.add_argument('source1', help='第一个数据源路径')
    parser.add_argument('source2', help='第二个数据源路径')
    parser.add_argument('--type', choices=['file', 'dir', 'db'], 
                       default='file', help='比对类型')
    args = parser.parse_args()
    if args.type == 'dir':
        diffs = compare_directories(args.source1, args.source2)
        if diffs:
            print("发现差异：")
            for diff in diffs:
                print(f"  - {diff}")
        else:
            print("目录完全一致")
    else:
        # 简单文件比对
        hash1 = file_hash(args.source1)
        hash2 = file_hash(args.source2)
        if hash1 == hash2:
            print("文件完全一致")
        else:
            print("文件不同")
if __name__ == '__main__':
    main()

实时监控比对脚本

import time
import hashlib
import json
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class FileMonitor:
    """文件和目录监控"""
    def __init__(self, watch_path, compare_path):
        self.watch_path = watch_path
        self.compare_path = compare_path
        self.event_handler = FileSystemEventHandler()
        self.event_handler.on_modified = self.on_modified
        self.event_handler.on_created = self.on_created
    def on_modified(self, event):
        if not event.is_directory:
            print(f"检测到修改: {event.src_path}")
            self.compare_file(event.src_path)
    def on_created(self, event):
        if not event.is_directory:
            print(f"检测到新文件: {event.src_path}")
            self.compare_file(event.src_path)
    def compare_file(self, file_path):
        # 这里可以添加实际比对逻辑
        filename = Path(file_path).name
        compare_file = Path(self.compare_path) / filename
        if compare_file.exists():
            if file_hash(file_path) != file_hash(compare_file):
                print(f"文件 {filename} 与目标文件不同")
            else:
                print(f"文件 {filename} 一致")
    def start_monitoring(self):
        observer = Observer()
        observer.schedule(self.event_handler, self.watch_path, recursive=True)
        observer.start()
        print(f"开始监控 {self.watch_path}...")
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            observer.stop()
        observer.join()

使用建议

数据库连接池（避免频繁连接）

from contextlib import contextmanager
@contextmanager
def get_db_connection(db_config):
    conn = pymysql.connect(**db_config)
    try:
        yield conn
    finally:
        conn.close()

日志记录

import logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('比对日志.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)
logger.info("开始比对...")

分批处理大数据量

def compare_in_batches(df1, df2, batch_size=10000):
    """分批比对大型数据集"""
    results = []
    for i in range(0, len(df1), batch_size):
        batch1 = df1.iloc[i:i+batch_size]
        # 进行批次比对
        # ...
        print(f"已处理 {i+len(batch1)}/{len(df1)} 条记录")
    return results

使用示例

# 调用示例
if __name__ == '__main__':
    # 方式1：比对两个Excel文件
    compare_files('数据源1.xlsx', '数据源2.xlsx', ['ID', '订单号'])
    # 方式2：比对数据库表
    comparer = DatabaseComparer({
        'host': 'localhost',
        'user': 'root',
        'password': 'password',
        'database': 'mydb'
    })
    results = comparer.compare_tables(
        'SELECT * FROM table1',
        'SELECT * FROM table2',
        ['id']
    )
    # 方式3：命令行使用
    # python compare_script.py file1.txt file2.txt

选择哪种方式取决于你的具体需求：

• 文件比对：使用Pandas方案
• 数据库比对：使用DatabaseComparer类
• 实时监控：使用FileMonitor类
• 目录比较大文件：使用哈希比对

脚本还可以根据实际需求扩展,比如添加邮件通知、Webhook报警、图表展示等功能。