Monitoramento, alertas e callbacks

Introdução ao Apache Airflow em Python

Mike Metzger

Data Engineer

Ciclo de vida do Dag

Cada execução de Dag e tarefa passa por estados
- Em fila (Queued)
- Em execução (Running)
- Final: Sucesso, Falha, Ignorado
O Airflow rastreia estados e transições

Diagrama do ciclo de vida do Dag: de em fila a em execução até um estado final

Callbacks

Funções no Dag run ou Task que o Airflow chama automaticamente
Callbacks para transições de estado
- on_failure_callback - Dag/tarefa falhou
- on_success_callback - Dag/tarefa teve sucesso
Callbacks específicos de tarefa
- on_retry_callback
- on_skipped_callback
- on_execute_callback

Ilustração de callbacks do Airflow acionados em transições de estado

Contexto do callback

O Airflow passa um dicionário context automaticamente
context traz infos do Dag/tarefa
- context["dag"].dag_id - Nome do Dag
- context["task_instance"].task_id - Nome da tarefa
- context["logical_date"] - Data da execução do Dag
A função de callback deve aceitar 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.")

Exemplo 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

Funções ligadas a callbacks
Enviam alertas a sistemas externos
Notificadores disponíveis:
- SmtpNotifier - Envia alertas por e-mail
- SlackNotifier - Publica no canal do Slack
- Específicos: PagerDuty, OpsGenie

Ilustração de notifiers do Airflow enviando alertas para sistemas externos

SmtpNotifier

Envia e-mail no callback
Na biblioteca airflow.providers.smtp.notifications.smtp
Requer from_email e to

Pode 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

Use o Audit Log para monitorar o Airflow
Contém a sequência de todos os eventos da instância

Página de Audit Log do Airflow listando eventos por tipo

Vamos praticar!

Introdução ao Apache Airflow em Python