Övervakning, aviseringar och callbacks

Introduktion till Apache Airflow i Python

Mike Metzger

Data Engineer

DAG-livscykel

Varje DAG-körning och uppgift genomgår en serie tillstånd
- I kö
- Körs
- Slutläge: Lyckades, Misslyckades, Hoppades över
Airflow spårar tillstånd och övergångar

Diagram över DAG-körningens livscykel från kö till körning till slutläge

Callbacks

Funktioner på DAG-körning eller uppgift som Airflow anropar automatiskt
Callbacks för specifika tillståndsövergångar
- on_failure_callback – DAG-körning/uppgift misslyckas
- on_success_callback – DAG-körning/uppgift lyckas
Uppgiftsspecifika callbacks
- on_retry_callback
- on_skipped_callback
- on_execute_callback

Illustration av Airflow-callbacks som utlöses vid tillståndsövergångar

Callback-kontext

Airflow skickar automatiskt en context-ordlista
context innehåller information om DAG-körningen/uppgiften
- context["dag"].dag_id – DAG-namn
- context["task_instance"].task_id – Uppgiftsnamn
- context["logical_date"] – Datum för DAG-körningen
Callback-funktionen måste acceptera context

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-exempel

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.

Notifierare

Funktioner som kan kopplas till callbacks
Skickar aviseringar till externa system
Flera notifierare finns tillgängliga:
- SmtpNotifier – Skickar e-postaviseringar
- SlackNotifier – Publicerar meddelanden i en Slack-kanal
- Appspecifika notifierare – PagerDuty, OpsGenie

Illustration av Airflow-notifierare som skickar aviseringar till externa system

SmtpNotifier

Skickar ett e-postmeddelande vid callback
Finns i biblioteket airflow.providers.smtp.notifications.smtp
Kräver attributen from_email och to

Kan även inkludera subject, html_content m.m.

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

Granskningslogg

Använd granskningsloggen för att övervaka Airflow
Innehåller alla händelser på Airflow-instansen i ordning

Airflows granskningsloggsida med systemhändelser sorterade efter typ

Nu kör vi en övning!

Introduktion till Apache Airflow i Python