如何在Airflow中为特定任务设置trigger_rule（触发规则）

嘿，这个需求在Airflow里其实很好实现，我给你两种清晰的方案，推荐第一种，因为更直观好维护：

这是最直接的方式，你只需要分别指定task_1和task_2的上游都是task_init，不需要让task_2依赖task_1。这样task_init成功后，task_1和task_2会按调度逻辑启动，彼此的状态互不影响。

示例代码：

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from datetime import datetime

with DAG(
    dag_id="custom_dependency_example",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",
    catchup=False
) as dag:
    # 定义任务
    task_init = DummyOperator(task_id="task_init")
    task_1 = DummyOperator(task_id="task_1")
    task_2 = DummyOperator(task_id="task_2")

    # 设置依赖关系
    task_init >> task_1  # task_1 依赖 task_init 成功
    task_init >> task_2  # task_2 直接依赖 task_init 成功，与 task_1 无关

如果因为某些历史原因，task_2已经被默认关联到task_1（比如原依赖是task_init >> task_1 >> task_2），你可以通过设置trigger_rule来让task_2忽略task_1的状态，只关注task_init是否成功。

这种场景下，你需要把task_2的上游同时设为task_init和task_1，然后给task_2设置trigger_rule="none_failed_min_one_success"——这个规则的意思是：只要上游任务中没有失败的，且至少有一个成功（这里确保task_init成功），task_2就会运行。

示例代码：

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from datetime import datetime
from airflow.utils.trigger_rule import TriggerRule

with DAG(
    dag_id="trigger_rule_example",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",
    catchup=False
) as dag:
    task_init = DummyOperator(task_id="task_init")
    task_1 = DummyOperator(task_id="task_1")
    # 设置trigger_rule，忽略task_1的状态
    task_2 = DummyOperator(
        task_id="task_2",
        trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS
    )

    # 上游同时包含task_init和task_1
    [task_init, task_1] >> task_2

不过还是强烈推荐方案1，因为依赖关系更清晰，后续维护DAG的时候不容易产生误解。

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