Monitoring, Alerts und Callbacks

Einführung in Apache Airflow mit Python

Mike Metzger

Data Engineer

Dag‑Lebenszyklus

Jeder Dag‑Run und jede Task durchläuft Zustände
- Queued
- Running
- Final: Success, Failed, Skipped
Airflow verfolgt Zustände und Übergänge

Diagramm des Dag-Run-Lebenszyklus von queued über running bis zu einem Endzustand

Callbacks

Funktionen an Dag‑Run oder Task, die Airflow automatisch aufruft
Callbacks für bestimmte Zustandswechsel
- on_failure_callback – Dag‑Run/Task schlägt fehl
- on_success_callback – Dag‑Run/Task erfolgreich
Task‑spezifische Callbacks
- on_retry_callback
- on_skipped_callback
- on_execute_callback

Illustration von Airflow-Callbacks, die bei Zustandswechseln ausgelöst werden

Callback-Kontext

Airflow übergibt automatisch ein context‑Dictionary
context enthält Infos zum Dag‑Run/Task
- context["dag"].dag_id – Name des Dags
- context["task_instance"].task_id – Name der Task
- context["logical_date"] – Datum des Dag‑Runs
Callback‑Funktion muss context akzeptieren

def alert_on_failure(context):

    dag_id = context["dag"].dag_id
    task_id = context["task_instance"].task_id
    print(f"Task {task_id} in DAG {dag_id} has failed.")

Callback‑Beispiel

def alert_on_failure(context):
    dag_id = context["dag"].dag_id
    task_id = context["task_instance"].task_id
    print(f"Task {task_id} in Dag {dag_id} has failed.")


@dag(on_failure_callback=alert_on_failure)
def sales_etl_dag():
    @task()
    def data_import_task():
        raise ValueError("Simulated failure")


# Task data_import_task in Dag sales_etl_dag has failed.

Notifier

Funktionen, die an Callbacks gebunden werden können
Senden Alerts an externe Systeme
Verfügbare Notifier:
- SmtpNotifier – sendet E‑Mail‑Alerts
- SlackNotifier – postet in einen Slack‑Channel
- App‑spezifische Notifier – PagerDuty, OpsGenie

Illustration von Airflow-Notifiers, die Alerts an externe Systeme senden

SmtpNotifier

Sendet bei Callback eine E‑Mail
In der Bibliothek airflow.providers.smtp.notifications.smtp
Benötigt from_email und to

Optional: subject, html_content, usw.

@dag(dag_id=`sales_etl_dag`,
   on_failure_callback=SmtpNotifier(
       to='[email protected]',
       from_email='[email protected]',
       subject='Dag sales_etl_dag has failed'
   )
)

Audit Log

Nutze das Audit Log, um Airflow zu überwachen
Enthält die Folge aller Events der Airflow‑Instanz

Airflow-Audit-Log-Seite mit Systemereignissen nach Typ

Lass uns üben!

Einführung in Apache Airflow mit Python