如何删除Google Composer中Airflow UI里已不在GCS的DAG？

我来给你分享几个亲测有效的方法，避开你踩过的airflow resetdb这种大坑：

方法一：用Airflow Python API自动清理（推荐） #

Cloud Composer支持通过gcloud命令直接在环境中运行Airflow脚本，我们可以写一段代码自动对比GCS里的DAG文件和数据库中的记录，批量删除不存在的DAG。

操作步骤： #

1. 替换下面命令中的YOUR_COMPOSER_ENV_NAME（你的Composer环境名）、YOUR_REGION（环境所在区域，比如us-central1）、YOUR_COMPOSER_BUCKET（Composer对应的GCS桶名，可在环境详情里找到）
2. 直接在终端执行这段命令：

gcloud composer environments run YOUR_COMPOSER_ENV_NAME \
  --location YOUR_REGION \
  bash -- -c "airflow python << 'EOF'
from airflow.models import DagModel
from google.cloud import storage
import os

# 配置你的GCS信息
BUCKET_NAME = 'YOUR_COMPOSER_BUCKET'
DAGS_PREFIX = 'dags/'

# 获取GCS中所有DAG的ID（去掉.py后缀）
storage_client = storage.Client()
bucket = storage_client.bucket(BUCKET_NAME)
blobs = bucket.list_blobs(prefix=DAGS_PREFIX)
gcs_dag_ids = [os.path.splitext(os.path.basename(blob.name))[0] for blob in blobs if blob.name.endswith('.py')]

# 连接Airflow数据库，获取所有已注册的DAG
session = DagModel.session
db_dags = session.query(DagModel).all()

# 批量删除不在GCS中的DAG
for dag in db_dags:
    if dag.dag_id not in gcs_dag_ids:
        print(f"正在删除DAG: {dag.dag_id}")
        session.delete(dag)
session.commit()
EOF"

注意事项： #

• 确保Composer环境的服务账号拥有GCS桶的读取权限，以及Airflow数据库的写入权限（默认配置一般都满足）
• 执行前可以先把session.commit()注释掉，加上print预览要删除的DAG，确认无误再执行删除

方法二：直接操作Cloud SQL数据库（谨慎使用） #

如果上面的方法无法执行，你可以直接连接到Composer对应的Cloud SQL实例，手动执行SQL清理。务必先备份数据库！

操作步骤： #

1. 在Composer环境详情页找到对应的Cloud SQL实例名称
2. 用gcloud命令连接到数据库：

gcloud sql connect YOUR_SQL_INSTANCE_NAME --user=postgres

3. 输入数据库密码（可在Composer环境的“配置”页找到）
4. 执行以下SQL语句，替换'dag1','dag2'为你GCS中实际存在的DAG ID：

-- 删除主表中的无效DAG
DELETE FROM dag WHERE dag_id NOT IN ('dag1', 'dag2', 'dag3');

-- 清理关联表的冗余数据
DELETE FROM dag_run WHERE dag_id NOT IN (SELECT dag_id FROM dag);
DELETE FROM task_instance WHERE dag_id NOT IN (SELECT dag_id FROM dag);
DELETE FROM sla_miss WHERE dag_id NOT IN (SELECT dag_id FROM dag);
DELETE FROM log WHERE dag_id NOT IN (SELECT dag_id FROM dag);

重要提醒 #

绝对不要使用airflow resetdb命令，这个操作会清空Airflow的整个数据库，直接导致UI和所有调度任务崩溃，正如你之前遇到的问题。

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