Monitoraggio, avvisi e callback

Introduzione ad Apache Airflow in Python

Mike Metzger

Data Engineer

Ciclo di vita del Dag

Ogni esecuzione di Dag e task passa per una sequenza di stati
- In coda
- In esecuzione
- Stato finale: Riuscito, Fallito, Saltato
Airflow traccia stati e transizioni

Diagramma del ciclo di vita del Dag: in coda, in esecuzione, stato finale

Callback

Funzioni su esecuzione Dag o Task chiamate automaticamente da Airflow
Callback per specifiche transizioni di stato
- on_failure_callback - Dag/task fallisce
- on_success_callback - Dag/task riesce
Callback specifici del task
- on_retry_callback
- on_skipped_callback
- on_execute_callback

Illustrazione dei callback di Airflow attivati sulle transizioni di stato

Contesto del callback

Airflow passa automaticamente un dizionario context
context contiene info sull'esecuzione del Dag/task
- context["dag"].dag_id - Nome del Dag
- context["task_instance"].task_id - Nome del task
- context["logical_date"] - Data dell'esecuzione del Dag
La funzione di callback deve accettare 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.")

Esempio di callback

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

Funzioni collegabili ai callback
Invia avvisi a sistemi esterni
Vari notifier disponibili:
- SmtpNotifier - Invia avvisi email
- SlackNotifier - Pubblica messaggi su un canale Slack
- Notifier specifici per app - PagerDuty, OpsGenie

Illustrazione dei notifier di Airflow che inviano avvisi a sistemi esterni

SmtpNotifier

Invia un'email al callback
Nella libreria airflow.providers.smtp.notifications.smtp
Richiede gli attributi from_email e to

Può includere subject, html_content, ecc.

@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

Usa l'Audit Log per monitorare Airflow
Contiene la sequenza di tutti gli eventi dell'istanza Airflow

Pagina del registro di controllo di Airflow con gli eventi di sistema per tipo

Passiamo alla pratica!

Introduzione ad Apache Airflow in Python