Airflow:如何删除DAG？

Question

我已经启动了Airflow网络服务器并安排了一些dags.我可以在Web GUI上看到这些dags.

如何删除特定DAG在Web GUI中运行和显示？是否有Airflow CLI命令来执行此操作？

我环顾四周,但是一旦加载和安排DAG,就找不到简单的删除DAG的方法.

Answer 1

编辑8/27/18 - Airflow 1.10现已在PyPI上发布!

https://pypi.org/project/apache-airflow/1.10.0/

如何完全删除DAG

我们现在在Airflow≥1.10中有此功能!

向Airflow添加DAG删除的PR #2199(Jira:AIRFLOW-1002)现已合并,允许从所有相关表中完全删除DAG的条目.

核心delete_dag(...)代码现在是实验API的一部分,并且通过CLI和REST API提供了入口点.

CLI:

airflow delete_dag my_dag_id

REST API(在本地运行Web服务器):

curl -X "DELETE" http://127.0.0.1:8080/api/experimental/dags/my_dag_id

有关REST API的警告:确保您的Airflow集群在生产中使用身份验证.

安装/升级到Airflow 1.10(当前)

要升级,请运行:

export SLUGIFY_USES_TEXT_UNIDECODE=yes

要么:

export AIRFLOW_GPL_UNIDECODE=yes

然后:

pip install -U apache-airflow

请记得首先查看UPDATING.md以获取完整的详细信息!

Answer 2

这是我使用PostgresHook和默认connection_id的改编代码.

import sys
from airflow.hooks.postgres_hook import PostgresHook

dag_input = sys.argv[1]
hook=PostgresHook( postgres_conn_id= "airflow_db")

for t in ["xcom", "task_instance", "sla_miss", "log", "job", "dag_run", "dag" ]:
    sql="delete from {} where dag_id='{}'".format(t, dag_input)
    hook.run(sql, True)

Answer 3

不确定为什么Apache Airflow没有一种明显而简单的方法来删除DAG

提起https://issues.apache.org/jira/browse/AIRFLOW-1002

Answer 4

我刚写了一个脚本来删除与特定dag相关的所有内容,但这只适用于MySQL.如果使用PostgreSQL,可以编写不同的连接器方法.最初由Lance在https://groups.google.com/forum/#!topic/airbnb_airflow/GVsNsUxPRC0上发布的命令 我只是把它放在脚本中.希望这可以帮助.格式:python script.py dag_id

import sys
import MySQLdb

dag_input = sys.argv[1]

query = {'delete from xcom where dag_id = "' + dag_input + '"',
        'delete from task_instance where dag_id = "' + dag_input + '"',
        'delete from sla_miss where dag_id = "' + dag_input + '"',
        'delete from log where dag_id = "' + dag_input + '"',
        'delete from job where dag_id = "' + dag_input + '"',
        'delete from dag_run where dag_id = "' + dag_input + '"',
        'delete from dag where dag_id = "' + dag_input + '"' }

def connect(query):
        db = MySQLdb.connect(host="hostname", user="username", passwd="password", db="database")
        cur = db.cursor()
        cur.execute(query)
        db.commit()
        db.close()
        return

for value in query:
        print value
        connect(value)

Answer 5

可以在Airflow 1.10中删除DAG-s，但是操作的过程和顺序必须正确。有一个“鸡蛋和鸡肉问题”-如果在文件仍然存在的情况下从前端删除DAG，则会重新加载DAG（因为未删除文件）。如果先删除文件并刷新页面，则无法再从Web gui中删除DAG。因此，让我从前端删除DAG的操作序列为：

1. 删除DAG文件（在我的情况下，从管道存储库中删除并部署到气流服务器，尤其是调度程序）
2. 不要刷新Web GUI。
3. 在DAGs视图（常规首页）的Web GUI中，单击“删除dag”-> 最右边的红色图标。
4. 它将从数据库中清除此DAG的所有剩余内容。

Answer 6

我编写了一个脚本,删除与默认SQLite DB的特定dag相关的所有元数据.这是基于耶稣上面的回答,但改编自Postgres到SQLite.用户应设置../airflow.db相对于默认airflow.db文件(通常~/airflow)存储script.py的位置.要执行,请使用python script.py dag_id.

import sqlite3
import sys

conn = sqlite3.connect('../airflow.db')
c = conn.cursor()

dag_input = sys.argv[1]

for t in ["xcom", "task_instance", "sla_miss", "log", "job", "dag_run", "dag" ]:
    query = "delete from {} where dag_id='{}'".format(t, dag_input)
    c.execute(query)

conn.commit()
conn.close()

Answer 7

Airflow 1.10.1已发布。从文件系统删除了相应的DAG之后，此版本增加了从Web UI删除DAG的功能。

有关更多详细信息，请参见此票证：

[AIRFLOW-2657]添加了从Web ui删除DAG的功能

请注意，这实际上并没有从文件系统中删除DAG，您将需要首先手动执行此操作，否则DAG将被重新加载。