数据质量监控与告警实战

# 六维度指标计算示例completeness = non_null_count / total_count  # 完整度accuracy = in_range_count / non_null_count   # 准确度consistency = matched_count / total_count     # 一致性timeliness = on_time_count / total_count      # 时效性uniqueness = distinct_count / non_null_count  # 唯一性validity = valid_format_count / total_count   # 有效性

# Dashboard数据结构dashboard_data = {    "overview_cards": [        {"dimension": "completeness", "value": 0.997,         "slo": 0.999, "status": "yellow", "trend": "down"},    ],    "alert_list": [        {"severity": "critical", "table": "users",         "field": "email", "issue": "null_rate=8.2%"}    ],    "heatmap": {"users": {"completeness": 0.95, "uniqueness": 1.0}},}

# 动态阈值计算import statisticshistory = [0.98, 0.97, 0.99, 0.96, 0.98, 0.97, 0.99]mean = statistics.mean(history)       # 均值std = statistics.stdev(history)        # 标准差lower = mean - 2 * std                 # 下界 (mean - 2σ)upper = mean + 2 * std                 # 上界 (mean + 2σ)print(f"动态阈值: [{lower:.4f}, {upper:.4f}]")

# 质量门禁决策逻辑def quality_gate(check_results):    for r in check_results:        if r["severity"] == "critical":            return {"pass": False, "action": "hard_block",                    "reason": r["issue"]}        elif r["severity"] == "warning":            return {"pass": True, "action": "soft_alert",                    "quality_tag": "degraded"}    return {"pass": True, "action": "normal"}

# Airflow DAG 伪代码：质量检查集成with DAG("quality_check", schedule="@hourly") as dag:    extract = PythonOperator(task_id="extract_data")    schema_check = PythonOperator(task_id="schema_validation")    row_check = PythonOperator(task_id="row_level_checks")    stat_check = PythonOperator(task_id="statistical_checks")    gate = PythonOperator(task_id="quality_gate_decision")    notify = PythonOperator(task_id="send_alerts")     extract >> schema_check >> row_check >> stat_check >> gate >> notify

import mathfrom typing import List, Dict, Tuple, Optionalfrom collections import defaultdict  class DataQualityMonitor:    """数据质量监控引擎：计算六维度指标、检查阈值、生成告警"""     DIMENSIONS = ["completeness", "accuracy", "consistency",                  "timeliness", "uniqueness", "validity"]     def __init__(self, table_name: str):        self.table_name = table_name        self.baselines: Dict[str, Dict] = {}        self.alerts: List[Dict] = []        self.metrics_history: List[Dict] = []     def calc_completeness(self, column: str, values: List) -> float:        """完整度：非空值占比"""        if not values:            return 0.0        non_null = sum(1 for v in values if v is not None)        return non_null / len(values)     def calc_null_rate(self, column: str, values: List) -> float:        """空值率：空值占比"""        return 1.0 - self.calc_completeness(column, values)     def calc_accuracy(self, values: List, expected_range: Tuple) -> float:        """准确度：值在预期范围内的比例"""        low, high = expected_range        in_range = sum(1 for v in values if v is not None and low <= v <= high)        total = sum(1 for v in values if v is not None)        return in_range / total if total > 0 else 0.0     def calc_uniqueness(self, values: List) -> float:        """唯一性：不重复值占比"""        filtered = [v for v in values if v is not None]        if not filtered:            return 0.0        return len(set(filtered)) / len(filtered)     def calc_validity(self, values: List, pattern: str) -> float:        """有效性：符合格式模式的比例（简化为长度检查）"""        filtered = [v for v in values if v is not None]        if not filtered:            return 0.0        valid_count = sum(1 for v in filtered if len(str(v)) > 0)        return valid_count / len(filtered)     def calc_timeliness(self, timestamps: List, max_delay_sec: float) -> float:        """时效性：在最大延迟内到达的比例"""        if not timestamps:            return 0.0        import time        now = time.time()        timely = sum(1 for t in timestamps if (now - t) <= max_delay_sec)        return timely / len(timestamps)     def set_static_threshold(self, dimension: str, min_val: float, max_val: float):        """设定静态阈值"""        self.baselines[dimension] = {            "type": "static", "min": min_val, "max": max_val        }     def set_dynamic_threshold(self, dimension: str, values_history: List[float],                              sigma: float = 2.0):        """基于历史数据设定动态阈值: mean +/- sigma*std"""        if not values_history:            return        mean = sum(values_history) / len(values_history)        variance = sum((x - mean) ** 2 for x in values_history) / len(values_history)        std = math.sqrt(variance)        self.baselines[dimension] = {            "type": "dynamic",            "mean": mean, "std": std, "sigma": sigma,            "lower": mean - sigma * std,            "upper": mean + sigma * std        }     def check_threshold(self, dimension: str, value: float) -> Optional[Dict]:        """检查指标值是否超出阈值，返回告警或None"""        bl = self.baselines.get(dimension)        if not bl:            return None        if bl["type"] == "static":            if value < bl["min"] or value > bl["max"]:                return {"dimension": dimension, "value": value,                        "threshold": bl, "severity": "critical"                        if value < bl["min"] * 0.9 else "warning"}        elif bl["type"] == "dynamic":            if value < bl["lower"] or value > bl["upper"]:                deviation = abs(value - bl["mean"]) / bl["std"] if bl["std"] > 0 else 0                severity = "critical" if deviation > bl["sigma"] * 1.5 else "warning"                return {"dimension": dimension, "value": value,                        "threshold": bl, "severity": severity}        return None     def run_quality_check(self, data: Dict[str, List],                          constraints: Dict) -> Dict:        """执行完整的质量检查流水线"""        results = {}        for dim in self.DIMENSIONS:            if dim == "completeness" and "null_check_cols" in constraints:                for col in constraints["null_check_cols"]:                    score = self.calc_completeness(col, data.get(col, []))                    alert = self.check_threshold(dim, score)                    if alert:                        self.alerts.append(alert)                    results[f"{dim}_{col}"] = score            elif dim == "uniqueness" and "unique_cols" in constraints:                for col in constraints["unique_cols"]:                    score = self.calc_uniqueness(data.get(col, []))                    alert = self.check_threshold(dim, score)                    if alert:                        self.alerts.append(alert)                    results[f"{dim}_{col}"] = score            elif dim == "accuracy" and "range_checks" in constraints:                for col, rng in constraints["range_checks"].items():                    score = self.calc_accuracy(data.get(col, []), rng)                    alert = self.check_threshold(dim, score)                    if alert:                        self.alerts.append(alert)                    results[f"{dim}_{col}"] = score        self.metrics_history.append(results)        has_critical = any(a["severity"] == "critical" for a in self.alerts)        return {            "table": self.table_name, "metrics": results,            "alerts": self.alerts,            "gate_pass": not has_critical        }     def get_dashboard_data(self) -> Dict:        """生成Dashboard可视化数据"""        latest = self.metrics_history[-1] if self.metrics_history else {}        return {            "overview_cards": [                {"dim": dim, "value": latest.get(dim, "N/A"),                 "status": "green" if isinstance(latest.get(dim), float)                 and latest[dim] >= 0.95 else "red"}                for dim in self.DIMENSIONS            ],            "alert_summary": {                "total": len(self.alerts),                "critical": sum(1 for a in self.alerts                                if a["severity"] == "critical"),                "warning": sum(1 for a in self.alerts                               if a["severity"] == "warning")            },            "trend": self.metrics_history[-30:]        }  def demo_monitoring():    """演示数据质量监控流程"""    monitor = DataQualityMonitor("user_features")    monitor.set_static_threshold("completeness", 0.95, 1.0)    monitor.set_dynamic_threshold("accuracy",                                  [0.92, 0.94, 0.91, 0.93, 0.95], sigma=2.0)     data = {        "user_id": [1, 2, 3, None, 5, 3, 7, 8, 9, 10],        "age": [25, 30, -5, 40, 150, 28, 35, 22, 45, 33],        "email": ["a@b.com", "c@d.com", "", "e@f.com", "g@h.com",                  "i@j.com", "k@l.com", "m@n.com", "o@p.com", "q@r.com"]    }    constraints = {        "null_check_cols": ["user_id", "email"],        "unique_cols": ["user_id"],        "range_checks": {"age": (0, 120)}    }    result = monitor.run_quality_check(data, constraints)    print(f"质量门禁通过: {result['gate_pass']}")    print(f"告警数量: {len(result['alerts'])}")    for a in result["alerts"]:        print(f"  [{a['severity']}] {a['dimension']}: {a['value']:.2f}")     dashboard = monitor.get_dashboard_data()    print(f"\nDashboard概览:")    for card in dashboard["overview_cards"]:        print(f"  {card['dim']}: {card['value']} ({card['status']})")  if __name__ == "__main__":    demo_monitoring()