100个ETL任务谁先跑谁后跑——DAG调度的基本原理

数仓里有几百个ETL任务，它们之间有复杂的依赖关系——表A依赖表B和表C，表D依赖表A和表E……怎么确定执行顺序？这就是DAG调度解决的核心问题。在某短视频平台用Airflow管理100+个ETL任务时，最初没有梳理清楚依赖关系，经常出现'下游任务先于上游执行、读到了昨天的旧数据'的问题。后来花了一周理清所有依赖关系构建DAG，问题彻底解决。面试官问调度原理，考的就是你对DAG和任务编排的理解。

DAG的核心概念

# 一个典型的数仓DAG
#
#  数据源A ──→ ODS_A ──→ DWD_订单 ──→ DWS_日报 ──→ ADS_报表
#                           ↑
#  数据源B ──→ ODS_B ──→ DWD_用户 ─┘
#
# 节点（Task）：ODS_A, ODS_B, DWD_订单, DWD_用户, DWS_日报, ADS_报表
# 边（Dependency）：ODS_A完成后才跑DWD_订单，DWD_订单和DWD_用户都完成后才跑DWS
#
# 拓扑排序的执行顺序：
# 第1轮：ODS_A, ODS_B（没有上游依赖，可以并行）
# 第2轮：DWD_订单, DWD_用户（上游ODS完成后触发，可以并行）
# 第3轮：DWS_日报（DWD都完成后触发）
# 第4轮：ADS_报表（DWS完成后触发）
#
# 关键路径：ODS_A → DWD_订单 → DWS_日报 → ADS_报表
# 关键路径决定了整个DAG的最短完成时间

调度的三种触发方式

Airflow中的DAG定义

# Airflow用Python代码定义DAG和依赖关系
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
 
with DAG('daily_dw_etl',
         schedule='0 1 * * *',  # 每天凌晨1点
         start_date=datetime(2024, 1, 1),
         catchup=False) as dag:
 
    ods_orders = PythonOperator(
        task_id='ods_orders',
        python_callable=extract_orders)
 
    ods_users = PythonOperator(
        task_id='ods_users',
        python_callable=extract_users)
 
    dwd_orders = PythonOperator(
        task_id='dwd_orders',
        python_callable=transform_orders)
 
    dws_daily = PythonOperator(
        task_id='dws_daily',
        python_callable=aggregate_daily)
 
    # 定义依赖关系
    ods_orders >> dwd_orders  # ODS完成后跑DWD
    ods_users >> dwd_orders   # 用户表也要完成
    dwd_orders >> dws_daily   # DWD完成后跑DWS
    # 等价于：dwd_orders.set_upstream([ods_orders, ods_users])