监控数据校验失败的脚本如何编写

本文目录导读：

1. 基础数据校验监控脚本
2. 高级监控方案
3. 自动化部署与执行
4. 最佳实践建议

我来为您提供监控数据校验失败脚本的编写方案,涵盖基础到进阶的多种场景。

基础数据校验监控脚本

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监控方案。