Airflow 2.2中如何让任务X失败后DAG不阻塞且下游视为其成功？

方案1：强制将失败任务标记为成功（完全匹配需求） #

给任务X添加失败回调逻辑，在任务执行失败时自动把它的状态改为成功。这样下游任务会像任务X正常成功一样执行，DAG整体状态也不会被标记为失败。

示例代码：

from airflow.models import TaskInstance
from airflow.utils.state import State
from airflow.utils.context import Context

def force_success_on_failure(context: Context):
    ti: TaskInstance = context["task_instance"]
    # 记录任务实际失败的日志，方便后续排查问题
    ti.log.error(f"任务 {ti.task_id} 执行失败，已强制标记为成功")
    # 修改任务状态为成功
    ti.set_state(State.SUCCESS)

# 定义你的任务X
task_x = YourOperator(
    task_id="task_x",
    on_failure_callback=force_success_on_failure,
    # 其他业务参数
    ...
)

# 下游任务直接依赖task_x即可，无需额外修改触发器规则
task_x >> downstream_task_1 >> downstream_task_2

注意：这个方法会让Airflow UI里任务X显示为成功状态，所以一定要在回调里记录失败日志，避免后续排查时混淆任务实际执行情况。

方案2：用ALL_DONE触发器规则让下游强制执行 #

如果不需要修改任务X的状态，只是希望下游任务不管X成功还是失败都能执行，可以给下游任务设置trigger_rule="all_done"。

示例代码：

from airflow.utils.trigger_rule import TriggerRule

# 定义任务X
task_x = YourOperator(
    task_id="task_x",
    ...
)

# 下游任务设置触发器规则为ALL_DONE
downstream_task = YourDownstreamOperator(
    task_id="downstream_task",
    trigger_rule=TriggerRule.ALL_DONE,
    ...
)

task_x >> downstream_task

注意：此方法下，如果任务X失败，DAG整体状态会被标记为失败，但下游任务仍会正常执行。如果需要DAG状态保持成功，需要搭配方案1使用。

方案3：用ShortCircuitOperator做中转（权宜方案） #

正如你提到的ShortCircuitOperator，可以把它作为任务X和下游任务的中转节点，设置触发器规则为all_done，并让它始终返回True，确保下游任务执行。

示例代码：

from airflow.operators.short_circuit import ShortCircuitOperator
from airflow.utils.trigger_rule import TriggerRule
from airflow.utils.context import Context

def always_allow_downstream(**context):
    ti = context["task_instance"]
    upstream_ti = ti.get_upstream_task_instances()[0]
    if upstream_ti.state == State.FAILED:
        ti.log.warning(f"上游任务 {upstream_ti.task_id} 失败，仍允许下游执行")
    return True

short_circuit_node = ShortCircuitOperator(
    task_id="short_circuit_after_x",
    python_callable=always_allow_downstream,
    trigger_rule=TriggerRule.ALL_DONE,
    provide_context=True,
)

task_x >> short_circuit_node >> downstream_task

这个方法和方案2逻辑类似，只是多了一个中转节点，好处是可以在中转节点里添加自定义日志或额外逻辑，但同样不会改变任务X的失败状态，DAG整体仍会显示失败。

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