调度、补数与幂等

本章目标：能讲清 DAG、依赖和调度时间，写出一个带重试和告警的 Airflow DAG，设计可重跑的补数方案，并处理跨天依赖、SLA 和数据质量门禁。 前置知识：读过 数仓分层、Hive 分区；会一点 Python 和 bash。

1. 为什么需要它

离线数仓每天要跑几十到几百个任务：同步 MySQL、清洗 ODS、构建 DWD、汇总 DWS、导出 ADS、跑质量检查。手工执行有三个问题：

• 顺序容易错：DWD 还没跑完就跑了 DWS，结果少一天数据；
• 失败没人知道：凌晨任务挂了，早上报表打不开才发现；
• 补数很痛苦：某天源数据晚到，要人工把十几张表按顺序重跑。

调度系统把任务和依赖写成 DAG（有向无环图），按时间自动触发，失败重试、超时告警、支持补数。它不负责“算得快”，负责“什么时候算、先算谁、失败怎么办”。

实时链路也有调度，但更多是常驻作业和 checkpoint；本章讲的是离线批任务的调度，实时链路见 实时数仓。

2. 核心概念

2.1 DAG、任务和依赖

ods_orders ──→ dwd_order_detail ──→ dws_trade_daily ──→ ads_gmv_daily ──→ quality_check
     ↑                                     ↑
ods_order_items ───────────────────────────┘

• DAG：一组任务和依赖关系，不能有环，否则永远跑不完。
• Task：一个最小执行单元，例如“跑一段 SQL”“提交一个 Spark 作业”。
• 依赖：A >> B 表示 A 成功后才跑 B；B 是 A 的下游，A 是 B 的上游。
• TaskInstance：某个 DAG 在某次调度中的某个任务实例，例如 dwd_order_detail 在 8 月 1 日这天的运行。
• 调度周期：多久触发一次，例如每天 02:00。
• 逻辑日期/数据区间：这次运行对应哪一天的数据。它和“实际开始运行的时间”不是一回事。

一个常见误解是“02:00 启动的任务处理当天数据”。实际通常处理前一天：8 月 2 日 02:00 启动，处理 dt=2018-08-01 的数据，因为 8 月 2 日还没过完。

面试高频：任务成功不等于数据就绪。上游任务显示成功，但产出的分区可能是空的，下游仍会算错，所以要有数据质量检查和就绪判断。

2.2 调度系统在做什么

Scheduler（解析 DAG、到点触发）
      ↓
Metadata DB（DAG、任务状态、历史）
      ↓
Executor（本地/ Celery / Kubernetes）
      ↓
Worker（真正执行任务）

• Scheduler：扫描 DAG 文件，按周期生成 TaskInstance，判断依赖是否满足。
• Executor：决定任务在哪里跑，本地进程、Celery 队列或 K8s Pod。
• Metadata DB：存 DAG、运行状态、重试次数、日志索引，是调度系统的核心状态。
• Worker：执行具体命令，例如 spark-submit、hive -f、python。
• XCom：任务间传小结果（如行数、文件路径），不适合传大数据。
• Connection/Variable：集中管理数据库连接和配置，避免把密码写进代码。
• Pool：限制某类任务的并发，防止补数时打满集群。

2.3 常见调度工具

Airflow 的 DAG 是代码，版本管理和 Code Review 方便；DolphinScheduler 上手快，运维界面友好；DataWorks 省去部署，但绑定云平台。选型看团队技术栈、部署环境和云厂商，不是“Airflow 最流行就一定选它”。

2.4 补数、重跑和幂等

补数（backfill）：为历史日期批量生成运行实例，例如把 8 月 1 日到 8 月 7 日重跑一遍。

重跑（rerun）：只重跑失败或结果错误的那一次实例。

补数的核心要求是幂等：同一日期跑 1 次和跑 10 次，结果一样。常见做法：

• 按分区覆盖写：INSERT OVERWRITE ... PARTITION (dt='${ds}')，而不是 INSERT INTO 追加；
• 先删目标分区再写，删和写在同一个事务或同一次提交里；
• 用唯一键 upsert，让重复数据覆盖旧值；
• 输入范围固定为某个数据区间，不依赖“当前时间”；
• 随机数、now() 等非确定性逻辑要固定种子或改用数据日期。

反例：INSERT INTO dws_trade_daily SELECT ... WHERE dt='${ds}'，重跑一次就多一份，指标翻倍。这种任务不能直接补数。

2.5 跨天依赖

有些指标不只依赖当天数据：

• 用户留存：8 月 8 日的 7 日留存，要等 8 月 1 日的新增用户和 8 月 8 日的活跃数据；
• 同比环比：今天的 GMV 要和昨天、上周同日对比；
• 累计指标：本月累计 GMV 依赖本月至今的每一天。

处理方式不是“多等一会儿”，而是显式建模依赖：

• 任务依赖：用 ExternalTaskSensor 等待另一个 DAG 的某次运行成功；
• 数据依赖：用 SQLSensor 或自定义 Sensor 检查目标分区是否有数据、行数是否达标；
• 偏移依赖：等待 7 天前的实例，需要设置执行日期偏移，而不是等“昨天”。

任务成功和数据就绪要分开看。上游任务成功但分区为空时，ExternalTaskSensor 会放行，SQLSensor 能拦住。

2.6 SLA、重试和告警

SLA（服务等级协议）：对产出时间的承诺，例如“每天 08:00 前产出 ads_gmv_daily”。Airflow 可以配置 sla，超过时间触发回调。

重试：适合网络抖动、临时锁冲突、下游短暂不可用。重试要设次数、间隔和超时；逻辑错误、数据错误、权限错误重试多少次都没用，应该直接失败并告警。

告警：失败告警、SLA 告警、数据质量告警要发到值班群或电话，附上 DAG、任务、日期、日志链接和排查入口。只发邮件、没人看，等于没有告警。

超时：任务卡住不返回时，execution_timeout 能杀掉并标记失败，避免占着资源一整天。

2.7 数据质量门禁

调度不只是“跑完”，还要判断“跑对”。常见检查放在 DWD/DWS 之后、ADS 之前：

质量任务要能“失败即停”，否则下游照样产出错误报表。检查 SQL 和口径见 数据质量。

3. 动手示例

3.1 本地启动 Airflow

Airflow 依赖较多，推荐用 Docker 的官方 quickstart 在本地跑（本地模拟，不是生产部署）。本课程固定 Airflow 2.10：

curl -LfO https://airflow.apache.org/docs/apache-airflow/2.10.5/docker-compose.yaml
docker compose up -d

启动后访问 localhost:8080，默认账号密码都是 airflow。DAG 文件放到 dags/ 目录，Scheduler 会自动扫描。Windows 上用 Git Bash 执行，Docker Desktop 需要先启动。

注意：Airflow 3.x 改了 Web UI、部分 CLI 和 provider 包的组织方式，本章命令按 2.10 写。线上用什么版本，就以那个版本的文档为准，不要把 2.10 的写法直接套到 3.x。

3.2 写一个带依赖和重试的 DAG

新建 dags/ecom_daily.py。下面用占位的任务命令演示 DAG 结构，脚本名按你的项目替换；项目线里的抽取和建仓入口见项目 00 章。

from datetime import datetime, timedelta

from airflow.decorators import dag
from airflow.operators.bash import BashOperator

@dag(
    dag_id="ecom_daily",
    start_date=datetime(2018, 8, 1),
    schedule="0 2 * * *",
    catchup=False,
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        "execution_timeout": timedelta(hours=1),
    },
    tags=["ecom"],
)
def ecom_daily():
    ods_orders = BashOperator(
        task_id="ods_orders",
        bash_command="python scripts/ingest_orders.py --dt {{ ds }}",
    )
    dwd = BashOperator(
        task_id="dwd_order_detail",
        bash_command="python scripts/build_dwd.py --dt {{ ds }}",
    )
    dws = BashOperator(
        task_id="dws_trade_daily",
        bash_command="python scripts/build_dws.py --dt {{ ds }}",
    )
    quality = BashOperator(
        task_id="quality_check",
        bash_command="python scripts/check_quality.py --dt {{ ds }}",
    )
    ods_orders >> dwd >> dws >> quality

ecom_daily()

几个点：

• schedule="0 2 * * *" 表示每天 02:00 触发；
• {{ ds }} 是这次运行对应的数据日期，模板会在运行时替换；
• retries=2 表示失败后最多再试两次，间隔 5 分钟；
• catchup=False 表示首次上线不自动补历史，避免一下子生成大量运行；
• execution_timeout 防止任务卡死；
• quality_check 放在最后，失败会阻断后续导出任务。

3.3 验证、补数和重跑

# 列出 DAG
airflow dags list

# 只跑某个任务的一次实例，用于调试
airflow tasks test ecom_daily dwd_order_detail 2018-08-01

# 补 8 月 1 日到 8 月 7 日，限制并发
airflow dags backfill ecom_daily \
  --start-date 2018-08-01 \
  --end-date 2018-08-07 \
  --max-active-runs 2

补数前先确认任务幂等、下游能承受并发、源数据已经就绪。补数不是“点一下重跑”，它会占用集群资源，也可能覆盖别人正在用的分区。

3.4 幂等的分区覆盖 SQL

Hive/Spark 中常见的幂等写法：

INSERT OVERWRITE TABLE dws_category_1d PARTITION (dt = '${ds}')
SELECT
  category,
  COUNT(DISTINCT order_id) AS order_cnt,
  SUM(item_amount_with_freight) AS gmv
FROM dwd_order_detail
WHERE dt = '${ds}'
GROUP BY category;

INSERT OVERWRITE 覆盖当天分区，重跑多少次结果都一样。前提是输入范围只取 dt='${ds}'，不能混入其他日期；如果源表有重复数据，还要先去重。

注意：Doris 等不支持 Hive 分区覆盖语法的引擎，可以用 Unique Key upsert，或先按条件删除再写入，并确保删除和写入在同一批次生效。

3.5 跨天依赖：7 日留存

8 月 8 日的 7 日留存，需要 8 月 1 日的新增用户和 8 月 8 日的活跃用户。可以在 DAG 里等待 7 天前的上游实例：

from airflow.sensors.external_task import ExternalTaskSensor

wait_d7 = ExternalTaskSensor(
    task_id="wait_d7_cohort",
    external_dag_id="ecom_daily",
    external_task_id="dws_user_daily",
    execution_delta=timedelta(days=7),
    mode="reschedule",
    timeout=60 * 60 * 2,
)

execution_delta 表示“等 7 天前的那个实例”，不是等昨天。mode="reschedule" 让 Sensor 不长期占用 worker。更稳的做法是再加一个 SQLSensor，检查 8 月 1 日分区确实有数据。

4. 常见坑

1. 把调度时间当数据日期。02:00 跑的任务通常处理昨天，SQL 里却写了当天日期，结果查不到数据。
2. 用追加写做补数。INSERT INTO 重跑会重复，指标翻倍；必须分区覆盖或 upsert。
3. 首次上线打开 catchup。start_date 写得很早，一上线生成几百个历史实例，集群被打满。
4. 只依赖任务成功，不检查数据。上游产出空分区，下游照样算，报表全为 0。
5. 非幂等任务配置自动重试。失败重试后写了两遍，比不重试更糟。
6. 跨天依赖靠“多等半小时”。上游延迟时仍然算错，应该用 Sensor 或数据就绪检查。
7. 没有超时和 SLA。任务卡住一整天，没人知道；早上才发现报表没出。
8. 补数不限并发。几十个日期同时跑，打满数据库和集群，影响正常调度。
9. 告警只发邮件。没人看的告警等于没有，要进值班群并明确责任人。
10. 质量检查放在导出之后。错误数据已经服务出去，再检查就晚了。

5. 练习

练习 1：画 DAG 并找关键路径

订单链路包含：ods_orders、ods_order_items、dwd_order_detail（依赖两个 ODS）、dws_trade_daily、ads_gmv_daily、quality_check、export_mysql。画出依赖，指出关键路径和可以并行的任务。

ods_orders ─────┐
                ├─→ dwd_order_detail → dws_trade_daily → ads_gmv_daily → quality_check → export_mysql
ods_order_items ┘

练习 2：把追加写改成幂等

某任务每天执行 INSERT INTO ads_gmv_daily SELECT ... WHERE dt='${ds}'，补数后 GMV 翻倍。写出修改方案，并说明还需要检查什么。

练习 3：补 7 天数据

dws_trade_daily 因为源数据延迟，8 月 1 日到 8 月 7 日都要重跑。写出补数命令和至少三个注意事项。

airflow dags backfill ecom_daily \
  --start-date 2018-08-01 \
  --end-date 2018-08-07 \
  --max-active-runs 2

练习 4：设计跨天依赖

ads_user_retention 需要当天活跃和 7 天前的新增用户。说明如何保证 7 天前的分区已经就绪，任务成功是否足够。

练习 5：SLA 失败排查

ads_gmv_daily 承诺每天 08:00 前产出，今天 08:30 还没好。列出排查顺序。

练习 6：重试还是直接失败

判断下列情况应该自动重试还是直接失败并告警：数据库连接超时；SQL 字段名写错；源分区为空；下游服务短暂不可用；主键重复导致写入失败。

6. 面试延伸

面试题：Airflow、DolphinScheduler、DataWorks 怎么选？

是什么：三者都是调度平台。Airflow 用 Python 定义 DAG，DolphinScheduler 以可视化 DAG 为主，DataWorks 是云上托管的一站式数据开发平台。

解决什么问题：批任务的依赖编排、定时触发、失败重试、补数和告警。不同团队在开发方式、部署成本和云环境上需求不同。

我的实操边界：我在本地用 Docker 跑过 Airflow，写过带依赖和重试的 DAG，用命令行做过补数；DolphinScheduler 和 DataWorks 了解界面和定位，没有在生产环境运维。选型看团队是否会写 Python、是否上云、是否需要多租户和可视化运维。

面试题：补数怎么保证幂等？

是什么：同一数据日期重跑多次，结果保持一致。核心是分区覆盖写、唯一键 upsert 和确定的输入范围。

解决什么问题：补数、失败重试、上游延迟都会导致同一日期被多次计算，追加写会让指标翻倍。

我的实操边界：我在本地把追加写改成 INSERT OVERWRITE 分区覆盖，验证过重跑结果不变；没有在生产环境处理过跨多天的复杂补数。回答时要说清：覆盖同一分区、过滤条件固定、去重、删除和写入原子生效，最后用行数和指标抽查验证。

面试题：跨天依赖和 SLA 怎么处理？

是什么：跨天依赖是任务需要昨天、7 天前或本月累计的数据；SLA 是对产出时间的承诺。

解决什么问题：只按天串行会漏掉历史窗口；只靠人工等待无法保证准时和正确。

我的实操边界：我理解 ExternalTaskSensor、SQLSensor 和偏移依赖的用法，本地写过简单 Sensor；没有设计过公司级 SLA 体系。做法是显式建模依赖、检查数据就绪、设置超时和分级告警，并对关键路径做 SLA 监控。

面试题：调度里怎么做数据质量？

是什么：在关键层之后增加质量检查任务，校验行数、唯一性、空值、值域和波动，失败则阻断下游。

解决什么问题：任务成功不代表数据正确，错误数据流到报表后修复成本更高。

我的实操边界：我在项目里写过行数、主键唯一和空值率检查 SQL；没有搭建过完整的质量平台。落地时把检查做成独立任务或门禁，配置阈值和告警，严重问题阻断，轻微问题告警并留痕。

7. 本章验收

• 能画出 ODS → DWD → DWS → ADS 的 DAG，指出上下游和关键路径。
• 能解释调度时间、逻辑日期和数据区间的关系，不把 02:00 当成处理当天。
• 能写出一个带依赖、重试、超时和告警的 Airflow DAG。
• 能把追加写改成幂等的分区覆盖写，并说明重跑为什么不翻倍。
• 能执行补数命令，并说出并发、源数据、依赖顺序三个注意事项。
• 能用 Sensor 或数据检查处理跨天依赖，区分任务成功和数据就绪。
• 能设计行数、唯一性、空值、值域和波动检查，并让质量任务阻断下游。
• 能面对 SLA 超时，按调度、上游、资源、日志的顺序排查。