未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

warning 适用范围与待维护事项

本文是 Airflow 2.x 风格的编排片段与练习，未验证具体核心及 provider 版本，不保证兼容 3.x；需自行补齐导入、业务函数、连接、脚本和配置。ds 为逻辑日期，周一至周五判断不含节假日，10000 行阈值仅为教学示例。循环在解析期生成任务，并非运行期动态映射。all_done 叶子可能掩盖上游失败；跨 DAG 重复触发需处理 run_id 冲突。tasks test 不能替代完整 DAG 调度验证；ExternalTaskSensor 的机制仍需补充。

任务依赖与编排

abstract 本文核心内容

>> 语法与多种依赖关系的表达方式
cross_downstream 和 chain 简化复杂依赖
BranchPythonOperator 实现条件分支
TriggerDagRunOperator 跨 DAG 触发
TaskGroup 组织大型 DAG
动态 DAG 生成（循环生成 Task）

基础依赖语法：`>>`

Airflow 用 >> 表示"先做左边，再做右边"的依赖关系。

# 链式依赖：A → B → C
task_a >> task_b >> task_c

# 等价写法（用 set_downstream/set_upstream）
task_a.set_downstream(task_b)
task_b.set_downstream(task_c)

# 反向箭头（效果一样，只是方向表达不同）
task_c << task_b << task_a

扇出（一对多）

# 默认 all_success 下，A 成功后 B 和 C 才可被调度；能否同时执行取决于执行器、资源及并发限制
task_a >> [task_b, task_c]

# 等价写法
task_a >> task_b
task_a >> task_c

扇入（多对一）

# 默认 all_success 下，B 和 C 都成功后，D 才可被调度
[task_b, task_c] >> task_d

菱形依赖（扇出再扇入）

# A → [B, C] → D
# B 和 C 无彼此依赖，可并行调度；默认 all_success 下，两者都成功后 D 才可被调度

task_a >> [task_b, task_c] >> task_d

cross_downstream：避免重复代码

当有多对多的依赖关系时，手写 >> 会很繁琐，用 cross_downstream 更简洁：

from airflow.models.baseoperator import cross_downstream

# 手写方式（繁琐）
task_a >> task_c
task_a >> task_d
task_b >> task_c
task_b >> task_d

# cross_downstream 方式（简洁）
# 将 [a, b] 的每个任务都连接到 [c, d] 的每个任务
cross_downstream([task_a, task_b], [task_c, task_d])

chain：表达线性依赖链

from airflow.models.baseoperator import chain

# 手写方式
task_a >> task_b
task_b >> task_c
task_c >> task_d

# chain 方式
chain(task_a, task_b, task_c, task_d)

# chain 也支持列表（并行步骤）
# a → [b, c] → [d, e] → f
chain(task_a, [task_b, task_c], [task_d, task_e], task_f)
# 注意：chain 中相邻的两个列表长度必须一致；此处按位置连接 b→d、c→e，而非多对多连接

BranchPythonOperator：条件分支

根据业务逻辑决定执行哪个分支，跳过其他分支。

场景：根据数据量选择处理方式

from airflow.operators.python import BranchPythonOperator
from airflow.operators.empty import EmptyOperator

def choose_processing_branch(**context):
    """
    根据数据量选择处理策略：
    - 数据量 < 10000：轻量处理（用 Python 直接跑）
    - 数据量 >= 10000：重量处理（提交 Spark 作业）
    返回值是要执行的 task_id（字符串或列表）
    """
    ti = context["ti"]
    # 从上游任务拿到数据量
    metrics = ti.xcom_pull(task_ids="count_data", key="return_value")
    row_count = metrics["row_count"]

    print(f"数据量：{row_count}")

    if row_count < 10000:
        return "process_with_python"  # 返回要执行的 task_id
    else:
        return "process_with_spark"

# 分支决策任务
branch_task = BranchPythonOperator(
    task_id="choose_branch",
    python_callable=choose_processing_branch,
)

# 两个分支任务
python_process = PythonOperator(
    task_id="process_with_python",
    python_callable=run_python_etl,
)

spark_process = SparkSubmitOperator(
    task_id="process_with_spark",
    application="/jobs/spark_etl.py",
    # ... spark 配置
)

# 合并分支后的汇总任务
# trigger_rule="none_failed_min_one_success"：上游均已结束、没有 failed 或 upstream_failed，且至少一个 success 时继续
summarize = PythonOperator(
    task_id="summarize_result",
    python_callable=generate_summary,
    trigger_rule="none_failed_min_one_success",  # 关键！
)

count_task >> branch_task >> [python_process, spark_process] >> summarize

warning 分支后需要继续执行的合并任务通常应调整 trigger_rule

默认的 trigger_rule="all_success" 要求所有上游任务都成功，但分支中被跳过的任务状态是 skipped，不是 success。若未选分支被 skipped，默认 all_success 的合并任务通常也会被跳过；若上游失败，则可能被标记为 upstream_failed。解决方案：将合并任务的 trigger_rule 设为 "none_failed_min_one_success"。

trigger_rule 速查表

trigger_rule	含义
`all_success`（默认）	所有上游任务成功才触发
`all_failed`	所有上游任务均为 failed 或 upstream_failed 才触发
`all_done`	所有上游任务完成（不论成功失败）才触发
`one_success`	至少一个上游任务成功即触发
`one_failed`	至少一个上游任务为 failed 或 upstream_failed 即可触发，不要求等待所有上游结束
`none_failed`	所有上游任务均已结束且没有 failed 或 upstream_failed（允许 skipped）才触发
`none_failed_min_one_success`	所有上游任务均已结束，没有 failed 或 upstream_failed，且至少一个 success 才触发

场景：工作日/非工作日不同处理

from airflow.operators.python import BranchPythonOperator
import pendulum

def is_workday(**context):
    """判断执行日期是否是工作日"""
    ds = context["ds"]
    date = pendulum.parse(ds)
    # 按逻辑日期区分周一至周五（weekday 0-4）与周末；不处理法定节假日或调休
    if date.weekday() < 5:
        return "workday_processing"
    else:
        return "weekend_processing"

branch = BranchPythonOperator(
    task_id="check_workday",
    python_callable=is_workday,
)

workday_task = PythonOperator(
    task_id="workday_processing",
    python_callable=run_workday_etl,
)

weekend_task = PythonOperator(
    task_id="weekend_processing",
    python_callable=run_weekend_etl,
)

# 两个分支都不需要合并，各自结束
branch >> [workday_task, weekend_task]

TriggerDagRunOperator：跨 DAG 触发

在一个 DAG 中触发另一个 DAG 运行。

from airflow.operators.trigger_dagrun import TriggerDagRunOperator

# 场景：ODS ETL 完成后，自动触发 DWS 层的聚合 DAG
trigger_dws_dag = TriggerDagRunOperator(
    task_id="trigger_dws_aggregation",
    trigger_dag_id="dws_order_aggregation",       # 被触发的 DAG ID
    # 传递配置给被触发的 DAG（通过 dag_run.conf 读取）
    conf={"source_date": "{{ ds }}", "env": "prod"},
    # wait_for_completion=True 表示等待被触发的 DAG 完成
    wait_for_completion=True,
    # 如果被触发的 DAG 失败，本 Task 也失败
    failed_states=["failed"],
    # 触发后的 run_id 命名规则
    trigger_run_id="ods_triggered_{{ ds_nodash }}",
)

# 在被触发的 DAG 中读取 conf
def use_trigger_config(**context):
    conf = context["dag_run"].conf
    source_date = conf.get("source_date", context["ds"])
    print(f"使用触发日期：{source_date}")

TaskGroup：组织大型 DAG

当 DAG 的 Task 很多时，用 TaskGroup 把相关任务分组，Web UI 中可以折叠显示，更清晰。

from airflow.utils.task_group import TaskGroup

with DAG("large_etl_dag", ...) as dag:

    # ============================================================
    # 任务组 1：数据采集阶段
    # ============================================================
    with TaskGroup(group_id="extraction") as extraction_group:
        extract_orders = BashOperator(
            task_id="extract_orders",
            bash_command="python extract_orders.py --date {{ ds }}",
        )
        extract_users = BashOperator(
            task_id="extract_users",
            bash_command="python extract_users.py --date {{ ds }}",
        )
        extract_products = BashOperator(
            task_id="extract_products",
            bash_command="python extract_products.py --date {{ ds }}",
        )
        # 组内任务并行（没有显式依赖）

    # ============================================================
    # 任务组 2：数据清洗阶段
    # ============================================================
    with TaskGroup(group_id="transformation") as transformation_group:
        clean_orders = PythonOperator(
            task_id="clean_orders",
            python_callable=clean_order_data,
        )
        clean_users = PythonOperator(
            task_id="clean_users",
            python_callable=clean_user_data,
        )
        # 清洗后做联合检查
        validate = PythonOperator(
            task_id="validate_consistency",
            python_callable=validate_data_consistency,
        )
        [clean_orders, clean_users] >> validate

    # ============================================================
    # 任务组 3：数据加载阶段
    # ============================================================
    with TaskGroup(group_id="loading") as loading_group:
        load_to_ods = HiveOperator(
            task_id="load_to_ods",
            hql="INSERT OVERWRITE ...",
        )
        load_to_dws = HiveOperator(
            task_id="load_to_dws",
            hql="INSERT OVERWRITE ...",
        )
        load_to_ods >> load_to_dws

    # 组间依赖（整组对整组）
    extraction_group >> transformation_group >> loading_group

Web UI 中看到的是三个折叠的组（Extraction、Transformation、Loading），展开后才看到组内的 Task，界面清晰很多。

动态 DAG 生成

Airflow 的 DAG 是 Python 代码，意味着可以用循环动态生成 Task。

场景：为每张业务表生成相同的 ETL Task

# 需要同步的业务表列表（可以从数据库或配置文件读取）
TABLES_TO_SYNC = [
    {"table": "orders", "key_col": "order_id", "partition_col": "create_time"},
    {"table": "users", "key_col": "user_id", "partition_col": "register_time"},
    {"table": "products", "key_col": "product_id", "partition_col": "update_time"},
    {"table": "payments", "key_col": "payment_id", "partition_col": "pay_time"},
]

with DAG(
    dag_id="dynamic_table_sync",
    schedule_interval="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:

    start = EmptyOperator(task_id="start")
    end = EmptyOperator(task_id="end", trigger_rule="all_done")

    for table_config in TABLES_TO_SYNC:
        table_name = table_config["table"]

        # 为每张表动态生成两个 Task：抽取 + 验证
        with TaskGroup(group_id=f"sync_{table_name}") as table_group:

            extract = BashOperator(
                task_id=f"extract_{table_name}",
                bash_command=f"""
                    python extract.py \
                        --table {table_name} \
                        --key {table_config['key_col']} \
                        --date {{{{ ds }}}}
                """,
                # 注意：在 f-string 中，双括号 {{{{ }}}} 才能输出 {{ }}
            )

            validate = PythonOperator(
                task_id=f"validate_{table_name}",
                python_callable=validate_table_data,
                op_kwargs={"table_name": table_name},
            )

            extract >> validate

        start >> table_group >> end

warning 动态 DAG 的性能注意事项

DAG 文件会被周期性重新解析，具体周期取决于 Airflow 版本、解析配置和处理负载。如果动态生成逻辑里有数据库查询，每次解析都会查一次数据库，可能造成较大压力。建议把动态配置列表放在 Python 变量或 JSON 配置文件中，避免在 DAG 文件解析时查询数据库。

场景：从数据库读取配置动态生成（正确做法）

import json
import os

# 方案 1：从 JSON 配置文件读取（推荐）
CONFIG_FILE = os.path.join(os.path.dirname(__file__), "table_config.json")

with open(CONFIG_FILE) as f:
    TABLES = json.load(f)

# 方案 2：从 Airflow Variable 读取（也可以，但会在每次解析时查 MetaDB）
# from airflow.models import Variable
# TABLES = Variable.get("tables_to_sync", deserialize_json=True, default_var=[])

SubDAG vs TaskGroup：该用哪个

特性	SubDAG	TaskGroup
Airflow 支持	2.x 支持但不推荐	2.x 推荐
复杂度	高（子 DAG 独立运行）	低（同一 DAG 内分组）
执行与资源开销	子 DAG 有独立运行状态，增加调度和资源管理复杂度；并非需要单独部署 Scheduler	仅组织同一 DAG 内的任务，不创建独立子 DAG Run
可视化	一般	清晰（可折叠）
并行性	复杂	简单

结论：Airflow 2.x 的新开发通常优先用 TaskGroup 组织任务，不建议新建 SubDAG；TaskGroup 不提供独立子 DAG 的运行语义，迁移已有 SubDAG 时需检查调度与资源行为。

小结

场景	推荐方案
顺序依赖	`task_a >> task_b >> task_c`
并行执行	`task_a >> [task_b, task_c]`
多对多依赖	`cross_downstream([a, b], [c, d])`
条件分支	`BranchPythonOperator` + `trigger_rule="none_failed_min_one_success"`
等待其他 DAG	`ExternalTaskSensor` 或 `TriggerDagRunOperator(wait_for_completion=True)`
组织大型 DAG	`TaskGroup`
循环生成任务	Python for 循环 + TaskGroup

下一篇：07-调度策略与时间管理 — 深入理解 cron 表达式、数据区间和时区处理

所属系列：00-Airflow学习路线图 | 上一篇：05-常用Operator详解 | 下一篇：07-调度策略与时间管理

#Airflow #任务依赖 #BranchPythonOperator #TaskGroup #动态DAG #DAG编排

info 前置知识

熟悉 BashOperator 和 PythonOperator 的基本用法，参考常用 Operator 详解
理解 DAG Run 和 Task Instance 的概念，参考 Airflow 核心概念
Python 的 for 循环和函数式编程基础（动态 DAG 需要）

掌握检查

[ ] 能用 >> 语法表达以下三种依赖模式：顺序链、一对多（扇出）、多对一（扇入）
[ ] 能解释 BranchPythonOperator 跳过的 Task 的状态是什么（skipped），以及下游 Task 如何设置才能在分支后正常运行（trigger_rule="none_failed_min_one_success"）
[ ] 能说出 TriggerDagRunOperator 和 ExternalTaskSensor 实现跨 DAG 依赖的核心区别
[ ] 知道动态 DAG（for 循环生成 Task）的性能注意点：为什么不能在 DAG 文件顶层做数据库查询
[ ] 明白为什么 Airflow 2.x 中应该用 TaskGroup 而不是 SubDAG

实践练习

基础题：写一个 DAG，实现以下逻辑：先运行数据质量检查 Task，如果通过就执行数据加载 Task，如果不通过就执行发送告警 Task，两个分支各自结束（不合并）。用 BranchPythonOperator 实现，并用 airflow tasks test 分别验证两条分支。

进阶题：有 10 张 MySQL 表需要每日同步，用 for 循环动态生成 10 个 Task（每个 Task 同步一张表），所有表并行抽取，全部完成后触发一个汇总 Task。要求配置放在 Python 列表中（不查数据库），并用 TaskGroup 让 Web UI 更整洁。