Airflow BranchPythonOperator返回False时后续任务被跳过的解决方法

Airflow的BranchPythonOperator自带分支跳过逻辑：任务执行后，会将所有不在返回值列表里的直接下游任务标记为skipped状态。而Airflow默认的任务触发规则是all_success，该规则要求任务的所有直接上游必须全部执行成功才能运行，只要有一个上游是skipped状态，当前任务就会被级联标记为skipped，最终传导到后续所有任务。
异常运行的DAG状态如下图所示：

你的代码存在两个配置缺失：

• 未给send_email_notification_task设置下游指向join_task，分支走发邮件逻辑时没有汇合路径
• join_task使用默认的all_success触发规则，只要某一条分支路径被跳过，join_task就会被连带跳过，导致后续所有offload任务无法执行

分两步调整配置即可：

1. 调整join_task的触发规则，替换为none_failed_min_one_success（Airflow 2.2+版本支持，低版本可使用none_failed替代）。该规则的判定逻辑是：所有上游任务没有失败状态，且至少有一个上游任务执行成功，当前任务就可以正常运行，不会因为部分分支被跳过而阻断流程。
2. 补全分支汇合依赖，将send_email_notification_task的下游指向join_task，确保发邮件分支执行完成后能回到主流程。

首先导入触发规则枚举：

from airflow.utils.trigger_rule import TriggerRule

修改join_task的定义：

join_task = DummyOperator(
    task_id='join_task',
    trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,
    dag=dag
)

补全分支依赖，建议替换原有零散的set_upstream/set_downstream写法，用位运算符写法更清晰：

# 分支起点分出两条路径
will_send_email_task >> [send_email_notification_task, join_task]
# 发邮件分支汇合到join_task
send_email_notification_task >> join_task

生成offload任务的逻辑可以保留，也可以统一收集任务对象后批量设置依赖：

offload_task_list = []
for table, val in some_dict.items():
    offload_task = PythonOperator(
        task_id = f"offload_{table}_task",
        dag=dag,
        provide_context=True,
        python_callable=some_python_callable,
        op_kwargs={'table_name': table}
    )
    offload_task_list.append(offload_task)

# join_task执行完成后运行所有offload任务，最终流转到end_task
join_task >> offload_task_list >> end_task

修改后两种分支场景均符合预期：

• 当需要发送邮件（len(message) > 0）时：will_send_email_task选中send_email_notification_task，join_task从will_send_email_task直接过来的路径被标记为跳过，等send_email_notification_task执行成功后，join_task满足触发规则正常运行，后续所有offload任务正常执行
• 当不需要发送邮件（len(message) == 0）时：will_send_email_task选中join_task，send_email_notification_task被标记为跳过，join_task收到will_send_email_task的成功状态，满足触发规则正常运行，仅跳过邮件发送任务，后续所有offload任务正常执行

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