本文目录导读：
我来为您提供监控数据校验失败脚本的编写方案,涵盖基础到进阶的多种场景。
基础数据校验监控脚本
1 文件内容校验监控
#!/bin/bash
# 文件数据校验监控脚本
# 配置变量
LOG_FILE="/var/log/data_validation.log"
NOTIFY_EMAIL="admin@example.com"
THRESHOLD_FAILURE=5 # 允许的最大失败次数
# 日志函数
log_message() {
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" >> $LOG_FILE
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1"
}
# 发送告警
send_alert() {
local message=$1
echo "$message" | mail -s "数据校验失败告警" $NOTIFY_EMAIL
# 也可以集成企业微信、钉钉等
# curl -X POST -H "Content-Type: application/json" -d '{"msgtype":"text","text":{"content":"'$message'"}}' YOUR_WEBHOOK_URL
}
# 校验数据格式
validate_data_format() {
local file=$1
local expected_fields=$2
local delimiter=$3
if [ ! -f "$file" ]; then
log_message "ERROR: 文件 $file 不存在"
return 1
fi
local line_num=0
local failed_lines=0
while IFS= read -r line; do
((line_num++))
# 跳过空行和注释行
[ -z "$line" ] && continue
[[ "$line" =~ ^# ]] && continue
# 检查字段数量
fields_count=$(echo "$line" | awk -F"$delimiter" '{print NF}')
if [ "$fields_count" -ne "$expected_fields" ]; then
log_message "ERROR: 第 $line_num 行字段数异常 (期望: $expected_fields, 实际: $fields_count)"
((failed_lines++))
fi
# 检查关键字段不能为空
IFS="$delimiter" read -ra fields <<< "$line"
for i in "${!fields[@]}"; do
if [ -z "${fields[$i]}" ] && [ $i -lt 3 ]; then # 前3个字段不能为空
log_message "ERROR: 第 $line_num 行第 $((i+1)) 个字段为空"
((failed_lines++))
fi
done
done < "$file"
return $failed_lines
}
# 校验数值范围
validate_numeric_range() {
local value=$1
local min=$2
local max=$3
local field_name=$4
if ! [[ "$value" =~ ^[0-9]+(\.[0-9]+)?$ ]]; then
log_message "ERROR: $field_name 不是有效数值: $value"
return 1
fi
if (( $(echo "$value < $min" | bc -l) )) || (( $(echo "$value > $max" | bc -l) )); then
log_message "ERROR: $field_name 超出范围 [$min, $max]: $value"
return 1
fi
return 0
}
# 主校验函数
main_validation() {
local data_file="/path/to/data.csv"
local total_failures=0
log_message "开始数据校验..."
# 执行各种校验
validate_data_format "$data_file" 5 ","
if [ $? -gt 0 ]; then
((total_failures+=$?))
fi
# 检查文件完整性
local expected_lines=1000
local actual_lines=$(wc -l < "$data_file")
if [ "$actual_lines" -lt "$expected_lines" ]; then
log_message "ERROR: 数据行数不足 (期望: $expected_lines, 实际: $actual_lines)"
((total_failures++))
fi
# 检查文件更新时间
local file_age=$(stat -c %Y "$data_file")
local current_time=$(date +%s)
local time_diff=$(( (current_time - file_age) / 60 ))
if [ "$time_diff" -gt 60 ]; then # 超过60分钟未更新
log_message "WARNING: 数据文件已 $time_diff 分钟未更新"
((total_failures++))
fi
# 判断是否需要告警
if [ "$total_failures" -gt 0 ]; then
local alert_msg="数据校验失败次数: $total_failures\n请检查日志: $LOG_FILE"
send_alert "$alert_msg"
return 1
fi
log_message "数据校验完成，全部通过"
return 0
}
# 执行主函数
main_validation
exit $?
2 数据库数据校验监控
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据库数据校验监控脚本
"""
import pymysql
import logging
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime, timedelta
import json
class DataValidationMonitor:
def __init__(self, config_file='config.json'):
with open(config_file, 'r') as f:
self.config = json.load(f)
self.setup_logging()
self.failures = []
def setup_logging(self):
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('/var/log/data_validation_monitor.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger(__name__)
def connect_db(self):
"""连接数据库"""
try:
connection = pymysql.connect(
host=self.config['db_host'],
user=self.config['db_user'],
password=self.config['db_password'],
database=self.config['db_name'],
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
return connection
except Exception as e:
self.logger.error(f"数据库连接失败: {e}")
return None
# 检查数据完整性
def check_data_integrity(self):
"""检查数据完整性"""
self.logger.info("开始检查数据完整性...")
connection = self.connect_db()
if not connection:
return
try:
with connection.cursor() as cursor:
# 检查主键重复
cursor.execute("""
SELECT table_name, COUNT(*) as duplicate_count
FROM (
SELECT table_name, COUNT(*) as cnt
FROM information_schema.tables
WHERE table_schema = %s
GROUP BY table_name, id
HAVING cnt > 1
) t
GROUP BY table_name
""", (self.config['db_name'],))
duplicates = cursor.fetchall()
if duplicates:
for record in duplicates:
msg = f"表 {record['table_name']} 存在 {record['duplicate_count']} 条重复主键"
self.failures.append(msg)
self.logger.warning(msg)
# 检查外键完整性
cursor.execute("""
SELECT
tc.table_name,
tc.constraint_name
FROM information_schema.table_constraints tc
JOIN information_schema.referential_constraints rc
ON tc.constraint_name = rc.constraint_name
WHERE tc.constraint_type = 'FOREIGN KEY'
AND tc.table_schema = %s
""", (self.config['db_name'],))
constraints = cursor.fetchall()
for constraint in constraints:
cursor.execute(f"""
SELECT COUNT(*) as orphan_count
FROM {constraint['table_name']}
WHERE {constraint['constraint_name']}_id NOT IN (
SELECT id FROM referenced_table
)
""")
orphan_result = cursor.fetchone()
if orphan_result and orphan_result['orphan_count'] > 0:
msg = f"表 {constraint['table_name']} 存在 {orphan_result['orphan_count']} 条孤立记录"
self.failures.append(msg)
self.logger.warning(msg)
except Exception as e:
self.logger.error(f"数据完整性检查失败: {e}")
self.failures.append(f"数据完整性检查异常: {e}")
finally:
connection.close()
# 检查数据一致性
def check_data_consistency(self):
"""检查跨表数据一致性"""
self.logger.info("开始检查数据一致性...")
connection = self.connect_db()
if not connection:
return
try:
with connection.cursor() as cursor:
# 检查汇总数据一致性
checks = [
{
'name': '订单金额合计',
'sql': """
SELECT
(SELECT COALESCE(SUM(amount), 0) FROM orders) as total_orders,
(SELECT COALESCE(SUM(amount), 0) FROM order_items) as total_items
"""
},
{
'name': '用户统计一致性',
'sql': """
SELECT
(SELECT COUNT(*) FROM users) as user_count,
(SELECT COUNT(DISTINCT user_id) FROM user_profiles) as profile_count
"""
}
]
for check in checks:
cursor.execute(check['sql'])
result = cursor.fetchone()
if result and 'total_orders' in result:
if abs(float(result['total_orders']) - float(result['total_items'])) > 0.01:
msg = f"数据一致性检查失败: {check['name']}"
self.failures.append(msg)
self.logger.warning(f"{msg} - 差异: {abs(float(result['total_orders']) - float(result['total_items']))}")
except Exception as e:
self.logger.error(f"数据一致性检查失败: {e}")
self.failures.append(f"数据一致性检查异常: {e}")
finally:
connection.close()
# 检查数据时效性
def check_data_timeliness(self):
"""检查数据时效性"""
self.logger.info("开始检查数据时效性...")
connection = self.connect_db()
if not connection:
return
try:
with connection.cursor() as cursor:
threshold = datetime.now() - timedelta(hours=1)
cursor.execute("""
SELECT
table_name,
MAX(update_time) as last_update
FROM information_schema.tables
WHERE table_schema = %s
AND table_type = 'BASE TABLE'
GROUP BY table_name
""", (self.config['db_name'],))
tables = cursor.fetchall()
for table in tables:
if table['last_update'] < threshold:
minutes_diff = (datetime.now() - table['last_update']).total_seconds() / 60
msg = f"表 {table['table_name']} 已 {int(minutes_diff)} 分钟未更新"
if minutes_diff > 120: # 超过2小时告警
self.failures.append(msg)
self.logger.warning(msg)
except Exception as e:
self.logger.error(f"数据时效性检查失败: {e}")
finally:
connection.close()
# 发送告警
def send_alert(self):
"""发送告警通知"""
if not self.failures:
return
subject = f"数据校验失败告警 - {len(self.failures)} 个问题"
body = "以下数据校验失败:\n\n"
body += "\n".join([f"- {failure}" for failure in self.failures])
# 邮件告警
if self.config.get('email_enabled'):
try:
msg = MIMEMultipart()
msg['From'] = self.config['email_from']
msg['To'] = ', '.join(self.config['alert_emails'])
msg['Subject'] = subject
msg.attach(MIMEText(body, 'plain', 'utf-8'))
with smtplib.SMTP(self.config['smtp_server'], self.config['smtp_port']) as server:
server.starttls()
server.login(self.config['smtp_user'], self.config['smtp_password'])
server.send_message(msg)
self.logger.info("告警邮件发送成功")
except Exception as e:
self.logger.error(f"告警邮件发送失败: {e}")
# 写入告警日志
with open('/var/log/data_alert.log', 'a') as f:
f.write(f"{datetime.now()} - {subject}\n{body}\n\n")
def run(self):
"""运行所有校验"""
self.logger.info("="*50)
self.logger.info("开始数据校验监控")
self.logger.info("="*50)
# 执行各类校验
self.check_data_integrity()
self.check_data_consistency()
self.check_data_timeliness()
# 发送告警
if self.failures:
self.send_alert()
self.logger.warning(f"发现 {len(self.failures)} 个问题")
else:
self.logger.info("所有校验通过")
self.logger.info("数据校验监控完成")
return len(self.failures)
if __name__ == "__main__":
monitor = DataValidationMonitor('config.json')
exit_code = monitor.run()
exit(exit_code)
高级监控方案
1 带监控看板的完整方案
#!/usr/bin/env python3
"""
完整的API数据校验监控系统
"""
import requests
import time
import json
from statistics import mean, median
from collections import defaultdict
class APIResponseValidator:
def __init__(self, config):
self.config = config
self.metrics = defaultdict(list)
self.failures = []
def validate_response_time(self, url, response):
"""验证响应时间"""
response_time = response.elapsed.total_seconds() * 1000 # ms
self.metrics['response_time'].append(response_time)
if response_time > self.config['max_response_time']:
self.failures.append({
'type': 'response_time',
'url': url,
'value': response_time,
'threshold': self.config['max_response_time']
})
def validate_response_status(self, url, response):
"""验证响应状态码"""
if response.status_code != 200:
self.failures.append({
'type': 'status_code',
'url': url,
'value': response.status_code,
'expected': 200
})
def validate_response_schema(self, url, response, expected_schema):
"""验证响应JSON Schema"""
try:
data = response.json()
# 简洁的模式验证
self._validate_schema_recursive(data, expected_schema, url)
except ValueError:
self.failures.append({
'type': 'invalid_json',
'url': url,
'value': response.text[:100]
})
def _validate_schema_recursive(self, data, schema, url, path=''):
"""递归验证数据结构"""
for key, rules in schema.items():
current_path = f"{path}.{key}" if path else key
if key not in data:
if rules.get('required', False):
self.failures.append({
'type': 'missing_field',
'url': url,
'field': current_path
})
continue
value = data[key]
# 类型检查
if 'type' in rules:
if rules['type'] == 'int' and not isinstance(value, int):
self.failures.append({
'type': 'type_mismatch',
'url': url,
'field': current_path,
'expected': rules['type'],
'actual': type(value).__name__
})
# 范围检查
if 'min' in rules and value < rules['min']:
self.failures.append({
'type': 'value_out_of_range',
'url': url,
'field': current_path,
'value': value,
'min': rules['min']
})
if 'max' in rules and value > rules['max']:
self.failures.append({
'type': 'value_out_of_range',
'url': url,
'field': current_path,
'value': value,
'max': rules['max']
})
# 递归检查子结构
if 'properties' in rules and isinstance(value, dict):
self._validate_schema_recursive(value, rules['properties'], url, current_path)
def validate_data_consistency(self, api_responses):
"""验证多个API之间的数据一致性"""
# 示例：检查不同API返回的用户ID是否一致
user_ids = []
for response in api_responses:
try:
data = response.json()
if 'user_id' in data:
user_ids.append(data['user_id'])
except:
continue
if len(user_ids) > 1 and len(set(user_ids)) != 1:
self.failures.append({
'type': 'data_consistency',
'desc': '不同API返回的用户ID不一致',
'values': user_ids
})
def run_validation(self, endpoints):
"""运行所有验证"""
all_responses = []
for endpoint in endpoints:
try:
response = requests.get(
endpoint['url'],
headers=endpoint.get('headers', {}),
timeout=self.config.get('timeout', 10)
)
all_responses.append(response)
self.validate_response_time(endpoint['url'], response)
self.validate_response_status(endpoint['url'], response)
if 'schema' in endpoint:
self.validate_response_schema(endpoint['url'], response, endpoint['schema'])
except requests.exceptions.RequestException as e:
self.failures.append({
'type': 'request_error',
'url': endpoint['url'],
'error': str(e)
})
return self.failures, self.metrics
# 使用示例
if __name__ == "__main__":
config = {
'max_response_time': 500, # ms
'timeout': 30
}
validator = APIResponseValidator(config)
endpoints = [
{
'url': 'https://api.example.com/users/1',
'headers': {'Authorization': 'Bearer token123'},
'schema': {
'id': {'type': 'int', 'required': True},
'name': {'type': 'str', 'required': True},
'email': {'type': 'str', 'required': True},
'age': {'type': 'int', 'min': 0, 'max': 150}
}
}
]
failures, metrics = validator.run_validation(endpoints)
if failures:
print(f"发现 {len(failures)} 个问题:")
for failure in failures:
print(json.dumps(failure, indent=2))
2 配置文件示例
{
"db_host": "localhost",
"db_user": "monitor",
"db_password": "secure_password",
"db_name": "production_db",
"email_enabled": true,
"email_from": "monitor@example.com",
"alert_emails": ["admin@example.com", "developer@example.com"],
"smtp_server": "smtp.example.com",
"smtp_port": 587,
"smtp_user": "monitor@example.com",
"smtp_password": "smtp_password",
"max_response_time": 500,
"timeout": 30,
"check_interval": 300,
"alert_threshold": 3,
"excluded_tables": ["temp_data", "cache"]
}
自动化部署与执行
1 Cron任务配置
# 每5分钟执行一次数据校验 */5 * * * * /usr/local/bin/data_validation.sh # 每天凌晨2点执行完整性检查 0 2 * * * /usr/local/bin/db_integrity_check.sh # 每30分钟执行API响应验证 */30 * * * * /usr/local/bin/api_validator.py # 每小时检查日志文件 0 * * * * /usr/local/bin/log_validator.sh
2 Systemd服务配置
[Unit] Description=Data Validation Monitor Service After=network.target [Service] Type=simple User=monitor ExecStart=/usr/local/bin/data_validation_monitor.py Restart=on-failure RestartSec=10 StandardOutput=journal StandardError=journal [Install] WantedBy=multi-user.target
最佳实践建议
1 告警级别定义
ALERT_LEVELS = {
'CRITICAL': {
'threshold': 10,
'notify_methods': ['email', 'sms', 'phone'],
'retry_interval': 60 # seconds
},
'WARNING': {
'threshold': 5,
'notify_methods': ['email'],
'retry_interval': 300
},
'INFO': {
'threshold': 1,
'notify_methods': ['log'],
'retry_interval': 3600
}
}
2 故障自动恢复
def auto_retry_with_backoff(func, max_retries=3):
"""带指数退避的重试机制"""
for attempt in range(max_retries):
try:
return func()
except Exception as e:
wait_time = 2 ** attempt
logger.warning(f"第 {attempt + 1} 次重试，等待 {wait_time} 秒")
time.sleep(wait_time)
raise Exception("超过最大重试次数")
这些脚本涵盖了：
- ✅ 文件数据校验
- ✅ 数据库完整性检查
- ✅ API响应验证
- ✅ 数据一致性检查
- ✅ 告警通知机制
- ✅ 自动化部署方案
选择哪种方案取决于您的具体需求：简单文件校验用Shell脚本，复杂业务逻辑用Python，需要实时性则使用API监控方案。
