Airflow配置条件依赖实现任务成功/失败分支正确触发

现有配置问题点 #

• 依赖方向写反：代码末尾log_conf_3 >> ([log_conf_1, log_conf_2])的写法中，>>左侧为上游、右侧为下游，这行代码会把print-conf-3设为print-conf-1、print-conf-2的上游，和预期依赖关系完全颠倒。
• 失败分支触发规则配错：print-conf-failure当前设置的trigger_rule=TriggerRule.ALL_SUCCESS要求所有上游全部成功才会运行，无法实现「任意上游失败就执行」的逻辑。
• 冗余无效配置：代码里定义的print-conf-3不在描述的业务流程范围内，属于多余内容可直接删除。另外原配置给print-conf-failure绑定的是成功日志回调函数print_log，就算触发也不会输出失败日志，需要同步修正。

正确配置步骤 #

1. print-conf-success保持TriggerRule.ALL_SUCCESS触发规则不变，确保两个上游全部成功才会运行。
2. 把print-conf-failure的触发规则改为TriggerRule.ONE_FAILED，只要任意一个上游任务失败，该任务就会触发。
3. 删除错误的反向依赖代码，将print-conf-1、print-conf-2设置为两个分支任务的共同上游即可。

修正后的完整可运行代码：

"""Example DAG demonstrating the usage of the PythonOperator."""
from datetime import datetime
from airflow.utils.trigger_rule import TriggerRule
from airflow import DAG
from airflow.operators.python import PythonOperator

DEFAULT_ARGS = {
    'owner': 'admin',
    'depends_on_past': False,
    'start_date': datetime(2022, 5, 20, 0),
    'retries': 2
}


def print_log(**kwargs):
    print("--------------------")
    print("1, 2, 3")
    print("--------------------")


def print_log_failed(**kwargs):
    print("--------------------")
    print("1, 2, 3, failed")
    print("--------------------")


with DAG(dag_id="test_dag", schedule_interval=None, default_args=DEFAULT_ARGS, max_active_runs=10) as dag:
    # 全成功分支
    log_conf = PythonOperator(
        task_id='print-conf-success',
        provide_context=True,
        python_callable=print_log,
        trigger_rule=TriggerRule.ALL_SUCCESS
    )

    # 失败分支
    log_conf_failure = PythonOperator(
        task_id='print-conf-failure',
        provide_context=True,
        python_callable=print_log_failed,
        trigger_rule=TriggerRule.ONE_FAILED
    )

    # 两个前置任务
    log_conf_1 = PythonOperator(
        task_id='print-conf-1',
        provide_context=True,
        python_callable=print_log,
        trigger_rule=TriggerRule.ALL_SUCCESS
    )

    log_conf_2 = PythonOperator(
        task_id='print-conf-2',
        provide_context=True,
        python_callable=print_log,
        trigger_rule=TriggerRule.ALL_SUCCESS
    )

    # 绑定依赖：两个前置任务执行完后，按触发规则走对应分支
    [log_conf_1, log_conf_2] >> [log_conf, log_conf_failure]

配置生效后执行逻辑：print-conf-1和print-conf-2并行执行，两个全部成功时触发print-conf-success，print-conf-failure自动跳过；任意一个任务执行失败时触发print-conf-failure，print-conf-success自动跳过，完全匹配需求。

原错误配置生成的DAG流程：
错误DAG依赖图

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