Pipelines de données idempotents : écrasement de partitions, réexécutions sûres et rattrapages sans double comptage

Objectif

Rendre chaque tâche planifiée sûre à exécuter deux fois, de sorte que les nouveaux essais, les réexécutions manuelles et les rattrapages historiques ne dupliquent ni ne perdent jamais de lignes, et que « retraiter le mois dernier » soit une commande plutôt qu'un projet.

Prérequis

Un ordonnanceur qui associe un intervalle de données à chaque exécution. La documentation d'Airflow (citée) définit l'intervalle comme la plage temporelle sur laquelle porte une exécution, planifie une exécution après la fin de son intervalle afin que toutes les données de la période puissent être collectées, et appelle « backfill » (rattrapage) le fait d'exécuter un Dag pour une période historique donnée. Sa page de bonnes pratiques indique que, les tâches pouvant être réessayées, elles devraient produire le même résultat à chaque réexécution ; elle déconseille un simple INSERT à la réexécution (lignes dupliquées), recommande UPSERT, et préconise de lire et d'écrire dans une partition précise plutôt que dans « les dernières données disponibles ».

Étapes

1. Paramétrer chaque tâche par l'intervalle (data_interval_start, data_interval_end) ; jamais par « maintenant ». La sélection des entrées filtre sur l'intervalle, de sorte qu'une réexécution voit la même entrée, sauf si la source elle-même a changé.
2. Écrire la sortie comme un remplacement de la tranche de l'intervalle : suppression puis insertion au sein d'une même transaction dans une cible relationnelle, ou écrasement de partition dans une cible fondée sur des fichiers. Le INSERT OVERWRITE de Spark avec partitionOverwriteMode=dynamic n'écrase que les partitions qui reçoivent des données pendant l'exécution (cité) ; en mode statique, il supprime d'abord toute partition correspondant à la spécification de partitionnement.
3. Là où le remplacement est impossible (tables d'événements partagées avec d'autres écrivains), effectuer un upsert sur une clé déterministe dérivée de l'enregistrement, par exemple l'identifiant source associé à l'intervalle, de sorte qu'une réexécution mette à jour plutôt qu'elle n'ajoute.
4. Faire recalculer les agrégats en aval à partir de la tranche remplacée plutôt qu'en ajoutant des deltas ; un agrégat qui additionne des incréments comptera deux fois une partition rattrapée.
5. Pour un rattrapage, exécuter les intervalles dans l'ordre avec une concurrence bornée, et après chaque intervalle comparer le nombre de lignes et une somme de contrôle des colonnes clés avec la version précédente de cette partition ; journaliser l'écart.
6. Réexécuter les tâches dépendantes pour les mêmes intervalles ; rattraper une table sans ses consommateurs laisse l'entrepôt de données incohérent en interne.

Résultat attendu

Réexécuter n'importe quel intervalle laisse la sortie identique à celle d'une exécution propre unique. Rattraper une plage produit les mêmes tables que si le code corrigé avait tourné selon le calendrier prévu.

Limites et base de vérification

Les sources qui écrasent leur historique (tables opérationnelles muables) brisent le déterminisme des intervalles ; prendre un instantané par intervalle ou recourir à la capture de changement de données (CDC). Un rattrapage qui traverse un changement de schéma nécessite le nouveau schéma pour tous les intervalles. La méthode est un protocole dérivé de la documentation citée ; aucune mesure de son effet n'est revendiquée.