数据工程实战课程 · 编排与治理阶段
数据质量是数据工程的生命线，高质量数据是可靠决策的前提。
# 数据质量框架
import random, time
from datetime import datetime
from collections import defaultdict
from dataclasses import dataclass
@dataclass
class QualityRule:
name: str; field: str; rule_type: str; params: dict
severity: str = 'error' # error, warning, info
@dataclass
class QualityResult:
rule_name: str; passed: bool; total: int; failed: int; details: str
class DataQualityEngine:
def __init__(self):
self.rules = []; self.results = []
def add_rule(self, rule): self.rules.append(rule)
def validate(self, records):
self.results = []
for rule in self.rules:
total = len(records)
failed = 0
for r in records:
val = r.get(rule.field)
if rule.rule_type == 'not_null' and (val is None or val == ''): failed += 1
elif rule.rule_type == 'unique':
vals = [r.get(rule.field) for r in records]
if vals.count(val) > 1: failed += 1
elif rule.rule_type == 'range':
if isinstance(val,(int,float)) and (val < rule.params.get('min',float('-inf')) or val > rule.params.get('max',float('inf'))): failed += 1
elif rule.rule_type == 'accepted_values':
if val not in rule.params.get('values',[]): failed += 1
elif rule.rule_type == 'regex':
import re
if not re.match(rule.params.get('pattern','.*'), str(val)): failed += 1
elif rule.rule_type == 'custom':
if not rule.params.get('fn', lambda x: True)(val): failed += 1
passed = failed == 0
self.results.append(QualityResult(rule.name, passed, total, failed,
f"{failed}/{total} failed ({failed/total*100:.1f}%)" if total else "no records"))
return self.results
def quality_score(self):
if not self.results: return 0
weights = {'error': 1.0, 'warning': 0.5, 'info': 0.1}
total_weight = sum(weights.get(r.severity, 1.0) for r in self.rules)
passed_weight = sum(weights.get(r.severity, 1.0) for r, res in zip(self.rules, self.results) if res.passed)
return round(passed_weight / total_weight * 100, 1) if total_weight else 0
# 生成测试数据
data = []
for i in range(500):
data.append({
'id': i+1,
'email': random.choice([f'user{i}@test.com', '', None]),
'age': random.choice([random.randint(18,80), None, -1, 150]),
'amount': random.choice([round(random.uniform(0,10000),2), None, -100]),
'status': random.choice(['active','inactive','pending','INVALID']),
'phone': random.choice([f'1{random.randint(3-9,9)}{random.randint(100000000,999999999)}', 'invalid', ''])
})
# 定义质量规则
engine = DataQualityEngine()
engine.add_rule(QualityRule('email_not_null','email','not_null',{},'error'))
engine.add_rule(QualityRule('age_range','age','range',{'min':0,'max':120},'error'))
engine.add_rule(QualityRule('amount_positive','amount','range',{'min':0,'max':1000000},'warning'))
engine.add_rule(QualityRule('status_valid','status','accepted_values',{'values':['active','inactive','pending']},'error'))
engine.add_rule(QualityRule('id_unique','id','unique',{},'error'))
results = engine.validate(data)
print("数据质量报告:")
for r in results:
icon = "✅" if r.passed else "❌"
print(f" {icon} {r.rule_name}: {r.details}")
print(f"\n综合质量评分: {engine.quality_score()}/100")
print("✅ 验证通过 - 数据质量框架运行正常")🎁 下一课预告：数据血缘！
| 级别 | 特征 | 编排方式 | 治理水平 |
|---|---|---|---|
| L1 | cron+脚本 | 无统一调度 | 无治理 |
| L2 | Airflow/Dbt | DAG编排 | 基本质量检查 |
| L3 | 编排+治理一体 | 自动依赖发现 | 数据目录+血缘 |
| L4 | AI辅助编排 | 自适应调度 | 主动质量治理 |
在生产环境中，我们经常遇到教科书不会告诉你的问题。以下是常见的实战经验和解决方案：
| 指标 | 告警阈值 | 监控方式 |
|---|---|---|
| 管道延迟 | 超过SLA 120% | 端到端时间戳追踪 |
| 数据量偏差 | 日均偏差超过30% | 与历史同期对比 |
| 错误率 | 超过0.1% | 错误计数/总记录数 |
| 数据新鲜度 | 超过SLA 150% | 最新数据时间戳 |
| 资源利用率 | CPU持续>80% | 系统监控 |
A: 根据业务延迟要求决定：秒级用流处理、分钟级用微批、小时级用批处理。大部分场景批处理足够，流处理用于实时业务场景（风控、监控、推荐）。
A: 不会完全取代，但边界在模糊。湖仓一体（如Iceberg/Delta Lake）正在融合两者优势。未来趋势是统一存储+多引擎计算。
A: 先用云原生服务（BigQuery/Snowflake + Fivetran + dbt + Looker），验证业务后再考虑自建。过早优化是万恶之源。
A: 用钱说话：数据质量问题的直接经济损失（错误决策、合规罚款、重复劳动）远大于治理投入。量化问题是第一步。
| 角色 | 职责 | 交付物 |
|---|---|---|
| 数据工程师 | 管道开发与维护 | 数据管道、基础设施 |
| 数据分析师 | 数据分析与报表 | SQL查询、BI报表 |
| 数据科学家 | 模型与实验 | ML模型、实验报告 |
| 数据产品经理 | 需求与优先级 | PRD、数据产品规划 |
| 数据管家 | 数据质量与治理 | 质量规则、数据字典 |