气流：如何删除 DAG？

Question

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

如何删除正在运行并在 Web GUI 中显示的特定 DAG？是否有 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 中拥有此功能！

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

核心 delete_dag(...) 代码现在是实验 API 的一部分，并且有可用的入口点 via the CLI 和 via the REST API。

命令行界面：

airflow delete_dag my_dag_id

REST API（在本地运行网络服务器）：

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

---

关于 REST API 的警告：确保您的 Airflow 集群 uses authentication 在生产中。

安装/升级到 Airflow 1.10（当前）

要升级，请运行：

export SLUGIFY_USES_TEXT_UNIDECODE=yes

或：

export AIRFLOW_GPL_UNIDECODE=yes

然后：

pip install -U apache-airflow

记得先查看UPDATING.md了解完整详情！

Answer 2

这是我使用带有默认 connection_id 的 PostgresHook 改编的代码。

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

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

Answer 5

我刚刚写了一个脚本来删除与特定 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 6

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

查看此票以了解更多详情：

[AIRFLOW-2657] 添加从 web ui 中删除 DAG 的功能

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

Answer 7

我编写了一个脚本，用于删除与默认 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 8

对于直接访问airflow db的Postgrespsql控制台的用户，可以简单地执行以下请求来移除DAG：

\set dag_id YOUR_DAG_ID

delete from xcom where dag_id=:'dag_id';
delete from task_instance where dag_id=:'dag_id';
delete from sla_miss where dag_id=:'dag_id';
delete from log where dag_id=:'dag_id';
delete from job where dag_id=:'dag_id';
delete from dag_run where dag_id=:'dag_id';
delete from dag where dag_id=:'dag_id';

类似的（稍作改动）查询适用于其他数据库，例如 MySQL 和 SQLite。

Answer 9

Airflow 中没有任何内置功能可以为您做到这一点。要删除 DAG，请将其从存储库中删除并删除 Airflow Metastore 表 - dag 中的数据库条目。

Answer 10

您可以清除一组任务实例，就好像它们从未运行过一样：

airflow clear dag_id -s 2017-1-23 -e 2017-8-31

然后从 dags 文件夹中删除 dag 文件

Answer 11

根据@OlegYamin 的回答，我正在执行以下操作来删除由postgres 支持的dag，其中airflow 使用public 架构。

delete from public.dag_pickle where id = (
    select pickle_id from public.dag where dag_id = 'my_dag_id'
);
delete from public.dag_run where dag_id = 'my_dag_id';
delete from public.dag_stats where dag_id = 'my_dag_id';
delete from public.log where dag_id = 'my_dag_id';
delete from public.sla_miss where dag_id = 'my_dag_id';
delete from public.task_fail where dag_id = 'my_dag_id';
delete from public.task_instance where dag_id = 'my_dag_id';
delete from public.xcom where dag_id = 'my_dag_id';
delete from public.dag where dag_id = 'my_dag_id';

警告：我不知道第一个删除查询的效果/正确性。这只是假设需要它。

Answer 12

只需从 mysql 中删除它，对我来说效果很好。从下表中删除它们：

• dag

• dag_constructor

• dag_group_ship
• dag_pickle
• dag_run
• dag_stats

（未来版本中可能会有更多表格） 然后重启 webserver 和 worker。

Answer 13

版本 >= 1.10.0：

我有气流版本 1.10.2，我尝试执行气流 delete_dag 命令，但该命令引发以下错误：

bash-4.2# 气流 delete_dag dag_id

[2019-03-16 15:37:20,804] {settings.py:174} INFO - settings.configure_orm()：使用池设置。 pool_size=5，pool_recycle=1800，pid=28224 /usr/lib64/python2.7/site-packages/psycopg2/init.py:144：用户警告：psycopg2 轮子包将从 2.8 版重命名；为了继续从二进制安装，请改用“pip install psycopg2-binary”。详情见：http://initd.org/psycopg/docs/install.html#binary-install-from-pypi。 """) 这将删除与指定 DAG 相关的所有现有记录。继续？ (y/n)y 回溯（最近一次通话最后）： 文件“/usr/bin/airflow”，第 32 行，在 args.func(args) 包装器中的文件“/usr/lib/python2.7/site-packages/airflow/utils/cli.py”，第 74 行 返回 f(*args, **kwargs) 文件“/usr/lib/python2.7/site-packages/airflow/bin/cli.py”，第 258 行，在 delete_dag 引发 AirflowException（错误） airflow.exceptions.AirflowException：服务器错误

虽然我可以通过 Curl 命令删除。 请让我知道是否有人知道此命令的执行，这是已知的还是我做错了什么。

版本

没有删除dag的命令，所以需要先删除dag文件，然后从airflow元数据库中删除所有对dag_id的引用。

警告

您可以重置气流元数据库，您将删除所有内容，包括 dag，但请记住，您还将删除历史记录、池、变量等。

airflow resetdb 然后airflow initdb

Answer 14

第一 --> 从 $AIRFLOW_HOME/dags 文件夹中删除 DAG 文件。 注意：根据您是否使用过子目录，您可能需要翻遍子目录才能找到 DAG 文件并将其删除。

第二--> 使用删除按钮（圆圈中的 x）从 Web 服务器 UI 中删除 DAG

Answer 15

在新的气流版本中，UI 中有一个删除 dag（红色 x）按钮，位于 DAG 旁边

Answer 16

我遇到了鸡/蛋问题，我点击了正在运行的 DAG 实例的绿色小圆圈，它让您“标记为失败”等。我点击了“删除”，它不再卡住了。

Answer 17

如果您使用 Docker 运行 Airflow，您可以使用 DAG 中的 BashOperator 删除另一个 DAG：

t1 = BashOperator(task_id='delete_dag_task', bash_command=f'airflow dags delete -y {dag_id}')

dag_id 是 dag 的名称。这使用标准 CLI 命令，而不是自己从元数据库中删除记录。您还需要使用 PythonOperator 从 dags 目录中删除 DAG 文件。

我有这样一个 DAG：

from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.operators.python_operator import PythonOperator
from airflow.operators.bash import BashOperator
import os

# These args will get passed on to each operator
# You can override them on a per-task basis during operator initialization
default_args = {
    'start_date': days_ago(1),
    'owner': 'airflow',
    'retries': 1
}


def delete_dag(**context):
    conf = context["dag_run"].conf
    dag_id = conf["dag_name"]
    t1 = BashOperator(task_id='delete_dag_task', bash_command=f'airflow dags delete -y {dag_id}')
    t1.execute(context=context)


def delete_dag_file(**context):
    conf = context["dag_run"].conf
    dag_id = conf["dag_name"]
    script_dir = os.path.dirname(__file__)
    dag_file_path = os.path.join(script_dir, '{}.py'.format(dag_id))
    try:
        os.remove(dag_file_path)
    except OSError:
        pass


with DAG('dag-deleter',
         schedule_interval=None,
         default_args=default_args,
         is_paused_upon_creation=False,
         catchup=False) as dag:

    delete_dag = PythonOperator(
        task_id="delete_dag",
        python_callable=delete_dag,
        provide_context=True)

    delete_dag_file = PythonOperator(
        task_id="delete_dag_file",
        python_callable=delete_dag_file,
        provide_context=True
    )

    delete_dag >> delete_dag_file

我使用 REST API 触发 DAG，在 http 请求中传递以下负载：

{"conf": {"dag_name": "my_dag_name"} }

Answer 18

从 dags 文件夹中删除 dag（您要删除）并运行 airflow resetdb。

或者，您可以进入 airflow_db 并手动从 dag 表中删除这些条目（task_fail、xcom、task_instance、sla_miss、log、job、dag_run、dag、dag_stats）。

Answer 19

对于那些仍在寻找答案的人。在 Airflow 版本 1.8 上，删除 DAG 非常困难，您可以参考上面的答案。但是自从 1.9 已经发布，你只需要

删除 dags 文件夹中的 dag 并重启 webserver