🗓️ 06 · 数据平台与调度：DAG 编排 / 依赖 / 重跑
> 数据平台 = 调度（Airflow/DolphinScheduler）+ 血缘 + 质量 + 权限。
> 本节手写**任务 DAG、依赖调度、失败重试、血缘追踪**（设 mini 调度器）。
📋 本节要解决的问题
| # | 问题 | 方法 |
|---|---|---|
| 1 | 数据平台有哪些组件？ | 调度/血缘/质量/权限 |
| 2 | 任务依赖怎么表达？ | DAG（有向无环图） |
| 3 | 失败怎么处理？ | 重试/告警/上游阻断 |
| 4 | 血缘怎么维护？ | 表级依赖记录 |
%matplotlib inline
import matplotlib
import matplotlib.pyplot as plt
matplotlib.rcParams['font.sans-serif'] = ['Microsoft YaHei', 'SimHei', 'PingFang SC', 'Noto Sans CJK SC', 'Arial Unicode MS']
matplotlib.rcParams['axes.unicode_minus'] = False
import numpy as np
import pandas as pd
import sqlite3
import os
rng = np.random.default_rng(42)
1️⃣ 数据平台全景（必背）
print('''数据平台三大块：
1) 开发平台：SQL 编辑器/任务发布/版本管理（如 DataWorks/LH）；
2) 调度系统：DAG 编排、定时、依赖、重跑（Airflow/DolphinScheduler）；
3) 治理体系：血缘、质量（规则稽核）、元数据、权限审计。
面试常问："你们调度任务失败怎么处理？血缘有什么用？质量怎么保证？"
''')数据平台三大块：
1) 开发平台：SQL 编辑器/任务发布/版本管理（如 DataWorks/LH）；
2) 调度系统：DAG 编排、定时、依赖、重跑（Airflow/DolphinScheduler）；
3) 治理体系：血缘、质量（规则稽核）、元数据、权限审计。
面试常问："你们调度任务失败怎么处理？血缘有什么用？质量怎么保证？"
2️⃣ 构造任务 DAG（字典表示）
# 任务 DAG：每个任务依赖前序任务的产出
tasks = {
'ods_sync': [],
'dwd_clean': ['ods_sync'],
'dws_agg': ['dwd_clean'],
'ads_report': ['dws_agg'],
'ads_feature': ['dws_agg'],
}
for t, deps in tasks.items():
print(f' {t:<12} 依赖: {deps or "无（源任务）"}')
print('\n➡️ DAG 保证：ads_report 和 ads_feature 都等 dws_agg 完成') ods_sync 依赖: 无（源任务）
dwd_clean 依赖: ['ods_sync']
dws_agg 依赖: ['dwd_clean']
ads_report 依赖: ['dws_agg']
ads_feature 依赖: ['dws_agg']
➡️ DAG 保证：ads_report 和 ads_feature 都等 dws_agg 完成
3️⃣ 拓扑排序：找出可执行顺序
from collections import deque
def topo_sort(deps):
indeg = {t: len(d) for t, d in deps.items()}
adj = {t: [] for t in deps}
for t, ds in deps.items():
for d in ds:
adj[d].append(t)
q = deque([t for t, i in indeg.items() if i == 0])
order = []
while q:
cur = q.popleft(); order.append(cur)
for nxt in adj[cur]:
indeg[nxt] -= 1
if indeg[nxt] == 0:
q.append(nxt)
return order, len(order) == len(deps)
order, ok = topo_sort(tasks)
print('拓扑顺序:', order)
print('DAG 无环:', ok)
assert ok and order[0] == 'ods_sync'
print('✅ DAG 合法，调度顺序确定')拓扑顺序: ['ods_sync', 'dwd_clean', 'dws_agg', 'ads_report', 'ads_feature']
DAG 无环: True
✅ DAG 合法，调度顺序确定
4️⃣ 失败重试与依赖阻断（模拟调度器）
# 模拟：dwd_clean 第一次失败，重试 2 次成功；观察下游被阻断
def run_task(name, fail_first, state):
if fail_first and state.get(name, 0) == 0:
state[name] = state.get(name, 0) + 1
return False
state[name] = state.get(name, 0) + 1
return True
state = {}
plan = ['ods_sync', 'dwd_clean', 'dws_agg', 'ads_report']
fails = {'dwd_clean': 1} # 第一次失败
log = []
for t in plan:
attempt = 0
success = False
while attempt < 3 and not success:
attempt += 1
success = run_task(t, fails.get(t, 0) >= attempt, state)
log.append((t, attempt, success))
if not success:
print(f'❌ {t} 重试 3 次仍失败 -> 阻断下游')
break
print(f'✅ {t} 第 {attempt} 次成功')
print('\n➡️ 调度策略：指数退避重试 + 依赖失败阻断下游 + 告警通知')✅ ods_sync 第 1 次成功
✅ dwd_clean 第 2 次成功
✅ dws_agg 第 1 次成功
✅ ads_report 第 1 次成功
➡️ 调度策略：指数退避重试 + 依赖失败阻断下游 + 告警通知
# 可视化：调度 DAG 与失败重试位置
fig, ax = plt.subplots(figsize=(10, 4))
pos = {'ods_sync': (8, 50), 'dwd_clean': (32, 50), 'dws_agg': (56, 50),
'ads_report': (80, 72), 'ads_feature': (80, 28)}
for t, (x, y) in pos.items():
fc, ec = ('#F8CBAD', '#C00000') if t == 'dwd_clean' else ('#E2EFDA', '#548235')
ax.add_patch(plt.Rectangle((x - 7, y - 7), 15, 14, fc=fc, ec=ec))
ax.text(x, y, t, ha='center', va='center', fontsize=10)
for a, b in [('ods_sync', 'dwd_clean'), ('dwd_clean', 'dws_agg'),
('dws_agg', 'ads_report'), ('dws_agg', 'ads_feature')]:
ax.annotate('', xy=(pos[b][0] - 7.5, pos[b][1]), xytext=(pos[a][0] + 8, pos[a][1]),
arrowprops=dict(arrowstyle='-|>', color='#7F7F7F', lw=1.8))
ax.annotate('第1次失败\n重试成功', (32, 57), textcoords='offset points', xytext=(-8, 22),
fontsize=10.5, color='#C00000', ha='center',
arrowprops=dict(arrowstyle='->', color='#C00000'))
ax.text(56, 40, '重试成功前\n下游全部阻断', ha='center', fontsize=10, color='#C00000')
ax.set_xlim(0, 96); ax.set_ylim(10, 92); ax.axis('off')
ax.set_title('任务 DAG：拓扑顺序执行 + 失败重试 + 依赖阻断', fontsize=12)
plt.show()
数据时间 vs 运行时间（T-1 详解）
• **运行时间**：任务真实被调度执行的时刻。如"每天 01:30 跑"；
• **数据时间**：这批数据对应的**业务日期**。01:30 跑的是"昨天（T-1）"的数据。
为什么是 T-1？白天产生的数据要等当天结束才完整，凌晨汇总昨天的全量——
所以"今天的调度产出昨天的数"。排查数据问题时先问三件事：
**任务跑到几点？处理的数据时间（ds 分区）是哪天？是全量还是增量？**
**补数/回刷（backfill）**：历史分区数据错了或逻辑变更时，按分区批量重跑——
这正是"表按日期分区"的最大好处之一：**重跑粒度=天级分区，互不影响**。
Airflow vs DolphinScheduler（选型对比）
| 维度 | Airflow | DolphinScheduler |
|---|---|---|
| 定位 | Python 生态编排 | 国产可视化大数据调度 |
| 定义任务 | Python 代码（DAG as code） | 界面拖拽 + 代码均可 |
| 依赖/重试 | 代码配置 | 界面配置 |
| 生态 | 插件丰富、社区大 | 对国产组件/大数据栈友好 |
| 适用 | 算法/工程团队代码化 | 数据平台多人协作 |
还有主流选择：Dagster/Prefect（数据感知编排）、K8s 原生 Argo Workflows。
面试口径：**说清自己团队用什么 + 为什么**（易用性/生态/权限/多租户）。
5️⃣ 血缘追踪（表级依赖）
# 血缘：记录表与表之间的来源关系
lineage = [
('ods_orders', 'dwd_orders'),
('ods_users', 'dwd_orders'),
('dwd_orders', 'dws_daily_orders'),
('dws_daily_orders', 'ads_gmv_report'),
('dws_daily_orders', 'ads_user_feature'),
]
print('血缘（上游表 -> 下游表）:')
for up, down in lineage:
print(f' {up} -> {down}')
print('''
血缘的用途：
1) 影响分析：改一张 ODS 表，下游哪些报表/特征会变；
2) 故障排查：报表数据异常时快速定位源头；
3) 成本治理：找无下游的"孤儿表"下线省钱。
''')血缘（上游表 -> 下游表）:
ods_orders -> dwd_orders
ods_users -> dwd_orders
dwd_orders -> dws_daily_orders
dws_daily_orders -> ads_gmv_report
dws_daily_orders -> ads_user_feature
血缘的用途：
1) 影响分析：改一张 ODS 表，下游哪些报表/特征会变；
2) 故障排查：报表数据异常时快速定位源头；
3) 成本治理：找无下游的"孤儿表"下线省钱。
6️⃣ 调度系统面试速答
print('''1) 调度频率怎么定？ -> 分层：小时级/天级按数据时间（T-1 等）；依赖更细的任务用分钟级；
2) 数据时间 vs 运行时间？-> 今天 01:00 跑昨天的数据（T-1）；数据时间=业务时间；
3) 天级任务凌晨跑不完怎么办？-> 提前开始时间/并行度/拆任务/增量；
4) 任务积压如何追？-> 提高并行度、跳过不必要重跑、优先级；
5) 平台差异：Airflow（Python 生态）vs DolphinScheduler（可视化）.
''')1) 调度频率怎么定？ -> 分层：小时级/天级按数据时间（T-1 等）；依赖更细的任务用分钟级；
2) 数据时间 vs 运行时间？-> 今天 01:00 跑昨天的数据（T-1）；数据时间=业务时间；
3) 天级任务凌晨跑不完怎么办？-> 提前开始时间/并行度/拆任务/增量；
4) 任务积压如何追？-> 提高并行度、跳过不必要重跑、优先级；
5) 平台差异：Airflow（Python 生态）vs DolphinScheduler（可视化）.
📝 自测清单
• [ ] 能画出数据平台三块全景
• [ ] 能写 DAG 并表示任务依赖
• [ ] 能实现拓扑排序
• [ ] 能讲失败重试与依赖阻断策略
• [ ] 能说出血缘三大用途