Apache Airflow中使用TaskFlowAPI实现任务分支的正确方法

你写的代码不生效的核心原因是：DAG顶层的Python代码是在DAG解析阶段执行的，不是在DAG运行时执行的。
代码里的outcome_1是TaskFlow返回的XComArg占位对象，不是task_1_returns_boolean运行后实际返回的布尔值。Python在解析DAG时判断if outcome_1，只要这个对象非空就会判定为真，而且这个判断逻辑完全不会被注册到DAG的运行链路里，自然不可能根据task1的实际运行结果动态选择分支。

必须使用@task.branch装饰器定义专门的分支任务，分支逻辑要写在这个分支任务内部，它会在运行时拿到上游任务的实际返回值，返回下一个要执行的任务ID（或任务ID列表），Airflow会自动将未被选中的下游任务标记为跳过状态。

修正后的可运行代码示例：

from airflow.decorators import dag, task
from airflow.utils.trigger_rule import TriggerRule
import pendulum

@dag(
    schedule_interval=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=['test'],
)
def my_dag():
    @task()
    def task_1_returns_boolean():
        # 替换为实际判断逻辑，返回布尔值
        boolean_value = True
        return boolean_value

    # 定义分支任务
    @task.branch()
    def branch_task(upstream_result: bool):
        if upstream_result:
            # 返回真值分支要执行的任务task_id
            return "task_2_execute_if_true"
        # 假值分支如果不需要执行任何任务，直接return None即可，无需定义对应任务
        return "task_3_execute_if_false"

    @task()
    def task_2_execute_if_true():
        print("执行真值分支逻辑")

    @task()
    def task_3_execute_if_false():
        print("执行假值分支逻辑")

    # 分支后所有路径都要执行的汇合任务，必须修改trigger_rule，默认all_success会因上游有跳过任务无法触发
    @task(trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS)
    def final_join_task():
        print("分支汇合，执行收尾逻辑")

    # 配置任务依赖链路
    outcome_1 = task_1_returns_boolean()
    branch_result = branch_task(outcome_1)
    task2_res = task_2_execute_if_true()
    task3_res = task_3_execute_if_false()
    join_res = final_join_task()

    branch_result >> [task2_res, task3_res] >> join_res

executed = my_dag()

• 分支判断逻辑必须写在@task.branch装饰的任务内部，不能直接写在DAG顶层的解析代码里，这个专门的分支任务是必须的。
• 分支任务的返回值必须是下游任务的task_id（字符串格式），如果要同时执行多个下游任务就返回task_id组成的列表。
• 如果分支后存在所有分支都要汇合执行的公共任务，必须修改这个公共任务的trigger_rule，默认的all_success规则会因为上游有被跳过的任务直接判定为失败/跳过。
• 如果某个分支不需要执行任何后续任务，分支任务直接返回None即可。

内容的提问来源于stack exchange，提问作者matwasilewski