在 google composer - Airflow UI 中删除一个 DAG

Question

我想从 Airflow UI 中删除一个 DAG，它在 GCS/dags 文件夹中不再可用。我知道 Airflow 有一种 "new" 方法可以使用 airflow delete_dag my_dag_id 命令，见于

composer airflow 版本似乎还不支持 delete_dag 命令。

不要尝试这个： 我也试过使用 airflow resetdb 并且气流 UI 死了

有没有办法删除当前不在 gs://BUCKET/dags/ 文件夹中的 dag？

Answer 1

由于 cloud composer 使用的是最新的 stable 版本，即 1.9.0，删除 dag 的功能不可用。

然而，

文档中很少有删除 dag 的说明，如下所示：

 gcloud beta composer environments storage dags delete \
     --environment ENVIRONMENT_NAME \
     --location LOCATION \
     DAG_NAME.py

但不幸的是，这不会从 Airflow 网络界面中删除 DAG。

更多信息：https://cloud.google.com/composer/docs/how-to/using/managing-dags#deleting_a_dag

Answer 2

我创建了一个 DAG 来清理 UI。它从 Airflow Variables

读取 afDagID

from airflow import DAG
from airflow import models
from airflow.operators.mysql_operator import MySqlOperator
import logging
from datetime import datetime
from airflow.operators.dummy_operator import DummyOperator

dag = DAG('ManageAirFlow', description='Deletes Airflow DAGs from backend: Uses vars-  afDagID',
      schedule_interval=None,
      start_date=datetime(2018, 3, 20), catchup=False)

DeleteXComOperator = MySqlOperator(
  task_id='delete-xcom-record-task',
  mysql_conn_id='airflow_db',
  sql="DELETE from xcom where dag_id='{}'".format(models.Variable.get('afDagID')),
  dag=dag)

DeleteTaskOperator = MySqlOperator(
  task_id='delete-task-record-task',
  mysql_conn_id='airflow_db',
  sql="DELETE from task_instance where dag_id='{}'".format(models.Variable.get('afDagID')),
  dag=dag)

DeleteSLAMissOperator = MySqlOperator(
  task_id='delete-sla-record-task',
  mysql_conn_id='airflow_db',
  sql="DELETE from sla_miss where dag_id='{}'".format(models.Variable.get('afDagID')),
  dag=dag)

DeleteLogOperator = MySqlOperator(
  task_id='delete-log-record-task',
  mysql_conn_id='airflow_db',
  sql="DELETE from log where dag_id='{}'".format(models.Variable.get('afDagID')),
  dag=dag)

DeleteJobOperator = MySqlOperator(
  task_id='delete-job-record-task',
  mysql_conn_id='airflow_db',
  sql="DELETE from job where dag_id='{}'".format(models.Variable.get('afDagID')),
  dag=dag)

DeleteDagRunOperator = MySqlOperator(
  task_id='delete-dag_run-record-task',
  mysql_conn_id='airflow_db',
  sql="DELETE from dag_run where dag_id='{}'".format(models.Variable.get('afDagID')),
  dag=dag)

DeleteDagOperator = MySqlOperator(
  task_id='delete-dag-record-task',
  mysql_conn_id='airflow_db',
  sql="DELETE from dag where dag_id='{}'".format(models.Variable.get('afDagID')),
  dag=dag)



DeleteXComOperator >> DeleteTaskOperator >> DeleteSLAMissOperator >> DeleteLogOperator >> DeleteJobOperator >> DeleteDagRunOperator >> DeleteDagOperator

Answer 3

它将分两步进行：- 第 1 步：-

首先你必须使用存储桶中的命令删除airflow_monitoring.py文件。

gcloud composer environments storage dags delete --environment viu-etl-prod-composer --location us-central1 airflow_monitoring.py

第 2 步：-
如图所示点击红叉勾选

在 google composer - Airflow UI 中删除一个 DAG

Delete a DAG in google composer - Airflow UI

airflow

google-cloud-composer