Lorsqu'un pipeline critique échoue en pleine nuit, l'équipe de Data Engineering ne doit pas attendre le lendemain matin que les utilisateurs se plaignent de tableaux de bord non rafraîchis pour réagir.
Apache Airflow intègre un moteur de notification automatique permettant d'alerter instantanément les équipes par e-mail ou via des webhooks de messagerie (Slack, Microsoft Teams, Discord) dès qu'une tâche rencontre une anomalie.
flowchart TD
TASK["Tâche en cours d'exécution"] -->|Exception / Crash / Timeout| FAIL["Statut : Failed (Échec)"]
FAIL --> ALERT_MAIL["<b>1. Alerte E-mail Native</b><br>Détail de l'erreur, extrait des logs,<br>Lien direct vers l'UI Airflow"]
FAIL --> ALERT_CALLBACK["<b>2. Callback Personnalisé</b><br>Webhook Slack / Teams / PagerDuty<br>Gestion des astreintes d'astreinte"]
style TASK fill:#f1f5f9,stroke:#64748b
style FAIL fill:#fee2e2,stroke:#ef4444,stroke-width:2px,color:#0f172a
style ALERT_MAIL fill:#fed7aa,stroke:#ea580c,color:#0f172a
style ALERT_CALLBACK fill:#dcfce7,stroke:#16a34a,color:#0f172a
1. Configurer le serveur SMTP dans Docker Compose#
Pour qu'Airflow puisse expédier des courriers électroniques, vous devez renseigner les paramètres de votre serveur SMTP d'entreprise (ou d'un fournisseur cloud comme Amazon SES, SendGrid ou Gmail).
Dans votre fichier .env (ou dans la section environment de docker-compose.yaml) :
# Activation du backend d'envoi SMTP
AIRFLOW__EMAIL__EMAIL_BACKEND=airflow.utils.email.send_email_smtp
# Paramètres de connexion SMTP
AIRFLOW__SMTP__SMTP_HOST=smtp.gmail.com
AIRFLOW__SMTP__SMTP_PORT=587
AIRFLOW__SMTP__SMTP_STARTTLS=true
AIRFLOW__SMTP__SMTP_SSL=false
AIRFLOW__SMTP__SMTP_USER=alertes.data@votreentreprise.com
AIRFLOW__SMTP__SMTP_PASSWORD=votre_mot_de_passe_application_securise
AIRFLOW__SMTP__SMTP_MAIL_FROM=Airflow Production <alertes.data@votreentreprise.com>
[!WARNING] Si vous utilisez un compte Gmail ou Microsoft 365, vous ne pouvez pas utiliser votre mot de passe de connexion habituel. Vous devez obligatoirement générer un Mot de passe d'application (App Password) depuis les paramètres de sécurité de votre compte après avoir activé la double authentification (2FA).
2. Activer les alertes automatiques en cas d'échec#
Une fois le serveur SMTP configuré, il suffit d'activer deux paramètres dans le dictionnaire default_args de votre DAG pour que toutes ses tâches soient surveillées :
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.empty import EmptyOperator
default_args = {
"owner": "algojungle",
"retries": 1,
"retry_delay": timedelta(minutes=5),
# Configuration des alertes par courriel :
"email": ["astreinte-data@votreentreprise.com", "lead-dev@votreentreprise.com"],
"email_on_failure": True, # Déclenche un e-mail dès qu'une tâche échoue définitivement
"email_on_retry": False, # N'envoie pas d'e-mail lors d'un simple réessai automatique
}
with DAG(
dag_id="pipeline_ventes_surveille",
start_date=datetime(2026, 1, 1),
schedule="@daily",
default_args=default_args,
catchup=False,
) as dag:
debut = EmptyOperator(task_id="debut")
fin = EmptyOperator(task_id="fin")
debut >> fin
Contenu de l'e-mail d'alerte généré par Airflow#
Lorsqu'un échec survient, Airflow compile automatiquement un e-mail complet contenant :
- Le nom du DAG (dag_id) et l'identifiant précis de la tâche (task_id).
- La date logique de l'exécution (logical_date).
- Le nombre de tentatives exécutées.
- Les 50 dernières lignes du journal d'erreur (Stack Trace).
- Un lien hypertexte cliquable menant directement à la page de la tâche sur l'interface web pour faciliter le diagnostic immédiat.
3. Envoyer des e-mails personnalisés avec l'EmailOperator#
En dehors des alertes d'erreurs, vous pouvez vouloir envoyer un e-mail fonctionnel contenant un récapitulatif d'activité ou un fichier CSV en pièce jointe à la fin d'un traitement :
from datetime import datetime
from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.operators.email import EmailOperator
with DAG(
dag_id="notification_rapport_quotidien",
start_date=datetime(2026, 1, 1),
schedule="@daily",
catchup=False,
) as dag:
debut = EmptyOperator(task_id="debut")
envoyer_synthese = EmailOperator(
task_id="envoyer_synthese_direction",
to=["direction-financiere@votreentreprise.com"],
subject="Rapport Quotidien d'Activité - {{ ds }}",
html_content="""
<h3>Bonjour,</h3>
<p>Le pipeline de consolidation des ventes du <b>{{ ds }}</b> s'est achevé avec succès.</p>
<p>Toutes les métriques ont été chargées dans l'entrepôt de données et sont consultables dans vos tableaux de bord Power BI.</p>
<br>
<i>Message automatique généré par Apache Airflow.</i>
""",
files=["/opt/airflow/logs/synthese_ventes.csv"], # Pièce jointe optionnelle
)
debut >> envoyer_synthese
4. Notifications modernes par Webhooks (Slack / Teams / Discord)#
De nombreuses équipes d'ingénierie préfèrent centraliser les alertes dans un canal Slack ou Teams dédié plutôt que de saturer les boîtes de réception e-mail.
Airflow propose un mécanisme de rappel (Callbacks) extrêmement puissant :
import requests
def alerte_echec_slack(context):
"""Fonction de rappel exécutée automatiquement en cas d'échec d'une tâche."""
webhook_url = "https://hooks.slack.com/services/T00/B00/XXXXX"
dag_id = context.get("task_instance").dag_id
task_id = context.get("task_instance").task_id
date_exec = context.get("ds")
url_log = context.get("task_instance").log_url
payload = {
"text": f"🚨 *Incident de Pipeline Data !*\n"
f"• *DAG :* `{dag_id}`\n"
f"• *Tâche :* `{task_id}`\n"
f"• *Date :* `{date_exec}`\n"
f"• *Consulter les logs :* <{url_log}|Cliquez ici pour investiguer>"
}
requests.post(webhook_url, json=payload, timeout=10)
Pour attacher cette fonction à votre DAG :
with DAG(
dag_id="mon_pipeline_critique",
default_args=default_args,
on_failure_callback=alerte_echec_slack, # S'active dès qu'une tâche du DAG échoue
...
):
...
Vos flux de données sont désormais sous haute surveillance. Dans la prochaine leçon, nous allons apprendre à déclarer et sécuriser des connexions de bases de données dans Airflow pour ne jamais stocker de mots de passe en clair dans votre code !
Commentaires