Python 数据质量监控：Schema 校验比代码 Review 更可靠

Python 数据质量监控：Schema 校验比代码 Review 更可靠

一、下游报表数据对不上，排查发现上游两个字段的类型悄悄变了

数据团队的日常：上游数据提供方改了字段类型。
user_id从int变成了string。
下游 ETL 任务默默失败了三天，没人发现。
因为代码的 try-except 把异常吞了。

Code Review 能发现代码层面的问题。
但发现不了数据层面的漂移。
Schema 校验在数据进入管线时截住异常。
"这行数据的 user_id 是字符串，不是整数——拒绝处理并告警"。

数据质量的崩塌从来不是一夜之间发生的。
它是以"悄无声息"的方式蔓延的——上游改了一个字段类型，下游默默失败，失败被 try-except 吃掉，监控没有告警，直到两周后运营同学跑月度报表时发现数据对不上。
这时候排查的难度已经是初始状态的 10 倍，因为你需要找到两周前的变更记录、重建当时的数据快照、分析每一条失败链路。

我在数据团队的经历中，这类事故占了数据质量问题的 70% 以上。
根本原因不是代码写错了，而是数据的"契约"没有被强制。
上游和下游之间缺少一个硬性的接口约定——上游承诺字段类型和值域，下游在收到不符合约定的数据时立即告警。

Schema 校验就是这道"数据契约"的执行者。
它不依赖人工 Review，不依赖代码规范，不依赖大家的自觉性。
它在数据进入管线的第一时间做检查，不合格的数据要么拒绝、要么标记、要么告警——总之不可能悄无声息地漏过去。

二、数据质量监控的三个层面

flowchart TB A[数据流入] --> B[层面1: Schema 格式校验] B --> C{类型/必填/枚举} C -->|失败| D[拒绝 + 告警] C -->|通过| E[层面2: 统计特征监控] E --> F{均值/标准差/空值率} F -->|异常漂移| G[告警 + 人工确认] F -->|正常| H[层面3: 业务规则校验] H --> I{金额>0/日期合理/关联存在} I -->|失败| J[死信 + 标记] I -->|通过| K[正常写入]

三层递进设计遵循"越底层越严格、越上层越灵活"的原则。
Schema 校验是硬约束——类型错了就是错了，没有商量余地。
统计监控是软约束——指标漂移不一定是问题，需要人工判断。
业务规则是领域约束——金额为负在某些场景下可能合理（退款），需要结合上下文判断。

这样分层的好处是：告警的优先级和响应策略可以区分对待。
Schema 失败的告警应该发到 PagerDuty，立即处理。
统计漂移的告警可以发到 Slack 频道，工作时间查看。
业务规则违反的告警根据规则严重程度分级——资金类规则走 P1 通道，格式类规则走常规通道。

三、Schema 校验框架实现

下面的代码实现了三层数据质量监控：Schema 格式校验、统计特征监控、数据漂移检测。
DataQualityMonitor使用组合模式将三层串联，一条数据经过全链路检查后才放行。

""" data_quality.py - 数据质量监控 """ import json import logging from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Callable from datetime import datetime, timedelta from collections import defaultdict import statistics logger = logging.getLogger(__name__) @dataclass class ValidationError: """校验错误""" field: str value: Any rule: str message: str class SchemaValidator: """Schema 校验器""" TYPE_MAP = { "string": str, "integer": int, "float": float, "boolean": bool, "list": list, "dict": dict, "null": type(None), } def __init__(self, schema: Dict[str, Dict]): """ schema = { "user_id": {"type": "integer", "required": True}, "name": {"type": "string", "required": True, "max_length": 50}, "age": {"type": "integer", "min": 0, "max": 150}, "email": { "type": "string", "pattern": r'^[^@]+@[^@]+\.[^@]+$' }, "status": {"type": "string", "enum": ["active", "inactive"]}, } """ self.schema = schema def validate(self, record: Dict) -> List[ValidationError]: """校验单条记录""" errors = [] for field, rules in self.schema.items(): value = record.get(field) # 必填检查 if rules.get("required", False) and value is None: errors.append(ValidationError( field, None, "required", f"缺少必填字段 {field}" )) continue if value is None: continue # 类型检查 expected_type = rules.get("type") if expected_type: base_type = self.TYPE_MAP.get(expected_type) if base_type and not isinstance(value, base_type): errors.append(ValidationError( field, value, "type", f"类型错误: 期望 {expected_type}, " f"实际 {type(value).__name__}" )) continue # 范围检查 if "min" in rules and isinstance(value, (int, float)): if value < rules["min"]: errors.append(ValidationError( field, value, "min", f"值 {value} 小于最小值 {rules['min']}" )) if "max" in rules and isinstance(value, (int, float)): if value > rules["max"]: errors.append(ValidationError( field, value, "max", f"值 {value} 大于最大值 {rules['max']}" )) # 长度检查 if "max_length" in rules and isinstance(value, str): if len(value) > rules["max_length"]: errors.append(ValidationError( field, value[:50], "max_length", f"长度 {len(value)} 超过限制 {rules['max_length']}" )) # 枚举检查 if "enum" in rules and value not in rules["enum"]: errors.append(ValidationError( field, value, "enum", f"值 '{value}' 不在允许范围 {rules['enum']}" )) # 正则匹配 if "pattern" in rules and isinstance(value, str): import re if not re.match(rules["pattern"], value): errors.append(ValidationError( field, value, "pattern", f"格式不匹配正则: {rules['pattern']}" )) return errors class StatisticsMonitor: """统计特征监控器""" def __init__(self, window_size: int = 10000): self.window_size = window_size self.history: Dict[str, List] = defaultdict(list) self.baseline: Dict[str, Dict] = {} self.alerts: List[str] = [] def feed(self, record: Dict): """喂入数据，更新统计""" for field, value in record.items(): if isinstance(value, (int, float)): self.history[field].append(value) if len(self.history[field]) > self.window_size: self.history[field] = self.history[field][-self.window_size:] def compute_baseline(self): """计算基线统计值""" for field, values in self.history.items(): if len(values) < 100: continue self.baseline[field] = { "mean": statistics.mean(values), "stdev": statistics.stdev(values) if len(values) > 1 else 0, "min": min(values), "max": max(values), "null_rate": values.count(0) / len(values), } def check_drift(self, record: Dict, threshold: float = 3.0) -> List[str]: """ 检测数据漂移（3-sigma 规则） threshold: 偏离基线多少个标准差触发告警 """ alerts = [] for field, value in record.items(): if field not in self.baseline: continue if not isinstance(value, (int, float)): continue baseline = self.baseline[field] if baseline["stdev"] == 0: continue z_score = abs(value - baseline["mean"]) / baseline["stdev"] if z_score > threshold: alerts.append( f"字段 {field} 值 {value} 偏离基线 " f"(均值={baseline['mean']:.2f}, " f"z-score={z_score:.2f})" ) return alerts class DataQualityMonitor: """数据质量监控器（组合模式）""" def __init__(self, schema: Dict[str, Dict]): self.schema_validator = SchemaValidator(schema) self.stats_monitor = StatisticsMonitor() self.error_counts: Dict[str, int] = defaultdict(int) self.total_records = 0 def check(self, record: Dict) -> Dict[str, Any]: """综合数据质量检查""" self.total_records += 1 result = { "valid": True, "errors": [], "alerts": [], } # 1. Schema 校验 errors = self.schema_validator.validate(record) if errors: result["valid"] = False result["errors"] = errors for e in errors: self.error_counts[e.rule] += 1 # 2. 统计漂移检测（仅在数据量充足时） self.stats_monitor.feed(record) if self.total_records % 1000 == 0: self.stats_monitor.compute_baseline() if len(self.stats_monitor.history.get("user_id", [])) > 1000: alerts = self.stats_monitor.check_drift(record) if alerts: result["alerts"] = alerts return result def summary(self) -> str: """数据质量摘要""" error_rate = ( sum(self.error_counts.values()) / max(self.total_records, 1) ) return ( f"数据质量摘要:\n" f" 总记录: {self.total_records}\n" f" 错误率: {error_rate:.2%}\n" f" 错误分布: {dict(self.error_counts)}" ) # ---- 使用示例 ---- schema = { "user_id": {"type": "integer", "required": True}, "amount": {"type": "float", "min": 0}, } monitor = DataQualityMonitor(schema) # 正常数据 result = monitor.check({"user_id": 123, "amount": 99.9}) assert result["valid"] # 异常数据：类型错误 result = monitor.check({"user_id": "abc", "amount": 99.9}) print(f"错误: {[e.message for e in result['errors']]}") # 异常数据：金额为负 result = monitor.check({"user_id": 456, "amount": -50}) print(f"错误: {[e.message for e in result['errors']]}")

StatisticsMonitor的漂移检测基于 3-sigma 规则：当某个字段的值偏离基线均值超过 3 个标准差时触发告警。
这是一种在工业界广泛使用的异常检测方法，简单但有效。
但要注意：3-sigma 假设数据服从正态分布，对于非正态分布的数据（如订单金额通常呈长尾分布），可以考虑用 MAD（Median Absolute Deviation）替代。

四、校验成本与覆盖度

Schema 校验增加每条约 0.1-0.5ms 开销。
占 ETL 总耗时的 1-2%。
这个成本可以接受。

统计监控需要积累足够的基线数据（> 1000 条）。
数据量太少时，标准差不可靠。
建议在数据稳定后（> 1 万条）启用漂移检测。

不适合的场景：
Schema 频繁变化的探索性分析；
非结构化数据（如自由文本）；
数据量 < 100 的微型任务。

除了计算开销，Schema 定义本身的维护也是成本。
字段新增、类型变更需要同步更新 Schema 文件。
建议把 Schema 定义和数据源服务放在同一个仓库中，由数据源团队负责维护 Schema 的正确性。
不要由下游数据团队独自维护 Schema——那样上游改字段你不知道，Schema 就会过期。

还有一个实践建议：Schema 校验不应阻塞数据流，而是采用"校验 + 分流"模式。
校验通过的数据正常写入，校验失败的数据写入死信表并触发告警。
这样即使 Schema 规则写入有误，也不会造成整批数据丢失——你总是可以从死信中恢复。

五、总结

数据质量监控是 ETL 管线的前置防线。
Schema 校验捕获类型、范围、枚举等格式错误。
统计监控检测数据分布漂移。
两者组合覆盖格式异常和内容异常。
校验成本约 1-2% 的 ETL 时间，收益远大于代价。

如果只能选一项数据质量措施，选 Schema 校验。
它是成本最低、覆盖率最高、最容易落地的方案。
只需要一个 YAML 文件描述字段规则，就能拦截 80% 的数据质量问题。
等你建立了 Schema 校验的基础，再往上叠加统计监控和业务规则校验——这是一个渐进式的数据质量体系，而不是一次性的大工程。