Monitorování, upozornění a callbacky

Úvod do Apache Airflow v Pythonu

Mike Metzger

Data Engineer

Životní cyklus DAGu

Každý běh DAGu a úloha prochází posloupností stavů
- Ve frontě
- Probíhá
- Konečný stav: Úspěch, Selhání, Přeskočeno
Airflow sleduje stavy a přechody

Diagram životního cyklu běhu DAGu: fronta, spuštění a konečný stav

Callbacky

Funkce volané Airflow automaticky při běhu DAGu nebo úlohy
Callbacky pro konkrétní přechody stavů
- on_failure_callback – běh DAGu / úloha selhala
- on_success_callback – běh DAGu / úloha uspěla
Callbacky specifické pro úlohu
- on_retry_callback
- on_skipped_callback
- on_execute_callback

Ilustrace callbacků Airflow spouštěných při přechodech stavů

Kontext callbacku

Airflow automaticky předává slovník context
context obsahuje informace o běhu DAGu / úloze
- context["dag"].dag_id – název DAGu
- context["task_instance"].task_id – název úlohy
- context["logical_date"] – datum běhu DAGu
Callback funkce musí přijímat 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.")

Příklad callbacku

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.

Notifikátory

Funkce propojitelné s callbacky
Odesílají upozornění do externích systémů
Dostupné notifikátory:
- SmtpNotifier – odesílá e-mailová upozornění
- SlackNotifier – publikuje zprávy do Slack kanálu
- Aplikačně specifické notifikátory – PagerDuty, OpsGenie

Ilustrace notifikátorů Airflow odesílajících upozornění do externích systémů

SmtpNotifier

Odesílá e-mail při callbacku
Knihovna airflow.providers.smtp.notifications.smtp
Vyžaduje atributy from_email a to

Lze doplnit subject, html_content atd.

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

Protokol auditu

Sledujte Airflow pomocí protokolu auditu
Obsahuje záznamy všech událostí v instanci Airflow

Stránka protokolu auditu Airflow se seznamem systémových událostí podle typu

Pojďme procvičovat!

Úvod do Apache Airflow v Pythonu