如何在Airflow中实现：任务必执行且DAG随前置任务失败而失败

针对你提到的「不管前置任务成败必须终止集群，同时DAG状态要跟随前置任务失败而失败」的需求，可以通过拆分「强制执行终止任务」和「独立检查上游状态」两个逻辑来实现，具体方案如下：

核心思路 #

• 用TriggerRule.ALL_DONE保证终止集群的任务必执行，不受上游任务状态影响
• 新增一个状态检查任务，专门判断核心前置任务（比如提交任务）的状态，通过这个任务的成败来控制整个DAG的最终状态

代码实现示例 #

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.utils.trigger_rule import TriggerRule
from airflow.utils.state import State
from datetime import datetime

def check_upstream_task_status(**context):
    # 获取提交任务的实例状态
    submit_task_instance = context['dag_run'].get_task_instance('submit_task')
    if submit_task_instance.state != State.SUCCESS:
        # 上游任务失败时，主动抛出异常让当前任务失败
        raise RuntimeError(f"核心任务submit_task状态为{submit_task_instance.state}，标记DAG失败")

with DAG(
    dag_id="cluster_manage_dag",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    default_args={"retries": 0}
) as dag:
    # 1. 启动集群任务
    start_cluster = BashOperator(
        task_id="start_cluster",
        bash_command="echo '启动EC2/EMR集群'"
    )

    # 2. 提交业务任务（模拟失败场景）
    submit_task = BashOperator(
        task_id="submit_task",
        bash_command="echo '提交Spark任务'; exit 1"
    )

    # 3. 终止集群任务：无论上游成败必执行
    terminate_cluster = BashOperator(
        task_id="terminate_cluster",
        bash_command="echo '销毁集群资源'",
        trigger_rule=TriggerRule.ALL_DONE,
        depends_on_past=False
    )

    # 4. 状态检查任务：控制DAG最终状态
    check_task_status = PythonOperator(
        task_id="check_upstream_status",
        python_callable=check_upstream_task_status,
        provide_context=True,
        trigger_rule=TriggerRule.ALL_DONE,
        depends_on_past=False
    )

    # 设置依赖关系
    start_cluster >> submit_task
    # 终止集群依赖启动和提交任务
    [start_cluster, submit_task] >> terminate_cluster
    # 状态检查依赖提交任务
    submit_task >> check_task_status

逻辑说明 #

1. 终止集群的可靠性：terminate_cluster任务使用TriggerRule.ALL_DONE，只要start_cluster和submit_task完成（不管成功还是失败），就会立即执行，确保集群资源不会泄漏。
2. DAG状态的准确性：check_upstream_status任务会主动检查核心业务任务submit_task的状态，如果它失败，这个检查任务就会抛出异常并标记为失败，最终整个DAG会因为存在失败任务而被标记为失败；如果submit_task成功，检查任务也会成功，DAG最终状态为成功。
3. 扩展性：如果有多个需要检查的前置任务，可以在检查函数里遍历所有目标任务ID，只要其中一个状态不是SUCCESS就触发失败逻辑。

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