Удалить DAG в Google Composer - Airflow UI

Question

Я хочу удалить группу DAG из интерфейса Airflow, которая больше не доступна в папке GCS/dags.Я знаю, что у Airflow есть «новый» способ удаления меток из БД с помощью команды airflow delete_dag my_dag_id, замеченный в https://stackoverflow.com/a/49683543/5318634

Кажется, что в версии воздушного потока композитора команда delete_dag еще не поддерживается.

Не пытайтесь сделать это: Я также пытался использовать airflow resetdb, и пользовательский интерфейс воздушного потока умер

Есть ли способ удалить ярлыки, которые в данный момент отсутствуютпапка gs://BUCKET/dags/?

Answer 1

Я создал DAG для очистки пользовательского интерфейса.Он читает 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 2

Поскольку облачный композитор использует последнюю версию stable , т.е. 1.9.0, функция удаления метки недоступна.

Однако,

В документах есть несколько инструкций по удалению метки, как показано ниже:

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

, но, к сожалению, это не приведет к удалению DAG из сети Airflow.интерфейс.

Дополнительная информация: https://cloud.google.com/composer/docs/how-to/using/managing-dags#deleting_a_dag