如何终止Airflow中OFF状态DAG的运行任务并清理无效DAG

Hey there, let's tackle your Airflow issues one by one:

Great news—tasks in the "shutdown" state do NOT consume your parallel task slots (the 32 concurrent tasks you've configured).

Airflow only counts tasks actively in the Running state against your parallelism limit. The shutdown state is a terminal state (like failed, success, or skipped), meaning the task has stopped executing and released all allocated resources. The UI might show a lingering "Running tasks" label due to caching delays, but behind the scenes, those tasks aren't blocking new ones from starting.

If you want to wipe these DAGs completely and avoid any lingering UI or metadata clutter, follow these steps:

• Mark DAGs as Deleted

  • First, pause the DAGs in the Airflow UI (toggle the "On/Off" switch to Off).
  • Remove the DAG's Python file from your Airflow dags_folder—Airflow will automatically detect the missing file and mark the DAG as "Deleted" after a scheduler refresh.
  • If you want to keep the file but hide the DAG permanently, add is_paused_upon_creation=True to your DAG definition and set catchup=False to prevent backfills.

• Clean Up Task Instance Metadata
  Use Airflow's CLI tools to purge old or unwanted task instances from the metadata database:

  • Clear all task instances for a specific DAG (including shutdown state ones):

    airflow tasks clear -d <your_dag_id> --skip-all-clear
    

    The --skip-all-clear flag forces cleanup even for terminal state tasks.
  • Clean up all old task instances across all DAGs (specify a timeframe to avoid deleting recent data):

    airflow db clean --older-than 30d
    

• Restart Airflow Components
  If the UI still shows stale states, restart the Airflow Webserver, Scheduler, and Worker (if using CeleryExecutor) services. This refreshes the UI cache and ensures all state updates are synced.

• Manual Database Cleanup (Last Resort)
  If CLI commands don't resolve the issue, you can directly clean the metadata database (always back up first!):
  For PostgreSQL/MySQL, run this query to delete shutdown state tasks for your DAG:

  DELETE FROM task_instance WHERE dag_id = '<your_dag_id>' AND state = 'shutdown';
  

  Then restart Airflow services to reflect the changes.

• Clean Up Celery Worker Processes (If Applicable)
  If you're using CeleryExecutor, check for zombie worker processes that might be holding onto stale task references. Restarting the Celery Worker service will clear these.

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