Supervisión, alertas y callbacks

Introducción a Apache Airflow en Python

Mike Metzger

Data Engineer

Ciclo de vida de un Dag

Cada ejecución de Dag y tarea pasa por estados
- En cola
- En ejecución
- Estado final: Éxito, Error, Omitida
Airflow registra estados y transiciones

Diagrama del ciclo de vida de un Dag: en cola, en ejecución y estado final

Callbacks

Funciones en la ejecución del Dag o Tarea que Airflow llama automáticamente
Callbacks para transiciones de estado
- on_failure_callback - Falla el Dag/tarea
- on_success_callback - Éxito del Dag/tarea
Callbacks de tarea
- on_retry_callback
- on_skipped_callback
- on_execute_callback

Ilustración de callbacks de Airflow activados por transiciones de estado

Contexto del callback

Airflow pasa un diccionario context automáticamente
context contiene info del Dag/tarea
- context["dag"].dag_id - Nombre del Dag
- context["task_instance"].task_id - Nombre de la tarea
- context["logical_date"] - Fecha de la ejecución del Dag
La función callback debe aceptar 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.")

Ejemplo de 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.

Notifiers

Funciones vinculadas a callbacks
Envían alertas a sistemas externos
Varios notifiers disponibles:
- SmtpNotifier - Envía alertas por email
- SlackNotifier - Publica en un canal de Slack
- Notifiers específicos: PagerDuty, OpsGenie

Ilustración de notifiers de Airflow que envían alertas a sistemas externos

SmtpNotifier

Envía un email en el callback
En la librería airflow.providers.smtp.notifications.smtp
Requiere from_email y to

Puede incluir subject, html_content, etc.

@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 el Audit Log para supervisar Airflow
Contiene la secuencia de todos los eventos de la instancia

Página de Audit Log de Airflow con eventos del sistema por tipo

¡Vamos a practicar!

Introducción a Apache Airflow en Python