数据工程实战课程 · 编排与治理阶段
好的DAG设计是可维护、可观测、可恢复的，本课学习常见设计模式与最佳实践。
# DAG设计模式
import time, random
from datetime import datetime
from collections import defaultdict
class TaskNode:
def __init__(self, name, fn=None):
self.name = name; self.fn = fn; self.downstream = []
def __rshift__(self, other): self.downstream.append(other); return other
class DAGBuilder:
def __init__(self, dag_id):
self.dag_id = dag_id; self.tasks = {}
def task(self, name, fn=None):
t = TaskNode(name, fn); self.tasks[name] = t; return t
def run(self):
executed = set(); results = {}
while len(executed) < len(self.tasks):
for name, task in self.tasks.items():
if name in executed: continue
upstream = [n for n, t in self.tasks.items() if name in [d.name for d in t.downstream]]
if all(u in executed for u in upstream):
start = time.perf_counter()
if task.fn:
ctx = {k: results[k] for k in upstream if k in results}
results[name] = task.fn(ctx)
executed.add(name)
ms = (time.perf_counter()-start)*1000
print(f" {name}: {ms:.0f}ms")
# 扇出模式
print("扇出模式:")
dag1 = DAGBuilder('fan_out')
src = dag1.task('source', lambda ctx: {'data': list(range(100))})
dag1.task('branch_a', lambda ctx: {'count': len(ctx.get('source',{}).get('data',[]))})
dag1.task('branch_b', lambda ctx: {'sum': sum(ctx.get('source',{}).get('data',[]))})
dag1.task('branch_c', lambda ctx: {'avg': sum(ctx.get('source',{}).get('data',[]))/100})
merge = dag1.task('merge', lambda ctx: 'merged')
src >> dag1.tasks['branch_a'] >> merge
src >> dag1.tasks['branch_b'] >> merge
src >> dag1.tasks['branch_c'] >> merge
dag1.run()
# 动态DAG
print("\n动态DAG:")
regions = ['华东','华南','华北']
dag2 = DAGBuilder('dynamic')
parent = dag2.task('start', lambda ctx: 'ok')
for region in regions:
t = dag2.task(f'process_{region}', lambda ctx, r=region: f'{r}_done')
parent >> t
dag2.run()
print("✅ 验证通过 - DAG设计模式运行正常")🎁 下一课预告：dbt数据转换！
| 级别 | 特征 | 编排方式 | 治理水平 |
|---|---|---|---|
| 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、数据产品规划 |
| 数据管家 | 数据质量与治理 | 质量规则、数据字典 |
选择技术方案时，需要从多个维度综合评估，避免仅凭流行度或个人偏好决策：
| 评估维度 | 权重 | 评估要点 |
|---|---|---|
| 业务适配度 | 30% | 是否满足功能需求？延迟/吞吐是否达标？ |
| 团队能力 | 25% | 团队是否有经验？学习曲线如何？ |
| 社区生态 | 20% | 社区活跃度？文档质量？第三方集成？ |
| 运维成本 | 15% | 部署复杂度？监控能力？故障恢复？ |
| 成本效益 | 10% | License费用？资源消耗？扩展成本？ |
数据量级？ ├── <1GB/天 → SQLite/PostgreSQL + Python脚本 ├── 1-100GB/天 → 数据仓库(Snowflake/BQ) + dbt ├── 100GB-1TB/天 → Spark + 数据湖 └── >1TB/天 → Spark/Flink + 湖仓一体 延迟要求？ ├── T+1 → 批处理(Spark) + Airflow ├── 小时级 → 微批(Spark Streaming) + Airflow ├── 分钟级 → 流处理(Flink) + Kafka └── 秒级 → 纯流(Flink) + Kafka + 缓存 数据类型？ ├── 纯结构化 → 数据仓库 ├── 结构化+半结构化 → 湖仓一体 └── 全类型 → 数据湖 + 元数据管理