Das Zen von Python und Apache Airflow

Apache Airflow ist ein Python-Framework zur programmgesteuerten Erstellung von Workflows in DAGs, z.B. ETL-Prozesse, Erstellung von Berichten und tägliches Neutrainieren von Modellen. Dies ermöglicht prägnante und flexible Skripte, kann aber auch der Nachteil von Airflow sein. Da es sich um Python-Code handelt, gibt es unendlich viele Möglichkeiten, Ihre Pipelines zu definieren. The Zen of Python ist eine Liste von 19 Python-Designprinzipien und in diesem Blogbeitrag zeige ich einige dieser Prinzipien anhand von vier Airflow-Beispielen auf. Dieser Blogbeitrag wurde mit Airflow 1.10.2 geschrieben.

1. Der DAG-Kontextmanager

Ein DAG-Objekt kann auf zwei Arten instanziiert und in Aufgaben referenziert werden:

Option 1: Übergeben Sie explizit eine DAG-Referenz:

Option 2: Verwenden Sie die DAG im Kontextmanager, ohne auf das DAG-Objekt verweisen zu müssen:

Wenn Sie die Implementierung des Kontextmanagers überprüfen, sehen Sie, dass die DAG auf eine globale Variable gesetzt wird¹ _CONTEXT_MANAGER_DAG. Die Operatoren prüfen einfach, ob _CONTEXT_MANAGER_DAG vorhanden ist, und wenn ja, setzen sie dag=_CONTEXT_MANAGER_DAG. Wir müssen also nicht einmal as dag schreiben (es sei denn, Sie möchten irgendwo in Ihrem Code darauf verweisen):

Der DAG-Kontextmanager scheint wie von Zauberhand zu funktionieren, ohne dass Sie ihn in einer Aufgabe referenzieren müssen. Angesichts des Python-Designprinzips"Explizit ist besser als implizit" ziehe ich daher die Option 1 vor.

2. Abhängigkeiten zwischen Aufgaben festlegen

Die Operatoren >> und << können verwendet werden, um sowohl eine einzelne Aufgabe als auch eine Liste von Aufgaben zu verbinden. Die Bit-Shift-Operatoren wurden in Airflow 1.8 als Alternative zu set_downstream() (>>) und set_upstream() (<<) eingeführt. Die implementierenden Funktionen (lshift und rshift) übergeben einfach Variablen an set_downstream und set_upstream, die ihrerseits alle _set_relatives() aufrufen, das sowohl eine einzelne Aufgabe als auch eine Liste von Aufgaben akzeptiert (eine einzelne Aufgabe wird in eine Liste umgewandelt, die die einzelne Aufgabe enthält).

Dies ermöglicht das Verbinden von Listen mit Aufgaben:

rlshift und rrshift sind ebenfalls implementiert, um das Setzen von Abhängigkeiten in umgekehrter Richtung zu unterstützen. Diese Operationen werden auf dem Objekt der rechten Seite aufgerufen, wenn die linke Seite keine Implementierung für den Bit-Shift-Operator hat, was bei Listen der Fall ist:

Dies ermöglicht eine konsequente Verkettung von Aufgaben wie z.B.:

In Anbetracht des Grundsatzes"Die Lesbarkeit zählt" halte ich es der Klarheit halber für besser, die Richtungen in einer einzigen Aussage nicht zu vermischen. Und wenn möglich, sollten Sie immer die absteigende Richtung verwenden, da die meisten Menschen auf natürliche Weise von links nach rechts lesen: Beachten Sie, dass es nicht möglich ist, eine Liste auf beiden Seiten zu referenzieren, da eine Liste keine Implementierung für die Bitverschiebungsoperation hat:

3. Übergabe von Kontext an Aufgaben

Eine übliche Datenpipeline ist ein tägliches Laden von Daten, wobei die Daten in Partitionen pro Tag geschrieben werden, notiert als z.B. dt=yyyy-mm-dd. Airflow unterstützt diesen Anwendungsfall, indem es Zugriff auf den Aufgabenkontext bietet. Wenn wir provide_context=True einem Operator zur Verfügung stellen, geben wir die Airflow-Kontextvariablen weiter, die innerhalb des Operators verwendet werden sollen. Diese Kontextvariablen enthalten u.a. das Startdatum des Intervalls in verschiedenen Formaten (z.B. ds="yyyy-mm-dd", ds_nodash="yyyymmdd" und execution_date=pendulum.pendulum.Pendulum). Um alle verfügbaren Variablen einzusehen, drucken Sie den Kontext aus oder sehen Sie sich die Dokumentation unter default-variables an. In Python ist es üblich, Schlüsselwortargumente in einer Funktion mit dem Argumentnamen kwargs zu akzeptieren. Wenn ich Funktionen im Kontext von Airflow schreibe, ziehe ich es vor, diese Variable context zu nennen, um ihren Zweck für die Weitergabe des Kontexts der Airflow-Aufgabeninstanz anzugeben. Beachten Sie auch, dass wir die Funktion _print_exec_date wie folgt implementieren können, da wir wissen, dass der Airflow-Kontext eine Variable execution_date enthält. Die Variable context enthält nun alle Airflow-Kontextvariablen mit Ausnahme von execution_date, da diese an das Argument execution_date übergeben wird, und context enthält alle übrigen Schlüsselwortargumente. Um die Absicht zu zeigen, kein anderes Argument als executiondate zu verwenden, ist es noch deutlicher, alle anderen Argumente mit ** zu verwerfen: Meiner Meinung nach zeigt die Verwendung von **_ und nur der Variablen, die Sie aus dem Airflow-Kontext benötigen, dem Leser Ihres Codes die erwartete Eingabe und verbessert die Eindeutigkeit und Lesbarkeit.

4. Überspringen der Ausführung von Aufgaben

Ein weiteres Beispiel aus der Kategorie "explizit vor implizit". Manchmal möchten Sie die Ausführung von Aufgaben in Ihrer DAG auslassen. Sie laden zum Beispiel jeden Tag eine .zip-Datei von einem SFTP-Server herunter. Der SFTP-Server speichert den Verlauf von einer Woche. Das ist in Ordnung, wenn Sie einmal am Tag den Datenstapel von gestern herunterladen. Es funktioniert jedoch nicht, wenn Sie versuchen, mehr als eine Woche in die Vergangenheit zurückzurechnen. Für diesen Anwendungsfall könnten Sie die erste Aufgabe Ihrer DAG so einrichten, dass sie prüft, ob das Ausführungsdatum vor heute - einer Woche - liegt, und die Ausführung überspringt, wenn diese Bedingung erfüllt ist. Dies kann auf mehrere Arten erreicht werden. Die erste ist die ShortCircuitOperator. Sie erfordert eine Callable, die True zurückgibt, wenn nachgelagerte Aufgaben fortgesetzt werden sollen, und False, wenn nachgelagerte Aufgaben übersprungen werden sollen. Ein Beispiel: Eine andere Möglichkeit besteht darin, den PythonOperator zu verwenden und eine AirflowSkipException auszulösen, wenn eine bestimmte Bedingung nicht erfüllt ist. Dadurch wird die Aufgabe übersprungen und damit auch alle nachgelagerten Aufgaben. Beachten Sie, dass dies das Standardargument trigger_rule=TriggerRule.ALL_SUCCESS für die Operatoren voraussetzt. Die TriggerRule definiert die Bedingung für die Ausführung eines Operators. Wenn eine vorgelagerte Abhängigkeit eines Operators den Zustand übersprungen hat und seine trigger_rule TriggerRule.ALL_SUCCESS ist, wird er ebenfalls übersprungen. Was mir an dieser Lösung gefällt, ist, dass sie auch den Status der Aufgabe, die die Bedingung überprüft, auf "übersprungen" setzt, was meiner Meinung nach visuell klarer ist.

Letzte Worte

Die meisten Beispiele laufen auf Eindeutigkeit und Lesbarkeit hinaus. Das hilft Ihrem zukünftigen Ich und anderen, Ihren Code leicht zu verstehen, was ich weniger Zeichen vorziehe. Wenden Sie sich an uns, wenn Sie Fragen zu Airflow haben!

Sind Sie an einer Schulung zu Data Science mit Python oder Apache Airflow interessiert?

Nehmen Sie an einem unserer Kurse teil: - Apache Airflow - Python Grundlagen - Data Science mit Python Foundation - Fortgeschrittene Datenwissenschaft mit Python