Luftstrom: Wie lösche ich eine DAG?

Ich habe den Airflow-Webserver gestartet und einige Dags geplant. Ich kann die Dags auf der Web-GUI sehen.

Wie kann ich eine bestimmte DAG aus der Ausführung löschen und in der Web-GUI anzeigen? Gibt es dafür einen Airflow CLI-Befehl?

Ich habe mich umgesehen, konnte aber keine Antwort auf eine einfache Möglichkeit finden, eine DAG zu löschen, sobald sie geladen und geplant wurde.

Edit 27.08.18 - Airflow 1.10 ist jetzt auf PyPI veröffentlicht!

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

So löschen Sie eine DAG vollständig

Wir haben diese Funktion jetzt in Airflow ≥ 1.10!

Der PR # 2199 (Jira: AIRFLOW-1002 ), der die DAG-Entfernung zu Airflow hinzufügt, wurde nun zusammengeführt, wodurch die Einträge einer DAG vollständig aus allen zugehörigen Tabellen gelöscht werden können.

Der Kerncode delete_dag (...) ist jetzt Teil der experimentellen API, und es sind Einstiegspunkte über die CLI und auch über die REST-API verfügbar .

CLI:

airflow delete_dag my_dag_id

REST-API (Webserver lokal ausführen):

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

Warnung bezüglich der REST-API : Stellen Sie sicher, dass Ihr Airflow-Cluster die Authentifizierung in der Produktion verwendet.

Installation / Upgrade auf Airflow 1.10 (aktuell)

Führen Sie zum Upgrade Folgendes aus:

export SLUGIFY_USES_TEXT_UNIDECODE=yes

oder:

export AIRFLOW_GPL_UNIDECODE=yes

Dann:

pip install -U apache-airflow

Denken Sie daran, zuerst UPDATING.md für die vollständigen Details zu überprüfen !

Dies ist mein angepasster Code mit PostgresHook mit der Standard-Verbindungs-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)

Ich bin mir nicht sicher, warum Apache Airflow keine offensichtliche und einfache Möglichkeit zum Löschen einer DAG bietet

Abgelegt unter https://issues.apache.org/jira/browse/AIRFLOW-1002

Ich habe gerade ein Skript geschrieben, das alles löscht, was mit einem bestimmten Tag zu tun hat, aber dies ist nur für MySQL. Sie können eine andere Connector-Methode schreiben, wenn Sie PostgreSQL verwenden. Ursprünglich wurden die Befehle von Lance auf https://groups.google.com/forum/#!topic/airbnb_airflow/GVsNsUxPRC0 veröffentlicht. Ich habe sie einfach in ein Skript eingefügt. Hoffe das hilft. Format: 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)

DAG-s können in Airflow 1.10 gelöscht werden, aber der Prozess und die Reihenfolge der Aktionen müssen stimmen. Es gibt ein "Ei-Huhn-Problem" - wenn Sie die DAG aus dem Frontend löschen, während die Datei noch vorhanden ist, wird die DAG neu geladen (da die Datei nicht gelöscht wird). Wenn Sie die Datei zuerst löschen und die Seite aktualisieren, kann DAG nicht mehr aus der Web-GUI gelöscht werden. Die Reihenfolge der Aktionen, mit denen ich eine DAG aus dem Frontend löschen konnte, war also:

1. Löschen Sie die DAG-Datei (in meinem Fall aus dem Pipeline-Repository löschen und auf Airflow-Servern bereitstellen, insbesondere im Scheduler).
2. Aktualisieren Sie die Web-GUI NICHT.
3. Klicken Sie in der Web-GUI in der DAG-Ansicht (normale Startseite) auf "Tag löschen" -> das rote Symbol ganz rechts.
4. Es bereinigt alle Überreste dieser DAG aus der Datenbank.

Airflow 1.10.1 wurde freigegeben. Diese Version bietet die Möglichkeit, eine DAG von der Web-Benutzeroberfläche zu löschen, nachdem Sie die entsprechende DAG aus dem Dateisystem gelöscht haben.

Weitere Informationen finden Sie in diesem Ticket:

[AIRFLOW-2657] Möglichkeit zum Löschen von DAG aus der Web-Benutzeroberfläche hinzugefügt

Bitte beachten Sie, dass dadurch die DAG nicht tatsächlich aus dem Dateisystem gelöscht wird. Sie müssen dies zuerst manuell tun, da sonst die DAG neu geladen wird.

Ich habe ein Skript geschrieben, das alle Metadaten löscht, die sich auf einen bestimmten Tag für die Standard-SQLite-Datenbank beziehen. Dies basiert auf der obigen Antwort Jesu, wurde jedoch von Postgres an SQLite angepasst. Benutzer sollten festlegen ../airflow.db, wo script.py relativ zur Standarddatei airflow.db gespeichert ist (normalerweise ~/airflow). Verwenden Sie zum Ausführen 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()

In Airflow ist nichts eingebaut, was dies für Sie erledigt. Um die DAG zu löschen, löschen Sie sie aus dem Repository und löschen Sie die Datenbankeinträge in der Airflow-Metastore-Tabelle - dag.

Sie können eine Reihe von Aufgabeninstanzen löschen, als ob sie nie ausgeführt worden wären mit:

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

Und dann entfernen Sie die DAG-Datei aus dem DAG-Ordner

Basierend auf der Antwort von @OlegYamin gehe ich wie folgt vor, um einen von postgres gesicherten Tag zu löschen, bei dem der Luftstrom das publicSchema verwendet.

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';

WARNUNG : Der Effekt / die Richtigkeit der ersten Löschabfrage ist mir unbekannt. Es ist nur eine Annahme, dass es benötigt wird.

lösche es einfach aus mysql, funktioniert gut für mich. Löschen Sie sie aus den folgenden Tabellen:

• dag

• dag_constructor

• dag_group_ship
• dag_pickle
• dag_run
• dag_stats

(möglicherweise mehr Tabellen in zukünftigen Versionen) Starten Sie dann den Webserver und den Worker neu.

Versionen> = 1.10.0:

Ich habe die Airflow-Version 1.10.2 und habe versucht, den Befehl air_ete delete_dag auszuführen, aber der Befehl gibt folgenden Fehler aus:

bash-4.2 # airflow delete_dag dag_id

[2019-03-16 15: 37: 20,804] {settings.py:174} INFO - settings.configure_orm (): Verwenden der Pooleinstellungen. pool_size = 5, pool_recycle = 1800, pid = 28224 /usr/lib64/python2.7/site-packages/psycopg2/ init .py: 144: UserWarning: Das Radpaket psycopg2 wird ab Version 2.8 umbenannt. Um die Installation von der Binärdatei aus fortzusetzen, verwenden Sie stattdessen "pip install psycopg2-binary". Weitere Informationen finden Sie unter: http://initd.org/psycopg/docs/install.html#binary-install-from-pypi. "" ") Dadurch werden alle vorhandenen Datensätze gelöscht, die sich auf die angegebene DAG beziehen. Weiter? (J / N) y Rückverfolgung (letzter Aufruf zuletzt): Datei" / usr / bin / airflow ", Zeile 32, in args.func ( args) Datei "/usr/lib/python2.7/site-packages/airflow/utils/cli.py", Zeile 74, im Wrapper return f (* args, ** kwargs) Datei "/ usr / lib / python2. 7 / site-packages / airflow / bin / cli.py ", Zeile 258, in delete_dag erhöhen AirflowException (err) airflow.exceptions.AirflowException: Serverfehler

Obwohl ich in der Lage bin, über den Befehl Curl zu löschen. Bitte lassen Sie mich wissen, wenn jemand eine Vorstellung von der Ausführung dieses Befehls hat, dies bekannt ist oder ich etwas falsch mache.

Versionen <= 1.9.0:

Es gibt keinen Befehl zum Löschen eines Tages, daher müssen Sie zuerst die Tag-Datei löschen und dann alle Verweise auf die Tag-ID aus der Luftstrom-Metadatendatenbank löschen.

WARNUNG

Sie können die Luftstrom-Metadatenbank zurücksetzen. Sie löschen alles, einschließlich der Dags. Denken Sie jedoch daran, dass Sie auch den Verlauf, die Pools, Variablen usw. löschen.

airflow resetdb und dann airflow initdb

Zuerst -> Löschen Sie die DAG-Datei aus dem Ordner $ AIRFLOW_HOME / dags. Hinweis: Je nachdem, ob Sie Unterverzeichnisse verwendet haben, müssen Sie möglicherweise die Unterverzeichnisse durchsuchen, um die DAG-Datei zu finden und zu löschen.

Zweitens -> Löschen Sie die DAG von der Webserver-Benutzeroberfläche mit der Schaltfläche Löschen (x im Kreis).

Für diejenigen, die direkten Zugriff auf die Postgres- psqlKonsole der Luftstromdatenbank haben, können Sie einfach die folgende Anforderung ausführen, um die DAG zu entfernen:

\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';

Eine ähnliche Abfrage (mit geringfügigen Änderungen) eignet sich für andere Datenbanken wie MySQL und SQLite.

Entfernen Sie den Dag (den Sie löschen möchten) aus dem Dags-Ordner und führen Sie ihn aus airflow resetdb.

Alternativ können Sie in die airflow_db gehen und diese Einträge manuell aus den dag-Tabellen löschen (task_fail, xcom, task_instance, sla_miss, log, job, dag_run, dag, dag_stats).

Für diejenigen, die noch Antworten finden. In Airflow Version 1.8 ist es sehr schwierig, eine DAG zu löschen. Sie können sich auf die obigen Antworten beziehen. Aber seit 1.9 veröffentlicht wurde, müssen Sie nur noch

Entfernen Sie den Tag im Ordner "Dags" und starten Sie den Webserver neu