Airflow DAG执行stop_dag后仍运行中间任务，如何终止无依赖任务

方案1：通过PythonOperator直接操作任务实例状态 #

在stop_dag任务中编写Python函数，直接查询当前DAG运行中仍在执行或等待的任务实例，将其状态标记为终止（如killed或failed）。示例代码如下：

from airflow.models import TaskInstance, DagRun
from airflow.utils.session import create_session
from airflow.operators.python import PythonOperator

def terminate_remaining_tasks(**context):
    # 获取当前DAG运行实例
    dag_run = context["dag_run"]
    with create_session() as session:
        # 筛选出当前DAG运行中，状态为running/queued且不是stop_dag的任务实例
        target_tasks = session.query(TaskInstance).filter(
            TaskInstance.dag_id == dag_run.dag_id,
            TaskInstance.run_id == dag_run.run_id,
            TaskInstance.state.in_(["running", "queued"]),
            TaskInstance.task_id != "stop_dag"
        ).all()

        # 逐个终止任务
        for task in target_tasks:
            task.set_state("killed", session=session)
            session.commit()

# 定义stop_dag任务
stop_dag = PythonOperator(
    task_id="stop_dag",
    python_callable=terminate_remaining_tasks,
    provide_context=True,
    dag=your_dag_object
)

方案2：调用Airflow CLI命令终止任务 #

如果允许在任务中执行系统命令，也可以通过subprocess调用Airflow的CLI命令来清理当前DAG运行的任务：

import subprocess
from airflow.operators.python import PythonOperator

def stop_all_tasks(**context):
    dag_id = context["dag"].dag_id
    run_id = context["dag_run"].run_id
    # 调用CLI命令终止当前DAG运行的所有任务
    subprocess.run(
        ["airflow", "tasks", "clear", dag_id, "--run-id", run_id, "--exclude-task-ids", "stop_dag", "--yes"],
        check=True
    )

stop_dag = PythonOperator(
    task_id="stop_dag",
    python_callable=stop_all_tasks,
    provide_context=True,
    dag=your_dag_object
)

关键注意事项 #

• 执行stop_dag任务的Worker需要具备修改Airflow元数据的权限（方案1）或执行CLI命令的权限（方案2）
• 选择任务终止状态时，killed会直接终止运行中的任务，failed会标记任务失败，可根据业务需求选择
• 若使用Airflow 2.x版本，确保代码适配对应的API和模型结构

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