Wie man den Prozess der Automatisierung mit Airflow und DAGs effizient gestaltet

Um den Prozess der Automatisierung mit Apache Airflow effizient zu gestalten, ist es notwendig, einen sogenannten Directed Acyclic Graph (DAG) zu erstellen. Ein DAG ermöglicht es, verschiedene Aufgaben in einem Workflowsystem zu definieren und deren Ausführung zu steuern. In diesem Zusammenhang wird in den folgenden Schritten erläutert, wie man einen DAG aufsetzt, um einen bestimmten Prozess regelmäßig, etwa täglich, auszuführen.

Zunächst müssen die erforderlichen Python- und Airflow-Bibliotheken importiert werden. Dazu gehören grundlegende Module wie json und pathlib, aber auch spezifische Airflow-Imports wie DAG, PythonOperator und BashOperator. Diese Bibliotheken sind notwendig, um den DAG zu erstellen und verschiedene Aufgaben innerhalb des DAGs zu definieren. Ein Beispiel für die ersten Schritte könnte so aussehen:

python
Copy code
import json
import pathlib
import airflow
import requests
from datetime import date
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.decorators import task
from datetime import datetime, timedelta

dag_owner = 'Kendrick'

Im nächsten Schritt wird eine Python-Funktion definiert, die ein Bild von der NASA-API abruft. Diese API liefert täglich ein neues Bild, das mithilfe des PythonOperators im DAG heruntergeladen und lokal gespeichert wird. Der Code für diese Funktion könnte wie folgt aussehen:

python
Copy code
def _get_pictures():
    pathlib.Path("/tmp/images").mkdir(parents=True, exist_ok=True)
    api_key = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
    url = f'https://api.nasa.gov/planetary/apod?api_key={api_key}'
    response = requests.get(url).json()
    today_image = response['hdurl']
    with open(f'todays_image_{date.today()}.png', 'wb') as f:
        f.write(requests.get(today_image).content)

Nachdem die Funktion definiert ist, geht es weiter mit der Festlegung der Standardargumente für den DAG. Diese Argumente geben an, wie Airflow standardmäßig mit den Aufgaben umgehen soll. Typische Parameter sind beispielsweise der Eigentümer des DAGs, ob frühere Ausführungen des DAGs berücksichtigt werden sollen (depends_on_past), die Anzahl der Versuche bei einem Fehler (retries) und die Verzögerung zwischen den Wiederholungsversuchen (retry_delay).

python
Copy code
default_args = {
    'owner': dag_owner,
    'depends_on_past': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=5)
}

Sobald diese Parameter definiert sind, wird der eigentliche DAG mit den entsprechenden Attributen wie dag_id, description, start_date und schedule_interval initialisiert. In unserem Beispiel wird der DAG so eingestellt, dass er täglich ausgeführt wird:

python
Copy code
with DAG(dag_id='download_ASOD_image',
         default_args=default_args,
         description='Download und Benachrichtigung',
         start_date=airflow.utils.dates.days_ago(0),
         schedule_interval='@daily',
         catchup=False,
         tags=['None']):

Der schedule_interval ist ein wichtiger Parameter, der festlegt, wie häufig der DAG ausgeführt wird. In unserem Fall wird der DAG mit der Funktion @daily so eingestellt, dass er täglich läuft. Es gibt jedoch auch komplexere Anforderungen, bei denen ein DAG nur an bestimmten Wochentagen oder zu bestimmten Zeiten ausgeführt werden soll, was ebenfalls über den schedule_interval gesteuert werden kann.

Ein weiteres wichtiges Element ist das Attribut catchup. Dieses steuert, ob Airflow verpasste Ausführungen des DAGs nachholen soll, falls der Startzeitpunkt des DAGs in der Vergangenheit liegt. In unserem Beispiel haben wir catchup=False gesetzt, sodass nur die zukünftigen Ausführungen geplant werden. Wenn der Startzeitpunkt beispielsweise auf vor sieben Tagen festgelegt wird und catchup=True gesetzt ist, würde Airflow den DAG siebenmal ausführen, um die verpassten Intervalle nachzuholen.

Nach der Initialisierung des DAGs werden die Aufgaben des DAGs definiert. Zuerst wird die Aufgabe zum Abrufen der Bilder mit einem PythonOperator verknüpft. Der Operator stellt sicher, dass die definierte Python-Funktion ausgeführt wird:

python
Copy code
get_pictures = PythonOperator(
    task_id="get_pictures",
    python_callable=_get_pictures,
)

Anschließend wird eine Benachrichtigungsaufgabe hinzugefügt, die uns informiert, dass das Bild erfolgreich heruntergeladen wurde. In diesem Fall wird der BashOperator verwendet, um eine einfache Nachricht in der Konsole auszugeben. Es ist jedoch auch möglich, die Benachrichtigung beispielsweise per E-Mail oder über Slack zu versenden:

python
Copy code
notify = BashOperator(
    task_id="notify",
    bash_command='echo f"Das Bild für heute wurde hinzugefügt!"',
)

Schließlich wird die Ausführungsreihenfolge der Aufgaben im DAG festgelegt. Die Aufgabe zum Abrufen der Bilder wird als erste Aufgabe definiert, während die Benachrichtigung als nachfolgende Aufgabe festgelegt wird:

python
Copy code
get_pictures >> notify

Der gesamte Ablauf zeigt, wie man mit Airflow einen einfachen DAG erstellt, um regelmäßig eine Aufgabe (hier das Abrufen eines Bildes) auszuführen und eine Benachrichtigung über den erfolgreichen Abschluss der Aufgabe zu erhalten.

Es ist wichtig zu verstehen, dass das Schreiben eines DAGs zunächst komplex erscheinen mag, insbesondere aufgrund der vielen Wiederholungen und der Vielzahl an Parametern. Jedoch handelt es sich häufig um wiederkehrende Aufgaben, bei denen die Grundlogik des DAGs unverändert bleibt. Ein Schlüsselvorteil von Airflow liegt in der Möglichkeit, manuelle und zeitintensive Aufgaben zu automatisieren und in kleinere, überschaubare Tasks zu unterteilen. Die Automatisierung dieser Aufgaben trägt dazu bei, den gesamten Arbeitsablauf effizienter und fehlerfreier zu gestalten.

Neben der grundlegenden Erstellung und Verwaltung von DAGs gibt es noch viele fortgeschrittene Konzepte und Möglichkeiten, wie etwa das parallele Ausführen von Aufgaben oder das Erstellen von komplexeren Workflows mit Abhängigkeiten zwischen den Aufgaben. Diese Aspekte werden in weiteren Kapiteln vertieft, um dem Leser ein vollständiges Verständnis für die vielseitigen Einsatzmöglichkeiten von Airflow zu vermitteln.

Wie testet man Airflow-Workflows effektiv und beobachtet ihre Stabilität in der Produktion?

In modernen datengetriebenen Architekturen ist die Automatisierung von Tests innerhalb des CI/CD-Prozesses nicht nur eine Effizienzfrage, sondern eine Notwendigkeit zur Sicherstellung von Zuverlässigkeit und Skalierbarkeit. Tests sollten kontextsensitiv in Umgebungen ausgeführt werden, die sowohl den operativen Rahmenbedingungen des Unternehmens als auch der Phase des Entwicklungszyklus entsprechen. Leichte Tests wie Smoke- oder Unit-Tests laufen idealerweise vollständig isoliert innerhalb des CI/CD-Systems, wohingegen funktionale Tests in dedizierten QA-Umgebungen stattfinden, in denen gezielte Datenmanipulation für spezifische Testszenarien erlaubt ist. Performance-Tests erfordern eine Staging-Umgebung, die der Produktionsumgebung nahezu vollständig entspricht.

Die Zusammenarbeit mit Infrastruktur-, Security- oder DevOps-Teams ist essenziell, um eine Teststrategie zu entwickeln, die sowohl unternehmensweite Standards erfüllt als auch die individuellen Anforderungen des Teams widerspiegelt. Dabei sollte eine klar abgestufte Testhierarchie eingehalten werden, bei der günstigere Tests vor ressourcenintensiven durchgeführt werden, um unnötige Kosten und Zeitverluste zu vermeiden.

Das Testen von Providern stellt dabei eine verhältnismäßig geringe Herausforderung dar, da hier die funktionalen Anforderungen in der Regel überschaubar und gut automatisierbar sind. Smoke-Tests dienen der Grundvalidierung: Lässt sich das Python-Package installieren? Ist der grundlegende Import erfolgreich? Unit-Tests isolieren kleinste Codeeinheiten, meist unter Verwendung von Mocks, um Abhängigkeiten von externen Diensten zu vermeiden. Funktionale bzw. Integrationstests hingegen validieren das Zusammenspiel des Codes mit realen, laufenden Systemen. Hierbei müssen oftmals Datenbanken oder andere Dienste temporär aufgebaut und wieder abgebaut sowie mit Testdaten vorbefüllt werden.

Ein häufiges Missverständnis besteht darin, Airflow selbst zu testen. Dies ist nur dann ratsam, wenn der Airflow-Code substanziell verändert wurde und ein Fork unterhalten wird – was nur in äußersten Ausnahmefällen geschehen sollte. Der Airflow-Core ist upstream durch die Maintainer intensiv getestet. Sollte man dennoch geneigt sein, Airflow auf Korrektheit zu testen, ist es dringend anzuraten, die eigene Strategie zu überdenken und gegebenenfalls die Community einzubeziehen, anstatt Ressourcen in redundante Tests zu investieren.

Sobald die Workflows in die Produktionsphase übergehen, verschiebt sich der Fokus von reinen Tests hin zur kontinuierlichen Beobachtung und Überwachung – sowohl auf System- als auch auf DAG-Ebene. Dabei wird zwischen aktivem und suppressivem Monitoring unterschieden. Aktives Monitoring prüft regelmäßig den Zustand eines Dienstes und reagiert unmittelbar auf dessen Rückgabewerte. Suppressives Monitoring hingegen basiert auf dem Ausbleiben einer erwarteten Zustandsmeldung, wobei das zu überwachende System periodisch Lebenszeichen sendet, um etwaige Alarmierungen zu unterdrücken.

Im Kernsystem von Airflow sind insbesondere Scheduler, Webserver und Metadatenbank kritisch für die Betriebsfähigkeit. Bereits einfache Checks – wie eine Anfrage an das /health/-Endpoint des Webservers – ermöglichen grundlegende Aussagen über den Zustand des Deployments. Der Scheduler ist dabei besonders sensibel: Obwohl er einen "healthy" Status zurückgeben kann, bedeutet dies nicht zwangsläufig, dass er tatsächlich korrekt arbeitet. In der Praxis werden deshalb sogenannte Canary-DAGs eingesetzt – einfache DAGs mit minimalem Inhalt, die durch ihren erfolgreichen Durchlauf eine indirekte Bestätigung der Systemgesundheit liefern.

Zu überwachende Metriken im Scheduler sind unter anderem die Dauer des Hauptschleifenzyklus (scheduler.scheduler_loop_duration), die Anzahl der "verhungernden" Tasks ohne verfügbare Slots (scheduler.tasks.starving) sowie die Menge ausführbarer, aber nicht gestarteter Tasks (scheduler.tasks.executable). Ein Anstieg dieser Werte über längere Zeiträume hinweg deutet auf Engpässe hin – sei es auf Ebene der DAG-Architektur oder in der verfügbaren Rechenkapazität des Executors.

Die Metadatenbank wiederum speichert sämtliche Informationen über vorherige Ausführungen, Rollen, Berechtigungen und Konfigurationen. Sie ist das Rückgrat der Airflow-Instanz und muss dementsprechend durchgängig verfügbar sein. Der Verlust dieser Daten hat unmittelbare, teils irreversible Auswirkungen. Aus diesem Grund wird dringend empfohlen, eine verwaltete Datenbanklösung zu verwenden und ein belastbares Disaster-Recovery-Konzept zu etablieren. Auch hier sind Metriken wie die Nutzung des Verbindungspools oder die Schreib-/Leseoperationen von Bedeutung.

Darüber hinaus ist es für Leser wichtig zu verstehen, dass Testen und Beobachten nicht als getrennte Phasen betrachtet werden dürfen. Vielmehr ist eine symbiotische Integration beider Disziplinen im gesamten Lebenszyklus eines Workflows essenziell. Ein guter Test deckt keine Codefehler allein auf, sondern prüft auch, ob das System sich unter realen Bedingungen wie erwartet verhält. Ebenso ist Monitoring nicht nur Fehlerdetektion, sondern dient der Verifikation von Systemverhalten über die Zeit – ein kontinuierlicher Test in Produktion.

Wesentlich ist auch, dass die Auswahl von Monitoring- und Telemetrie-Tools nicht isoliert vom Rest des Stacks geschieht. Teams sollten eng mit ihren Plattformverantwortlichen zusammenarbeiten, um vorhandene Tools effizient zu konfigurieren und konsistent in bestehende Alerting-Infrastrukturen einzubinden. Die Sichtbarkeit auf DAG-Ebene ist genauso entscheidend wie auf Systemebene – denn ein fehlerhafter DAG kann ein ganzes Orchestrierungssystem blockieren, auch wenn die darunterliegende Infrastruktur einwandfrei arbeitet.