Surveillance, alertes et callbacks

Introduction à Apache Airflow en Python

Mike Metzger

Data Engineer

Cycle de vie d’un DAG

 

  • Chaque exécution de DAG et chaque tâche suivent une suite d’états
    • En file d’attente
    • En cours d’exécution
    • État final : Réussite, Échec, Ignoré
  • Airflow suit les états et transitions

Schéma du cycle de vie d’un DAG : en file d’attente, en cours d’exécution, puis état final

Introduction à Apache Airflow en Python

Callbacks

  • Fonctions sur un DAG ou une tâche appelées automatiquement par Airflow
  • Callbacks pour des transitions d’état spécifiques
    • on_failure_callback - Échec du DAG / de la tâche
    • on_success_callback - Réussite du DAG / de la tâche
  • Callbacks spécifiques aux tâches
    • on_retry_callback
    • on_skipped_callback
    • on_execute_callback

Illustration des callbacks Airflow déclenchés lors des transitions d’état

Introduction à Apache Airflow en Python

Contexte de callback

  • Airflow passe automatiquement un dictionnaire context
  • context contient des infos sur le DAG / la tâche
    • context["dag"].dag_id - Nom du DAG
    • context["task_instance"].task_id - Nom de la tâche
    • context["logical_date"] - Date de l’exécution du DAG
  • La fonction de callback doit accepter 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.")
Introduction à Apache Airflow en Python

Exemple 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.
Introduction à Apache Airflow en Python

Notifiers

  • Fonctions liées à des callbacks
  • Envoient des alertes à des systèmes externes
  • Plusieurs notifiers disponibles :
    • SmtpNotifier - Envoie des alertes par e-mail
    • SlackNotifier - Publie des messages sur un canal Slack
    • Notifiers spécifiques - PagerDuty, OpsGenie

Illustration des notifiers Airflow envoyant des alertes à des systèmes externes

Introduction à Apache Airflow en Python

SmtpNotifier

  • Envoie un e-mail lors d’un callback
  • Dans la bibliothèque airflow.providers.smtp.notifications.smtp
  • Nécessite from_email et to
  • Peut inclure 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'
   )
)
Introduction à Apache Airflow en Python

Audit Log

  • Utilisez l’Audit Log pour surveiller Airflow
  • Contient la séquence de tous les événements de l’instance Airflow

Page du journal d’audit Airflow listant les événements système par type

Introduction à Apache Airflow en Python

Passons à la pratique !

Introduction à Apache Airflow en Python

Preparing Video For Download...