Airflow中如何用BranchPythonOperator返回含并行任务的链式序列？

你的代码核心问题在于并行任务（如task_1a、task_1b等）同时依赖option1和option2两个分支节点。当BranchPythonOperator选中其中一个分支（比如option2）时，未被选中的option1会被标记为skipped状态。而Airflow任务默认的trigger_rule是all_success，这意味着并行任务需要等待所有上游任务成功才能执行，但此时option1处于skipped状态，导致并行任务无法触发，最终整个下游链路被跳过。

另外，你的end_task连接了两组chain的输出，同样会因为部分上游任务skipped而无法执行（即使设置all_success也没用，因为skipped不等于success）。

针对这个问题，有两种可行的修正思路：

方案1：为不同分支创建独立的任务组 #

如果业务允许，为每个分支单独定义一套并行任务，避免任务复用导致的依赖冲突。这样每个分支的下游任务只依赖对应分支的起始节点，不会受另一个分支skipped状态的影响。

修正后的代码示例：

from airflow import DAG
from airflow.operators.python import BranchPythonOperator
from airflow.operators.dummy import DummyOperator
from airflow.utils.helpers import chain
from datetime import datetime

default_args = {
    'start_date': datetime(2023, 1, 1),
}

def _choose_best_model():
    value = 6
    if value > 10:
        return 'option1'
    else:
         return 'option2'

with DAG('branching', schedule_interval='@daily', default_args=default_args, catchup=False) as dag:
    choose_best_model = BranchPythonOperator(
        task_id='choose_best_model',
        python_callable=_choose_best_model
    )
    option1 = DummyOperator(task_id='option1')
    option2 = DummyOperator(task_id='option2')

    # Option1专属并行任务
    task_1a_opt1 = DummyOperator(task_id='task_1a_opt1')
    task_2a_opt1 = DummyOperator(task_id='task_2a_opt1')
    task_3a_opt1 = DummyOperator(task_id='task_3a_opt1')
    task_1b_opt1 = DummyOperator(task_id='task_1b_opt1')
    task_2b_opt1 = DummyOperator(task_id='task_2b_opt1')
    task_3b_opt1 = DummyOperator(task_id='task_3b_opt1')

    # Option2专属并行任务
    task_1a_opt2 = DummyOperator(task_id='task_1a_opt2')
    task_2a_opt2 = DummyOperator(task_id='task_2a_opt2')
    task_1b_opt2 = DummyOperator(task_id='task_1b_opt2')
    task_2b_opt2 = DummyOperator(task_id='task_2b_opt2')

    end_task = DummyOperator(
        task_id='end_task',
        trigger_rule='none_failed_min_one_success'  # 只要有一条分支成功就执行
    )

    choose_best_model >> [option1, option2]

    # 连接Option1的链路
    chain(option1, [task_1a_opt1, task_2a_opt1, task_3a_opt1], [task_1b_opt1, task_2b_opt1, task_3b_opt1], end_task)
    # 连接Option2的链路
    chain(option2, [task_1a_opt2, task_2a_opt2], [task_1b_opt2, task_2b_opt2], end_task)

方案2：修改共享任务的触发规则 #

如果必须复用并行任务，需要修改这些共享任务的trigger_rule，使其在部分上游skipped的情况下仍能执行。适合的触发规则是none_failed_min_one_success（只要没有失败的上游，且至少有一个上游成功），或者one_success（只要有一个上游成功即可）。

同时，end_task的触发规则也需要调整为none_failed_min_one_success，确保只要有一条分支执行完成就能触发。

修正后的代码示例：

from airflow import DAG
from airflow.operators.python import BranchPythonOperator
from airflow.operators.dummy import DummyOperator
from airflow.utils.helpers import chain
from datetime import datetime
from airflow.utils.trigger_rule import TriggerRule

default_args = {
    'start_date': datetime(2023, 1, 1),
}

def _choose_best_model():
    value = 6
    if value > 10:
        return 'option1'
    else:
         return 'option2'

with DAG('branching', schedule_interval='@daily', default_args=default_args, catchup=False) as dag:
    choose_best_model = BranchPythonOperator(
        task_id='choose_best_model',
        python_callable=_choose_best_model
    )
    option1 = DummyOperator(task_id='option1')
    option2 = DummyOperator(task_id='option2')

    # 共享并行任务，修改触发规则
    task_1a = DummyOperator(
        task_id='task_1a',
        trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS
    )
    task_2a = DummyOperator(
        task_id='task_2a',
        trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS
    )
    task_3a = DummyOperator(
        task_id='task_3a',
        trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS
    )

    task_1b = DummyOperator(
        task_id='task_1b',
        trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS
    )
    task_2b = DummyOperator(
        task_id='task_2b',
        trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS
    )
    task_3b = DummyOperator(
        task_id='task_3b',
        trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS
    )

    end_task = DummyOperator(
        task_id='end_task',
        trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS
    )

    choose_best_model >> [option1, option2]

    chain(option1, [task_1a, task_2a, task_3a], [task_1b, task_2b, task_3b], end_task)
    chain(option2, [task_1a, task_2a], [task_1b, task_2b], end_task)

• BranchPythonOperator的行为是：只有返回的任务ID会被标记为success，其他分支任务会被标记为skipped，这是正常的设计。
• 触发规则的选择：NONE_FAILED_MIN_ONE_SUCCESS是最安全的，它要求所有上游任务要么成功要么被跳过，且至少有一个成功；ONE_SUCCESS则只要有一个上游成功就执行，适合对上游skipped不敏感的场景。

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