Oozie → Airflow : ce qui ne se traduit pas

Migrer d'Oozie vers Airflow est présenté comme un portage : on traduit du XML en Python, on rebranche les jobs, on éteint l'ancien. Dans un contexte où les traitements ont une échéance réglementaire, cette lecture coûte cher. Les deux outils n'ont pas le même modèle de déclenchement, et c'est ce modèle — pas la syntaxe — qui décide si une chaîne arrive à l'heure.

Deux modèles mentaux différents

Un coordinator Oozie est fondamentalement orienté donnée. On lui déclare des jeux de données avec leur emplacement et leur périodicité, et il déclenche le workflow quand les instances d'entrée existent :

<coordinator-app name="declaratif" frequency="${coord:days(1)}" ...>
  <datasets>
    <dataset name="contrats" frequency="${coord:days(1)}">
      <uri-template>/data/contrats/${YEAR}${MONTH}${DAY}</uri-template>
      <done-flag>_SUCCESS</done-flag>
    </dataset>
  </datasets>
  <input-events>
    <data-in name="in" dataset="contrats">
      <instance>${coord:current(0)}</instance>
    </data-in>
  </input-events>
  ...
</coordinator-app>

Le fichier _SUCCESS est le contrat. Tant qu'il n'est pas là, rien ne part, et le coordinator reste en attente aussi longtemps qu'il le faut.

Un DAG Airflow, lui, est orienté temps. Il a une schedule, et à l'heure dite il démarre. La disponibilité de la donnée n'est pas une condition de déclenchement : c'est une condition qu'il faut réintroduire explicitement, avec un capteur.

with DAG("declaratif", schedule="0 1 * * *", catchup=False,
         default_args={"retries": 2}) as dag:

    attend_contrats = FileSensor(
        task_id="attend_contrats",
        filepath="/data/contrats/{{ ds_nodash }}/_SUCCESS",
        poke_interval=120,
        timeout=60 * 60 * 3,      # au-delà, on veut une alerte, pas une attente
        mode="reschedule",        # libère le worker entre deux vérifications
    )

Trois paramètres portent tout le sens. timeout transforme une attente infinie en incident visible — c'est ce qui manque le plus souvent aux premières migrations. mode en reschedule évite qu'un capteur monopolise un slot de worker pendant trois heures, ce qui est la cause numéro un de blocage d'un ordonnanceur fraîchement migré. Et poke_interval se règle sur la granularité réelle du dépôt, pas sur la nervosité de celui qui écrit le DAG.

Le piège du catchup

Airflow raisonne en intervalles. Un DAG quotidien exécuté le 12 août traite l'intervalle du 11. Si vous déployez un DAG avec start_date au 1er janvier et catchup=True — la valeur historique par défaut —, Airflow ordonnancera immédiatement toutes les exécutions manquantes depuis cette date.

Sur une chaîne de calcul, ça sature le cluster. Sur une chaîne de déclaratif, c'est pire : on régénère et on redépose des fichiers pour des périodes déjà closes.

Sur un périmètre réglementaire, catchup=False n'est pas une préférence, c'est une mesure de sécurité. Le rattrapage doit être une action délibérée, pas un effet de bord du déploiement.

Le corollaire : puisque le rattrapage devient manuel, il faut l'outiller. Une commande de reprise documentée, testée, et bornée à une plage de dates — pas un airflow dags backfill tapé dans l'urgence à deux heures du matin.

Ce qu'il faut réécrire à la main

Voici la liste de ce qui ne se traduit pas mécaniquement, dans l'ordre où ça mord.

• Les dépendances entre coordinators. Oozie les exprime par des jeux de données partagés. Airflow demande de choisir : un capteur sur la table de sortie, un ExternalTaskSensor sur l'autre DAG, ou une fusion des deux DAG. Chacun a des propriétés de reprise différentes, et c'est une décision d'architecture, pas de traduction.
• La sémantique des dates. Les fonctions ${coord:current(-1)} deviennent {{ data_interval_start }} — mais le décalage d'un intervalle est une source d'erreur systématique. Il faut vérifier chaque partition écrite, pas seulement chaque expression traduite.
• Les relances. Un workflow Oozie relancé repart de la première action non terminée. Airflow relance au niveau de la tâche. Une tâche qui écrit sans être idempotente produira des doublons dès la première relance — et la première relance arrive toujours.
• Les SLA. Oozie ne les porte pas nativement, Airflow oui. C'est l'occasion de les rendre explicites : une échéance légale se déclare dans le DAG, pas dans un tableur.

La stratégie de bascule : deux chaînes en parallèle

La seule méthode que je recommande pour un périmètre critique est le double run comparé. Les deux orchestrations produisent leur sortie sur la même période, et un job de comparaison bloque la bascule tant qu'une seule ligne diffère.

val ancien  = spark.read.parquet(s"/prod/oozie/declaratif/$jour")
val nouveau = spark.read.parquet(s"/prod/airflow/declaratif/$jour")

val cles = Seq("id_contrat", "id_personne")

val ecarts = ancien.as("a")
  .join(nouveau.as("n"), cles, "full_outer")
  .filter(cles.map(k => col(s"a.$k").isNull || col(s"n.$k").isNull)
              .reduce(_ || _) || col("a.montant") =!= col("n.montant"))

require(ecarts.isEmpty, s"${ecarts.count()} écarts entre les deux chaînes")

La jointure full_outer est essentielle : une jointure interne ne verrait ni les lignes apparues, ni les lignes disparues — c'est-à-dire précisément la classe de régression qu'on cherche.

Le critère de bascule doit être un nombre de jours consécutifs sans écart, pas une impression. Sur un déclaratif à cadence mensuelle, ça implique de simuler des journées passées plutôt que d'attendre des mois.

Ce qu'on gagne vraiment

Le gain habituellement mis en avant est le passage de XML à Python. Il est réel mais secondaire. Les trois gains qui comptent sur la durée sont ailleurs.

Les dépendances deviennent lisibles. Une vue de graphe qui montre où une chaîne est bloquée remplace la lecture croisée de plusieurs fichiers de coordination.

Les tâches deviennent testables. Un DAG est du code Python : on peut instancier le graphe dans un test, vérifier qu'il n'a pas de cycle, que chaque tâche critique porte bien un timeout et un SLA. Ce test-là attrape des régressions d'ordonnancement avant la production.

def test_les_taches_critiques_ont_un_timeout():
    dag = DagBag().get_dag("declaratif")
    for nom in ("attend_contrats", "depot"):
        assert dag.get_task(nom).execution_timeout is not None

L'observabilité devient native. Durée par tâche, historique, alerte sur dépassement : ce sont les données qui permettent de dire « ce traitement a gagné une heure » avec un graphique plutôt qu'avec un souvenir.

Si vous ne deviez retenir qu'une règle de cette migration : ne portez pas les coordinators, réexprimez les contrats de données. Le XML est le symptôme ; le modèle de déclenchement est le sujet.