本文目录导读：
我来介绍几种常见的数据比对脚本编写方法,从简单到复杂：
使用 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报警、图表展示等功能。
