本文目录导读：
我来介绍几种在Python中记录数据不一致日志的方法：
使用标准库 logging（推荐）
import logging
from datetime import datetime
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('data_inconsistency.log', encoding='utf-8'),
logging.StreamHandler() # 同时输出到控制台
]
)
logger = logging.getLogger(__name__)
class DataValidator:
def check_data_consistency(self, data1, data2):
"""检查数据一致性"""
inconsistencies = []
# 检查字段不匹配
if data1.get('id') != data2.get('id'):
inconsistencies.append({
'type': 'ID_MISMATCH',
'field': 'id',
'expected': data1.get('id'),
'actual': data2.get('id')
})
# 检查数值不一致
if data1.get('total_amount') != data2.get('total_amount'):
inconsistencies.append({
'type': 'AMOUNT_MISMATCH',
'field': 'total_amount',
'expected': data1.get('total_amount'),
'actual': data2.get('total_amount'),
'difference': data1.get('total_amount', 0) - data2.get('total_amount', 0)
})
# 记录不一致日志
if inconsistencies:
logger.warning(f"发现 {len(inconsistencies)} 条数据不一致")
for issue in inconsistencies:
logger.error(
f"数据不一致 - 类型: {issue['type']}, "
f"字段: {issue['field']}, "
f"期望值: {issue['expected']}, "
f"实际值: {issue['actual']}"
)
return inconsistencies
def log_record_mismatch(self, record_type, source, record_id, details):
"""记录单条记录不一致"""
logger.warning(
f"记录不一致 - 类型: {record_type}, "
f"源: {source}, "
f"ID: {record_id}, "
f"详情: {details}"
)
# 使用示例
validator = DataValidator()
data_source_a = {'id': 1, 'total_amount': 100.50, 'status': 'active'}
data_source_b = {'id': 1, 'total_amount': 99.50, 'status': 'pending'}
inconsistencies = validator.check_data_consistency(data_source_a, data_source_b)
带统计功能的日志记录
import logging
from collections import Counter
from datetime import datetime
import json
class InconsistencyLogger:
def __init__(self, log_file='data_inconsistency.log'):
self.logger = self._setup_logger(log_file)
self.stats = Counter()
self.inconsistencies_history = []
def _setup_logger(self, log_file):
logger = logging.getLogger('DataInconsistency')
logger.setLevel(logging.DEBUG)
# 文件处理器 - 详细日志
file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_handler.setLevel(logging.INFO)
file_format = logging.Formatter('%(asctime)s | %(levelname)s | %(message)s')
file_handler.setFormatter(file_format)
# JSON 格式的专门日志
json_handler = logging.FileHandler('inconsistency_export.json', encoding='utf-8')
json_handler.setLevel(logging.WARNING)
logger.addHandler(file_handler)
logger.addHandler(json_handler)
return logger
def log_inconsistency(self, record_id, fields_issues, severity='WARNING'):
"""
记录数据不一致
Args:
record_id: 记录标识
fields_issues: 字段问题列表 [{'field': 'amount', 'expected': 100, 'actual': 99}, ...]
severity: 严重级别 ERROR/WARNING/INFO
"""
timestamp = datetime.now().isoformat()
# 更新统计
self.stats['total_inconsistencies'] += 1
for issue in fields_issues:
self.stats[f"field_{issue['field']}"] += 1
# 创建日志记录
log_entry = {
'timestamp': timestamp,
'record_id': record_id,
'severity': severity,
'issues': fields_issues,
'total_issues': len(fields_issues)
}
# 根据严重级别记录
if severity == 'ERROR':
self.logger.error(json.dumps(log_entry, ensure_ascii=False))
elif severity == 'WARNING':
self.logger.warning(json.dumps(log_entry, ensure_ascii=False))
else:
self.logger.info(json.dumps(log_entry, ensure_ascii=False))
# 保存到历史记录
self.inconsistencies_history.append(log_entry)
# 控制台输出（可选）
print(f"[{timestamp}] {severity}: 记录 {record_id} 发现 {len(fields_issues)} 个问题")
def log_comparison_result(self, source1_name, source2_name, mismatch_count, details):
"""记录源数据比较结果"""
log_msg = (
f"数据源比较: {source1_name} vs {source2_name} | "
f"不匹配数: {mismatch_count} | "
f"详情: {details}"
)
self.logger.info(log_msg)
print(log_msg)
def get_statistics(self):
"""获取不一致统计"""
return dict(self.stats)
# 使用示例
inconsistency_logger = InconsistencyLogger()
# 模拟数据不一致
issues = [
{'field': 'amount', 'expected': 100.50, 'actual': 99.50, 'difference': 1.00},
{'field': 'status', 'expected': 'active', 'actual': 'inactive'}
]
inconsistency_logger.log_inconsistency('RECORD_001', issues, 'ERROR')
inconsistency_logger.log_comparison_result('DatabaseA', 'DatabaseB', 5, '总记录数不匹配')
# 获取统计
stats = inconsistency_logger.get_statistics()
print(f"不一致统计: {stats}")
数据库比较日志
import logging
import pandas as pd
from datetime import datetime
class DatabaseComparisonLogger:
def __init__(self, source_db_name, target_db_name):
self.logger = self._setup_logger()
self.source_db = source_db_name
self.target_db = target_db_name
def _setup_logger(self):
logger = logging.getLogger('DBComparison')
logger.setLevel(logging.DEBUG)
# 主日志文件
handler = logging.FileHandler('db_comparison.log', encoding='utf-8')
handler.setLevel(logging.INFO)
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
logger.addHandler(handler)
# 错误日志单独文件
error_handler = logging.FileHandler('db_errors.log', encoding='utf-8')
error_handler.setLevel(logging.ERROR)
error_handler.setFormatter(formatter)
logger.addHandler(error_handler)
return logger
def compare_record_count(self, source_count, target_count):
"""比较记录数"""
if source_count != target_count:
self.logger.error(
f"记录数不一致 - {self.source_db}: {source_count}, "
f"{self.target_db}: {target_count}"
)
return False
else:
self.logger.info(
f"记录数一致 - 共 {source_count} 条"
)
return True
def compare_dataframes(self, df_source, df_target, key_column='id'):
"""比较两个DataFrame"""
mismatches = []
# 找出差异
merged = pd.merge(df_source, df_target,
on=key_column, how='outer',
suffixes=('_source', '_target'),
indicator=True)
# 记录缺失记录
missing_in_target = merged[merged['_merge'] == 'left_only']
missing_in_source = merged[merged['_merge'] == 'right_only']
if len(missing_in_target) > 0:
self.logger.warning(
f"在 {self.target_db} 中缺失 {len(missing_in_target)} 条记录"
)
for _, row in missing_in_target.iterrows():
self.logger.warning(
f"缺失记录 - ID: {row[key_column]}, "
f"来源: {self.source_db}"
)
if len(missing_in_source) > 0:
self.logger.warning(
f"在 {self.source_db} 中缺失 {len(missing_in_source)} 条记录"
)
# 比较字段值
common_records = merged[merged['_merge'] == 'both']
for col in df_source.columns:
if col == key_column:
continue
source_col = f"{col}_source"
target_col = f"{col}_target"
if source_col in common_records.columns and target_col in common_records.columns:
diff = common_records[common_records[source_col] != common_records[target_col]]
if len(diff) > 0:
mismatches.append({
'field': col,
'count': len(diff),
'records': diff[[key_column, source_col, target_col]].to_dict('records')
})
self.logger.warning(
f"字段 {col} 有 {len(diff)} 条不一致"
)
return mismatches
# 使用示例
comparison_logger = DatabaseComparisonLogger('DB_Main', 'DB_Backup')
# 模拟记录数比较
comparison_logger.compare_record_count(1000, 998)
# 模拟DataFrame比较
import pandas as pd
df1 = pd.DataFrame({'id': [1, 2, 3], 'value': [100, 200, 300]})
df2 = pd.DataFrame({'id': [1, 2, 4], 'value': [100, 250, 400]})
mismatches = comparison_logger.compare_dataframes(df1, df2)
带统计报表的日志系统
import logging
import json
from datetime import datetime, timedelta
from collections import defaultdict
class AdvancedInconsistencyTracker:
def __init__(self):
self.logger = self._setup_logger()
self.daily_stats = defaultdict(lambda: defaultdict(int))
self.alert_threshold = 10 # 触发告警的不一致数量
def _setup_logger(self):
logger = logging.getLogger('AdvancedTracker')
logger.setLevel(logging.DEBUG)
# 主日志
main_handler = logging.FileHandler('inconsistency_tracker.log')
main_handler.setLevel(logging.INFO)
logger.addHandler(main_handler)
# 详细日志
detail_handler = logging.FileHandler('inconsistency_details.log')
detail_handler.setLevel(logging.DEBUG)
logger.addHandler(detail_handler)
return logger
def log_with_context(self, inconsistency_type, context, details):
"""
记录带上下文的不一致信息
Args:
inconsistency_type: 不一致类型
context: 上下文信息（数据源、表名等）
details: 详细信息
"""
timestamp = datetime.now()
date_key = timestamp.strftime('%Y-%m-%d')
# 更新统计
self.daily_stats[date_key][inconsistency_type] += 1
self.daily_stats[date_key]['total'] += 1
# 记录日志
log_entry = {
'timestamp': timestamp.isoformat(),
'type': inconsistency_type,
'context': context,
'details': details,
'daily_count': self.daily_stats[date_key][inconsistency_type]
}
self.logger.info(json.dumps(log_entry, ensure_ascii=False))
# 检查是否需要告警
if self.daily_stats[date_key]['total'] >= self.alert_threshold:
self._send_alert(date_key)
def _send_alert(self, date_key):
"""发送告警"""
alert_msg = (
f"[ALERT] 数据不一致数量已达到 {self.alert_threshold}!"
f" 日期: {date_key}, "
f"总不一致: {self.daily_stats[date_key]['total']}"
)
self.logger.warning(alert_msg)
print(f"⚠️ {alert_msg}")
def generate_report(self, days=7):
"""生成统计报告"""
report = []
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
current = start_date
while current <= end_date:
date_key = current.strftime('%Y-%m-%d')
stats = self.daily_stats.get(date_key, {'total': 0})
report.append({
'date': date_key,
'total_inconsistencies': stats['total'],
'types': {k: v for k, v in stats.items() if k != 'total'}
})
current += timedelta(days=1)
# 保存报告
report_file = f'inconsistency_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json'
with open(report_file, 'w', encoding='utf-8') as f:
json.dump(report, f, ensure_ascii=False, indent=2)
self.logger.info(f"报告已保存到: {report_file}")
return report
# 使用示例
tracker = AdvancedInconsistencyTracker()
# 模拟不同场景的不一致
inconsistencies = [
('FIELD_MISMATCH', {'source': 'order_db', 'table': 'orders'},
{'field': 'amount', 'expected': 100, 'actual': 99}),
('MISSING_RECORD', {'source': 'user_db', 'table': 'users'},
{'id': '12345', 'expected': 'present', 'actual': 'missing'}),
('RECORD_DUPLICATE', {'source': 'product_db', 'table': 'products'},
{'id': 'P001', 'duplicate_count': 2})
]
for inc_type, context, details in inconsistencies:
tracker.log_with_context(inc_type, context, details)
# 生成报告
report = tracker.generate_report(days=7)
最佳实践建议
- 分级日志：使用不同级别记录不同严重程度的不一致
- 结构化数据：使用JSON格式便于后续分析
- 自动统计：记录不一致的数量和类型统计
- 告警机制：当不一致超过阈值时自动告警
- 定期清理：定期归档和清理旧日志
选择哪种方式取决于你的具体需求,建议在项目初期就建立完善的数据不一致日志系统。
