本文目录导读：
我来为您介绍监控数据质量规则执行的脚本编写方法,包含多种实现方式。
Python 脚本示例
基础监控脚本
import pandas as pd
import numpy as np
import json
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
import logging
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('data_quality_monitor.log'),
logging.StreamHandler()
]
)
class DataQualityMonitor:
def __init__(self, config_file='dq_rules.json'):
self.rules = self.load_rules(config_file)
self.results = []
def load_rules(self, config_file):
"""加载数据质量规则配置"""
with open(config_file, 'r') as f:
return json.load(f)
def check_completeness(self, df, column, threshold=0.9):
"""检查数据完整性"""
total = len(df)
missing = df[column].isnull().sum()
completeness = 1 - (missing / total)
result = {
'rule': f'completeness_{column}',
'column': column,
'metric': 'completeness',
'value': completeness,
'threshold': threshold,
'status': 'PASS' if completeness >= threshold else 'FAIL',
'timestamp': datetime.now().isoformat(),
'details': {
'total_records': total,
'missing_records': missing,
'missing_rate': missing / total
}
}
if result['status'] == 'FAIL':
logging.error(f"数据完整性检查失败: {column} - {completeness}")
return result
def check_uniqueness(self, df, columns, threshold=1.0):
"""检查数据唯一性"""
if isinstance(columns, str):
columns = [columns]
dups = df.duplicated(subset=columns, keep=False)
unique_rate = 1 - (dups.sum() / len(df))
result = {
'rule': f'uniqueness_{"_".join(columns)}',
'columns': columns,
'metric': 'uniqueness',
'value': unique_rate,
'threshold': threshold,
'status': 'PASS' if unique_rate >= threshold else 'FAIL',
'timestamp': datetime.now().isoformat(),
'details': {
'total_records': len(df),
'duplicate_records': dups.sum(),
'duplicate_rate': dups.sum() / len(df)
}
}
return result
def check_value_range(self, df, column, min_val=None, max_val=None):
"""检查数据值范围"""
col_data = df[column]
within_range = pd.Series([True] * len(df))
if min_val is not None:
within_range = within_range & (col_data >= min_val)
if max_val is not None:
within_range = within_range & (col_data <= max_val)
pass_rate = within_range.mean()
result = {
'rule': f'range_{column}',
'column': column,
'metric': 'value_range',
'value': pass_rate,
'threshold': 1.0,
'status': 'PASS' if pass_rate == 1.0 else 'FAIL',
'timestamp': datetime.now().isoformat(),
'details': {
'total_records': len(df),
'violations': (~within_range).sum(),
'violation_rate': 1 - pass_rate,
'min_value': col_data.min() if len(col_data) > 0 else None,
'max_value': col_data.max() if len(col_data) > 0 else None
}
}
return result
def check_data_format(self, df, column, pattern):
"""检查数据格式（正则表达式）"""
import re
matches = df[column].astype(str).str.match(pattern)
pass_rate = matches.mean()
result = {
'rule': f'format_{column}',
'column': column,
'metric': 'data_format',
'value': pass_rate,
'threshold': 1.0,
'status': 'PASS' if pass_rate == 1.0 else 'FAIL',
'timestamp': datetime.now().isoformat(),
'details': {
'total_records': len(df),
'format_mismatch': (~matches).sum(),
'pattern': pattern
}
}
return result
def check_custom_rule(self, df, rule_config):
"""执行自定义规则检查"""
column = rule_config.get('column')
rule_type = rule_config.get('type')
params = rule_config.get('params', {})
if rule_type == 'completeness':
return self.check_completeness(df, column, params.get('threshold', 0.9))
elif rule_type == 'uniqueness':
return self.check_uniqueness(df, column, params.get('threshold', 1.0))
elif rule_type == 'range':
return self.check_value_range(df, column,
params.get('min'),
params.get('max'))
elif rule_type == 'format':
return self.check_data_format(df, column, params.get('pattern'))
else:
logging.warning(f"未识别的规则类型: {rule_type}")
return None
def run_all_checks(self, df):
"""运行所有数据质量检查"""
self.results = []
for rule in self.rules['rules']:
try:
result = self.check_custom_rule(df, rule)
if result:
self.results.append(result)
logging.info(f"规则检查完成: {result['rule']} - {result['status']}")
except Exception as e:
logging.error(f"规则检查失败: {rule.get('column')} - {str(e)}")
return self.results
def generate_report(self):
"""生成数据质量报告"""
report = {
'report_time': datetime.now().isoformat(),
'total_rules': len(self.results),
'passed_rules': sum(1 for r in self.results if r['status'] == 'PASS'),
'failed_rules': sum(1 for r in self.results if r['status'] == 'FAIL'),
'pass_rate': sum(1 for r in self.results if r['status'] == 'PASS') / len(self.results) if self.results else 0,
'results': self.results
}
return report
def send_alert(self, report, recipients):
"""发送告警通知"""
failed_rules = [r for r in self.results if r['status'] == 'FAIL']
if not failed_rules:
return
subject = f"[数据质量告警] {len(failed_rules)} 条规则检查失败"
body = self.format_alert_message(report)
# 发送邮件通知
self.send_email(subject, body, recipients)
def format_alert_message(self, report):
"""格式化告警消息"""
messages = []
messages.append(f"数据质量检查报告")
messages.append(f"检查时间: {report['report_time']}")
messages.append(f"通过率: {report['pass_rate']:.2%}")
messages.append("")
messages.append("失败规则详情:")
for result in self.results:
if result['status'] == 'FAIL':
messages.append(f" - 规则: {result['rule']}")
messages.append(f" 指标值: {result['value']:.4f}")
messages.append(f" 阈值: {result['threshold']}")
messages.append(f" 详情: {result['details']}")
messages.append("")
return "\n".join(messages)
def send_email(self, subject, body, recipients):
"""发送邮件"""
# 配置邮件服务器
smtp_server = "smtp.gmail.com"
smtp_port = 587
sender_email = "your_email@gmail.com"
sender_password = "your_password"
msg = MIMEText(body)
msg['Subject'] = subject
msg['From'] = sender_email
msg['To'] = ", ".join(recipients)
try:
with smtplib.SMTP(smtp_server, smtp_port) as server:
server.starttls()
server.login(sender_email, sender_password)
server.send_message(msg)
logging.info(f"告警邮件已发送至: {recipients}")
except Exception as e:
logging.error(f"发送邮件失败: {str(e)}")
# 规则配置文件示例 (dq_rules.json)
"""
{
"rules": [
{
"column": "user_id",
"type": "completeness",
"params": {
"threshold": 0.95
}
},
{
"column": "email",
"type": "format",
"params": {
"pattern": "^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$"
}
},
{
"column": "age",
"type": "range",
"params": {
"min": 0,
"max": 150
}
},
{
"column": ["user_id", "email"],
"type": "uniqueness",
"params": {
"threshold": 1.0
}
}
],
"schedules": {
"frequency": "daily",
"time": "02:00",
"monitor_tables": ["users", "orders", "products"]
},
"alerts": {
"enabled": true,
"recipients": ["data_team@company.com", "alert@company.com"],
"severity_threshold": "WARNING"
}
}
"""
# 主程序
def main():
# 初始化监控器
monitor = DataQualityMonitor('dq_rules.json')
# 读取数据 (示例)
df = pd.read_csv('data.csv')
# 运行所有检查
results = monitor.run_all_checks(df)
# 生成报告
report = monitor.generate_report()
# 保存报告
with open(f'dq_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json', 'w') as f:
json.dump(report, f, indent=2, default=str)
# 发送告警
if report['failed_rules'] > 0:
monitor.send_alert(report, ['data_team@company.com'])
# 输出摘要
print(f"数据质量检查完成")
print(f"通过: {report['passed_rules']}/{report['total_rules']}")
print(f"失败: {report['failed_rules']}/{report['total_rules']}")
print(f"通过率: {report['pass_rate']:.2%}")
if __name__ == "__main__":
main()
Shell 脚本示例
#!/bin/bash
# 数据质量监控脚本
# 用途：监控数据表的质量指标
# 配置
DATABASE="your_database"
USER="db_user"
PASSWORD="db_password"
HOST="localhost"
ALERT_EMAIL="data_team@company.com"
LOG_FILE="/var/log/data_quality_monitor.log"
# 日志函数
log() {
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a "$LOG_FILE"
}
# 发送告警函数
send_alert() {
local subject="$1"
local message="$2"
echo "$message" | mail -s "$subject" "$ALERT_EMAIL"
}
# 检查表完整性
check_table_completeness() {
local table=$1
log "检查表 $table 的完整性..."
# 检查表是否存在
mysql -u$USER -p$PASSWORD -h$HOST -e "
SELECT COUNT(*) as record_count
FROM information_schema.tables
WHERE table_schema='$DATABASE' AND table_name='$table'
" 2>/dev/null | grep -v "COUNT"
}
# 检查记录数变化
check_record_count() {
local table=$1
local threshold=$2
log "检查表 $table 的记录数..."
# 获取当前记录数
current_count=$(mysql -u$USER -p$PASSWORD -h$HOST -D$DATABASE -e "
SELECT COUNT(*) FROM $table
" 2>/dev/null | grep -v "COUNT")
# 获取上次记录数
if [ -f "/tmp/${table}_record_count" ]; then
last_count=$(cat "/tmp/${table}_record_count")
else
last_count=$current_count
fi
# 计算变化率
if [ $last_count -gt 0 ]; then
change_rate=$(echo "scale=4; ($current_count - $last_count) / $last_count * 100" | bc)
else
change_rate=0
fi
# 记录当前值
echo $current_count > "/tmp/${table}_record_count"
log "表 $table: 当前记录数=$current_count, 上次记录数=$last_count, 变化率=${change_rate}%"
# 检查是否超出阈值
if (( $(echo "$change_rate > $threshold" | bc -l) )); then
send_alert "数据量异常告警" "表 $table 记录数变化异常: ${change_rate}% (阈值: $threshold%)"
return 1
fi
return 0
}
# 检查空值比例
check_null_ratio() {
local table=$1
local column=$2
local threshold=$3
log "检查表 $table 的 $column 列空值比例..."
# 获取空值比例
null_ratio=$(mysql -u$USER -p$PASSWORD -h$HOST -D$DATABASE -N -e "
SELECT round(SUM(CASE WHEN $column IS NULL THEN 1 ELSE 0 END) / COUNT(*) * 100, 2)
FROM $table
" 2>/dev/null)
log "列 $column 空值比例: ${null_ratio}% (阈值: $threshold%)"
if (( $(echo "$null_ratio > $threshold" | bc -l) )); then
send_alert "空值异常告警" "表 $table 的 $column 列空值比例异常: ${null_ratio}% (阈值: $threshold%)"
return 1
fi
return 0
}
# 检查重复数据
check_duplicates() {
local table=$1
local columns=$2
local threshold=$3
log "检查表 $table 的重复数据..."
# 获取重复记录数
duplicates=$(mysql -u$USER -p$PASSWORD -h$HOST -D$DATABASE -N -e "
SELECT COUNT(*)
FROM (
SELECT $columns, COUNT(*) as cnt
FROM $table
GROUP BY $columns
HAVING COUNT(*) > 1
) as dups
" 2>/dev/null)
log "表 $table 重复记录数: $duplicates (阈值: $threshold)"
if [ $duplicates -gt $threshold ]; then
send_alert "数据重复告警" "表 $table 发现 $duplicates 条重复记录 (阈值: $threshold)"
return 1
fi
return 0
}
# 检查数据新鲜度
check_freshness() {
local table=$1
local date_column=$2
local max_age_hours=$3
log "检查表 $table 的数据新鲜度..."
# 获取最新数据时间
latest_date=$(mysql -u$USER -p$PASSWORD -h$HOST -D$DATABASE -N -e "
SELECT MAX($date_column) FROM $table
" 2>/dev/null)
if [ -z "$latest_date" ]; then
log "警告: 表 $table 没有数据"
return 1
fi
# 计算时间差（小时）
current_time=$(date +%s)
latest_time=$(date -d "$latest_date" +%s 2>/dev/null || echo 0)
if [ $latest_time -eq 0 ]; then
log "警告: 无法解析时间 $latest_date"
return 1
fi
hours_diff=$(( ($current_time - $latest_time) / 3600 ))
log "表 $table 最新数据时间: $latest_date, 距今 ${hours_diff}小时 (阈值: $max_age_hours小时)"
if [ $hours_diff -gt $max_age_hours ]; then
send_alert "数据延迟告警" "表 $table 数据延迟 ${hours_diff}小时 (阈值: $max_age_hours小时)"
return 1
fi
return 0
}
# 主监控函数
monitor_all_tables() {
log "开始数据质量监控..."
# 定义监控表列表
tables=("users" "orders" "products" "transactions")
for table in "${tables[@]}"; do
log "监控表: $table"
# 检查记录数变化
check_record_count "$table" 50 # 50% 变化阈值
# 检查空值比例
case $table in
"users")
check_null_ratio "$table" "email" 5
check_null_ratio "$table" "phone" 10
;;
"orders")
check_null_ratio "$table" "order_amount" 1
check_null_ratio "$table" "customer_id" 0.5
;;
esac
# 检查重复数据
case $table in
"users")
check_duplicates "$table" "email" 0
;;
"orders")
check_duplicates "$table" "order_id" 0
;;
esac
# 检查数据新鲜度
check_freshness "$table" "created_at" 24 # 24小时内必须有新数据
done
log "数据质量监控完成"
}
# 运行监控
monitor_all_tables
SQL 监控脚本
-- 数据质量监控SQL脚本
-- 用途：定期执行数据质量检查
-- 1. 数据完整性检查
SELECT
'completeness_check' as check_name,
'users' as table_name,
'email' as column_name,
COUNT(*) as total_records,
SUM(CASE WHEN email IS NULL THEN 1 ELSE 0 END) as null_count,
ROUND(SUM(CASE WHEN email IS NULL THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) as null_percentage,
CASE
WHEN SUM(CASE WHEN email IS NULL THEN 1 ELSE 0 END) * 100.0 / COUNT(*) > 5
THEN 'FAIL'
ELSE 'PASS'
END as status
FROM users;
-- 2. 数据唯一性检查
SELECT
'uniqueness_check' as check_name,
'users' as table_name,
'email' as column_name,
COUNT(*) as total_count,
COUNT(DISTINCT email) as unique_count,
COUNT(*) - COUNT(DISTINCT email) as duplicate_count,
CASE
WHEN COUNT(*) = COUNT(DISTINCT email) THEN 'PASS'
ELSE 'FAIL'
END as status
FROM users;
-- 3. 数据范围检查
SELECT
'range_check' as check_name,
'users' as table_name,
'age' as column_name,
MIN(age) as min_value,
MAX(age) as max_value,
AVG(age) as avg_value,
SUM(CASE WHEN age < 0 OR age > 150 THEN 1 ELSE 0 END) as out_of_range_count,
CASE
WHEN SUM(CASE WHEN age < 0 OR age > 150 THEN 1 ELSE 0 END) > 0
THEN 'FAIL'
ELSE 'PASS'
END as status
FROM users;
-- 4. 数据格式检查
SELECT
'format_check' as check_name,
'users' as table_name,
'phone' as column_name,
COUNT(*) as total_records,
SUM(CASE
WHEN phone REGEXP '^[0-9]{3}-[0-9]{3}-[0-9]{4}$'
THEN 1 ELSE 0
END) as valid_format_count,
SUM(CASE
WHEN phone REGEXP '^[0-9]{3}-[0-9]{3}-[0-9]{4}$'
THEN 0 ELSE 1
END) as invalid_format_count,
CASE
WHEN SUM(CASE
WHEN phone REGEXP '^[0-9]{3}-[0-9]{3}-[0-9]{4}$'
THEN 0 ELSE 1
END) > 0 THEN 'FAIL'
ELSE 'PASS'
END as status
FROM users;
-- 5. 数据一致性检查
SELECT
'consistency_check' as check_name,
'orders' as table_name,
'customer_id' as column_name,
COUNT(*) as total_orders,
COUNT(DISTINCT o.customer_id) as unique_customers,
SUM(CASE WHEN u.id IS NULL THEN 1 ELSE 0 END) as orphan_records,
CASE
WHEN SUM(CASE WHEN u.id IS NULL THEN 1 ELSE 0 END) > 0
THEN 'FAIL'
ELSE 'PASS'
END as status
FROM orders o
LEFT JOIN users u ON o.customer_id = u.id;
-- 6. 数据时效性检查
SELECT
'freshness_check' as check_name,
'orders' as table_name,
'created_at' as date_column,
MAX(created_at) as latest_record,
TIMESTAMPDIFF(HOUR, MAX(created_at), NOW()) as hours_since_last_update,
CASE
WHEN TIMESTAMPDIFF(HOUR, MAX(created_at), NOW()) > 24
THEN 'FAIL'
ELSE 'PASS'
END as status
FROM orders;
配置文件示例
{
"monitor_config": {
"name": "数据质量监控系统",
"version": "1.0.0",
"enabled": true,
"check_interval_minutes": 60
},
"database": {
"type": "mysql",
"host": "localhost",
"port": 3306,
"database": "production_db",
"user": "monitor_user",
"password": "encrypted_password"
},
"tables": [
{
"name": "users",
"checks": [
{
"type": "completeness",
"columns": ["name", "email", "phone"],
"threshold": 95,
"severity": "critical"
},
{
"type": "uniqueness",
"columns": ["email"],
"threshold": 100,
"severity": "critical"
},
{
"type": "freshness",
"date_column": "created_at",
"max_hours": 24,
"severity": "warning"
}
]
},
{
"name": "orders",
"checks": [
{
"type": "completeness",
"columns": ["order_id", "customer_id", "amount"],
"threshold": 100,
"severity": "critical"
},
{
"type": "consistency",
"reference_table": "users",
"join_column": "customer_id",
"severity": "critical"
},
{
"type": "value_range",
"column": "amount",
"min": 0,
"max": 100000,
"severity": "warning"
}
]
}
],
"alerts": {
"email": {
"enabled": true,
"smtp_server": "smtp.company.com",
"smtp_port": 587,
"sender": "monitor@company.com",
"recipients": ["data-team@company.com", "dba@company.com"],
"subject_prefix": "[数据质量]"
},
"slack": {
"enabled": true,
"webhook_url": "https://hooks.slack.com/services/xxx/yyy/zzz",
"channel": "#data-quality",
"username": "Data Monitor"
},
"pagerduty": {
"enabled": false,
"integration_key": "your_key"
}
},
"reporting": {
"save_results": true,
"results_directory": "/var/log/data_quality/",
"retention_days": 90,
"summary_enabled": true,
"detailed_enabled": true
}
}
监控脚本使用说明
安装依赖
# Python 依赖 pip install pandas numpy # 监控脚本 chmod +x data_quality_monitor.sh # 设置定时任务 crontab -e # 添加以下行（每小时执行一次） 0 * * * * /path/to/data_quality_monitor.sh
使用方法
# Python 使用示例
from data_quality_monitor import DataQualityMonitor
# 初始化监控器
monitor = DataQualityMonitor('dq_rules.json')
# 读取数据
df = pd.read_csv('your_data.csv')
# 执行检查
results = monitor.run_all_checks(df)
# 生成报告
report = monitor.generate_report()
# 保存报告
with open('report.json', 'w') as f:
json.dump(report, f, indent=2)
这些脚本提供了完整的数据质量监控解决方案,包括完整性、唯一性、范围、格式和一致性检查，支持告警通知和报告生成，您可以根据实际需求进行定制和扩展。
