Automatisation CI/CD d'Apache Airflow avec exemples pratiques

Une orchestration de données fiable repose sur des déploiements reproductibles et sécurisés. Le CI/CD Apache Airflow transforme les modifications apportées aux DAGs et aux plugins en incréments livrables avec un retour rapide et un risque minimal. Dans ce guide, vous allez :

• Inventorier les versions et la topologie pour obtenir des résultats prévisibles.
• Mettre en place un petit flux CI réaliste pour linter, parser et tester les DAGs.
• Déployer avec un mécanisme de publication atomique basé sur des liens symboliques qui rend le retour arrière instantané.
• Vérifier le comportement attendu à l'aide de l'interface CLI d'Airflow.
• Diagnostiquer les modes de défaillance courants et récupérer en toute sécurité.

Une approche structurée aide les équipes à passer de changements proposés à des incréments revus et livrables. Séparer la rédaction, la validation et le déploiement réduit le retravail et accélère l'apprentissage dans les premières étapes de l'automatisation.

Inventaire des versions et de l'environnement

La cohérence commence par le verrouillage des versions, la liste des exécuteurs et la décision de la manière dont Airflow découvre vos DAGs. La matrice de versions suivante est un exemple construit que vous pouvez adapter à votre pile technique.

Prérequis :

• Dépôt Git contenant :
• dags/ pour les DAGs
• plugins/ pour les opérateurs/hooks personnalisés
• tests/ avec tests d'importation et de politiques DAG
• requirements.txt et constraints.txt pour verrouiller les dépendances
• scripts/ avec assistants de parsing, déploiement et retour arrière
• Python 3.9+ disponible localement et dans les runners CI
• Accès CLI Airflow sur les hôtes de staging et production
• Accès SSH ou mécanisme pull-from-git sur les hôtes Airflow
• Connexions et Variables Airflow documentées pour chaque environnement

Notes de topologie :

• Pointer le dags_folder d'Airflow vers un chemin stable (par exemple, /opt/airflow/dags/current) et gérer un répertoire releases/ avec liens symboliques.
• S'assurer que les utilisateurs scheduler et webserver peuvent lire depuis le dossier DAGs et tous ses sous-répertoires.

Chemin de configuration sécurisé

Commencez petit. Le premier pilote doit être étroit, mesurable et facile à inspecter localement avant le déploiement. Utilisez un DAG de fumée et quelques tests pour prouver le chemin de bout en bout.

Structure du dépôt

repo-root/
├── dags/
│   └── example_smoke_dag.py
├── plugins/
├── tests/
│   ├── test_dag_imports.py
│   └── test_dag_policies.py
├── requirements.txt
├── constraints.txt
└── scripts/
    ├── deploy_push.sh
    └── rollback.sh

Exemple de DAG de fumée

# fichier : dags/example_smoke_dag.py
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    dag_id="example_smoke_dag",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
    tags=["smoke"],
) as dag:
    echo_env = BashOperator(
        task_id="echo_env",
        bash_command="echo AIRFLOW_VERSION=$(airflow version)"
    )

Verrouiller les dépendances pour la reproductibilité

# requirements.txt
apache-airflow==2.7.3
pendulum==2.1.2

# constraints.txt
apache-airflow==2.7.3
pendulum==2.1.2

Tests légers sans scheduler

Test d'importation DAG :

# fichier : tests/test_dag_imports.py
import os
from airflow.models import DagBag

def test_no_import_errors():
    dag_folder = os.path.join(os.getcwd(), "dags")
    dag_bag = DagBag(dag_folder=dag_folder, include_examples=False)
    assert len(dag_bag.import_errors) == 0, f"Échecs d'importation DAG : {dag_bag.import_errors}"
    assert len(dag_bag.dags) > 0, "Aucun DAG découvert"

Test de politique pour empêcher les backfills incontrôlés en interdisant les start_date dynamiques :

# fichier : tests/test_dag_policies.py
import glob

def test_no_dynamic_start_date():
    for path in glob.glob("dags/**/*.py", recursive=True):
        with open(path, "r", encoding="utf-8") as f:
            src = f.read()
        assert "datetime.now(" not in src, f"start_date dynamique dans {path}"
        assert "pendulum.now(" not in src, f"start_date dynamique dans {path}"

Exécution locale des tests

python -m pip install -r requirements.txt -c constraints.txt
pytest -q
# Attendu : tests réussis, zéro erreur d'importation

Options de déploiement

• Pull-based : Les hôtes Airflow tirent depuis une branche Git ou un tag de release vers le répertoire releases/ et mettent à jour le lien symbolique current. Cela minimise les permissions CI sur la production.
• Push-based : Le CI se connecte via SSH pour rsync les DAGs/plugins vers un nouveau répertoire de release, puis bascule atomiquement le lien symbolique current.

Configurer dags_folder dans airflow.cfg sur chaque hôte :

[core]
dags_folder = /opt/airflow/dags/current

Script de déploiement push-based

# fichier : scripts/deploy_push.sh
#!/usr/bin/env bash
set -euo pipefail

AIRFLOW_HOST="${AIRFLOW_HOST:?définir AIRFLOW_HOST}"
AIRFLOW_DAGS_ROOT="/opt/airflow/dags"
RELEASES_DIR="${AIRFLOW_DAGS_ROOT}/releases"
RELEASE_ID="${1:?usage : $0 <release-id>}"
SRC_DIR="${2:-$(pwd)}"

ssh "$AIRFLOW_HOST" "mkdir -p ${RELEASES_DIR}/${RELEASE_ID}"
rsync -av --delete "${SRC_DIR}/dags/" "$AIRFLOW_HOST:${RELEASES_DIR}/${RELEASE_ID}/"
rsync -av --delete "${SRC_DIR}/plugins/" "$AIRFLOW_HOST:${RELEASES_DIR}/${RELEASE_ID}/plugins/" || true

ssh "$AIRFLOW_HOST" "ln -sfn ${RELEASES_DIR}/${RELEASE_ID} ${AIRFLOW_DAGS_ROOT}/current && ls -l ${AIRFLOW_DAGS_ROOT}"

Script de retour arrière

# fichier : scripts/rollback.sh
#!/usr/bin/env bash
set -euo pipefail

AIRFLOW_HOST="${AIRFLOW_HOST:?définir AIRFLOW_HOST}"
AIRFLOW_DAGS_ROOT="/opt/airflow/dags"
PREV_RELEASE_ID="${1:?usage : $0 <previous-release-id>}"

ssh "$AIRFLOW_HOST" "ln -sfn ${AIRFLOW_DAGS_ROOT}/releases/${PREV_RELEASE_ID} ${AIRFLOW_DAGS_ROOT}/current && ls -l ${AIRFLOW_DAGS_ROOT}"

Vérification et diagnostics

Contrôles pré-déploiement (local ou CI)

• Installer les dépendances avec contraintes
• Linter et tests unitaires
• Parser les DAGs avec DagBag pour assurer zéro erreur d'importation
• Optionnel : dry-run d'une tâche avec tasks test

Exemples de commandes et sorties attendues

python -m pip install -r requirements.txt -c constraints.txt
pytest -q
# Attendu : "2 passed" (ou plus), et aucune import_errors

# Test de tâche optionnel sans créer d'exécution DAG
airflow tasks test example_smoke_dag echo_env 2024-01-01
# Dernière ligne attendue : "Task exited with return code 0" ou état SUCCESS

Contrôles post-déploiement sur staging ou production

# Découverte DAG
airflow dags list | grep example_smoke_dag
# Attendu : une ligne montrant example_smoke_dag

# Liste des tâches
airflow tasks list example_smoke_dag
# Attendu : echo_env listé

# Déclenchement de fumée (manuel)
RUN_ID="ci-smoke-$(date +%s)"
airflow dags trigger -r "$RUN_ID" example_smoke_dag
sleep 5

# Observer les exécutions
airflow dags list-runs -d example_smoke_dag | head -n 5
# Attendu : exécution la plus récente avec état running/success

# Inspecter le log d'instance de tâche (ajuster la date d'exécution si nécessaire)
# Pour les exécutions manuelles, utiliser l'UI ou list-runs pour trouver la date d'exécution.

Ce qu'il faut observer

• Le DAG apparaît dans la liste dans les 60 secondes suivant le déploiement.
• La tâche de fumée réussit et les logs incluent un écho de version Airflow.
• Les logs du scheduler montrent zéro erreur d'importation et aucune erreur de permission lors du balayage du répertoire DAG.

Modes de défaillance et récupération

Utilisez le tableau ci-dessous pour relier les symptômes aux causes probables, diagnostics et correctifs sûrs. Les éléments sont des exemples construits que vous pouvez adapter.

Modèles de récupération

• Releases atomiques : Conserver N releases historiques dans /opt/airflow/dags/releases/<id>. Le lien symbolique current pointe vers l'actif.
• Retour arrière instantané : Repointer le lien symbolique current vers la release précédente. Aucun copier de fichier n'est nécessaire.
• Isolation des dépendances : Si vous installez les packages Python au niveau système, le retour arrière inclut la restauration des contraintes précédentes. Préférez un virtualenv par release lorsque c'est faisable.
• Nettoyer les exécutions non intentionnelles : Si un DAG a créé des backfills excessifs, le désactiver, définir catchup=False, et nettoyer les instances de tâche indésirables après le retour arrière.

Liste de contrôle opérationnelle

Avant fusion

• Confirmer que le DAG respecte les politiques d'équipe (start_date statique, retries bornés, tags appropriés)
• Ajouter ou mettre à jour les tests unitaires pour les opérateurs personnalisés et le parsing DAG
• Passer les tests d'importation et de politique localement

Avant déploiement (CI)

• Installer avec contraintes : pip install -r requirements.txt -c constraints.txt
• Exécuter pytest -q et assurer zéro erreur d'importation
• Optionnel : airflow tasks test <dag> <task> <date> pour les tâches critiques
• Produire un identifiant de release (timestamp ou SHA de commit)

Déploiement

• Créer le répertoire de release sur l'hôte cible : /opt/airflow/dags/releases/<id>
• Synchroniser dags/ et plugins/ vers le répertoire de release
• Basculer atomiquement ln -sfn le lien symbolique current vers la nouvelle release

Validation post-déploiement

• airflow dags list | grep <votre_dag> montre le DAG
• airflow tasks list <votre_dag> liste les tâches attendues
• Déclencher le DAG de fumée et vérifier le succès dans le SLA
• Scanner les logs du scheduler pour erreurs d'importation ou de permission

Retour arrière (si validation échoue)

• Repointer current vers la release précédente : rollback.sh
• Confirmer que la liste DAG montre les versions précédentes
• Réexécuter les vérifications de fumée pour confirmer la santé
• Ouvrir un ticket avec diagnostics et reproduction minimale

Revue et amélioration continue

• Capturer le temps moyen de détection (MTTD) et de récupération (MTTR)
• Ajouter des tests de politique pour toute nouvelle classe d'incident rencontrée
• Étendre les DAGs de fumée pour inclure vérifications de Connexions et Variables critiques

Exemples pratiques en contexte

Flux de bout en bout pour un petit changement

1. Le développeur modifie example_smoke_dag.py pour ajouter une seconde tâche Bash qui affiche le hostname.
2. Exécute localement :

• pip install -r requirements.txt -c constraints.txt
• pytest -q → tous les tests passent
• airflow tasks test example_smoke_dag echo_env 2024-01-01 → SUCCESS

1. Ouvre une demande de changement ; le CI exécute les mêmes tests et produit un ID de release comme 20240115-abcdef.
2. Après approbation, le CI exécute deploy_push.sh 20240115-abcdef vers le staging.
3. Sur l'hôte de staging :

• airflow dags list | grep example_smoke_dag → visible
• airflow tasks list example_smoke_dag → montre echo_env et la nouvelle tâche
• Déclencher et observer les logs → les deux tâches réussissent

1. Promouvoir vers la production avec le même ID de release ; valider ; surveiller pendant 30 minutes.
2. Si un problème survient, exécuter rollback.sh 20240110-123456 pour revenir instantanément.

Conclusion

Vous disposez maintenant d'une approche pratique et défendable pour le CI/CD Apache Airflow. En inventorant les versions et la topologie, vous assurez que les changements se comportent de la même manière dans tous les environnements. Des tests rapides et déterministes capturent les erreurs d'importation et les violations de politique avant qu'elles n'atteignent la production. Les déploiements atomiques avec un répertoire de releases et un lien symbolique current vous donnent un retour arrière instantané sans copier de fichiers. La vérification avec l'interface CLI Airflow et un DAG de fumée simple confirme la santé de l'environnement de bout en bout. Enfin, une carte claire des modes de défaillance courants — DAGs manquants, erreurs d'importation, backfills incontrôlés, échecs de tâches spécifiques à l'environnement et état obsolète du scheduler — vous permet de diagnostiquer et récupérer rapidement. Un pilote étroit et inspectable construit la confiance vite et crée un motif que vous pouvez étendre à l'ensemble de votre domaine d'orchestration.