ZrobDev/pipelines-ETL

Pipelines ETL

Ingestion par lots dans PostgreSQL (magasin par défaut) avec Python (uv, ruff, pytest). Transformations locales : polars 2.x ou DuckDB 2.0.

Procédure

1. Énoncer le grain : une ligne signifie quoi, identifiée par quelle clé naturelle. Pas de clé, pas de chargement idempotent ; définissez-en une (ou un hash de contenu) d'abord.
2. Choisir le modèle de chargement (ci-dessous). Choisir la stratégie incrémentale. Choisir l'orchestrateur en dernier.
3. Écrire l'exécution pour que toute ré-exécution d'une fenêtre donne le même état final. Le prouver : l'exécuter deux fois dans un test et vérifier que les comptes de lignes et une somme de contrôle restent inchangés.
4. Ajouter les vérifications d'observabilité (ci-dessous) avant la première exécution en production.
5. Passer en revue la checklist à la fin.

Modèle de chargement idempotent

• Les lignes ont une clé naturelle stable et peuvent changer : INSERT ... ON CONFLICT (key) DO UPDATE (Postgres) ou MERGE. Ajouter WHERE target.hash IS DISTINCT FROM excluded.hash pour que les lignes inchangées ne s'agite pas.
• Les données arrivent en une partition entière (jour, identifiant de lot) et sont réécrites en une unité : delete-insert de cette partition dans UNE transaction. Un delete validé séparément de l'insert laisse un trou en cas de crash.
• La table entière est reconstruite : charger dans une table de staging, valider, puis échanger (renommer dans une transaction). Les lecteurs ne voient jamais une table à moitié chargée.
• Faits en append-only sans clé : ajouter un batch_id / hash de fichier source déterministe et faire en sorte que le chargement refuse ou remplace un batch_id déjà chargé.
• Ne jamais faire un INSERT aveugle puis dédoublonner plus tard. Les doublons atteignent un tableau de bord avant que le dédoublonnage ne s'exécute.
• Charger via COPY dans une table temporaire/staging, puis un upsert basé sur un ensemble. Les upserts ligne par ligne depuis Python sont 10 à 100 fois plus lents.

Chargements incrémentaux

• Watermark sur une colonne dont la source garantit la monotonie et l'indexation (updated_at, identifiant de séquence). La stocker dans une table d'état, écrite dans la même transaction que le chargement, jamais dans un fichier à part.
• Lire watermark - lookback (par ex. 1 à 24 h), pas > watermark. Les commits source ne sont pas ordonnés par horodatage ; le lookback plus l'upsert absorbe les lignes tardives. Dimensionner le lookback en fonction de la latence mesurée, pas d'une estimation.
• Colonne de watermark non fiable (non mise à jour lors des mises à jour, fuseau horaire mélangé, suppressions logiques manquantes) : faire une réconciliation complète périodique (différence d'ensemble de clés ou count + checksum par partition) au lieu de lui faire confiance.
• Les suppressions physiques à la source sont invisibles pour les watermarks. Décider explicitement : réconcilier par différence de clés, ou accepter et documenter.
• CDC (réplication logique Postgres, lecteurs de logs de style Debezium) : utiliser uniquement lorsque vous avez besoin de suppressions, d'une latence inférieure à la minute, ou que la source ne peut pas être interrogée par horodatage. Cela ajoute des slots, un risque de rétention (un slot inactif remplit le disque source) et une gestion de l'ordre/snapshot. Sinon, restez avec les watermarks.

Backfills

• Chaque job prend --start et --end (demi-ouvert [start, end), UTC) et en dérive tout. Pas de now() dans la transformation.
• Un backfill est l'exécution normale du job sur d'anciennes fenêtres. S'il a besoin d'un code différent, le job normal n'est pas idempotent ; corrigez-le.
• Découper par partition, exécuter les morceaux indépendamment, enregistrer le statut par morceau, de sorte qu'un échec au morceau 40 reprenne à 40.
• Backfill dans la même cible via le même upsert/remplacement de partition. Vérifier l'absence de double comptage : count(*) et count(distinct key) par fenêtre avant et après.
• Limiter la charge contre l'API/BD source ; un backfill est un pic de charge auto-infligé.

Retries, timeouts, « exactly once »

• La livraison exactly-once n'existe pas sur un réseau. Construisez une livraison at-least-once plus des écritures idempotentes ; le résultat est effectivement once.
• Retry uniquement les étapes idempotentes, avec backoff exponentiel et jitter, un plafond de tentatives, et un timeout strict sur chaque appel réseau (connect et read définis séparément). Retry sur 429/5xx/timeouts, pas sur les 4xx.
• Honorer Retry-After. Les retries sans limite de débit globale amplifient une panne.
• Une étape qui envoie un email, facture, ou publie sur un webhook n'est pas idempotente : protégez-la avec une clé d'idempotence stockée avant l'appel.
• En cas d'échec final, quitter avec un code non nul et alerter ; ne jamais avaler l'exception pour garder un planning vert.

Dérive de schéma

• La couche d'arrivée est permissive (stocker la charge utile brute en JSON/texte plus les métadonnées de chargement). La couche typée est stricte et échoue bruyamment en cas de violation de type/champ obligatoire.
• Nouvelles colonnes : additives et autorisées avec une alerte. Colonnes supprimées, renommées ou retypées : faire échouer l'exécution et afficher la différence ; ne pas forcer à NULL.
• Valider avec un schéma explicite (schéma polars, pydantic, ou DuckDB DESCRIBE) à la frontière, une fois, pas dispersé dans la transformation.

Scrape vers entrepôt

• Deux étapes. L'étape 1 récupère et stocke la réponse brute inchangée et immuable (chemin d'objet indexé par source, heure de récupération, hash de contenu). L'étape 2 analyse les données brutes en tables.
• Les bugs de parsing sont ensuite corrigés en ré-analysant les données brutes stockées, sans nouvelle récupération, sans risque de bannissement, et sans perte d'historique.
• Enregistrer l'URL, l'horodatage de récupération, le statut HTTP, la version du parseur sur chaque ligne analysée. L'analyse est une fonction pure des octets bruts et de la version du parseur.
• Les échecs de récupération et les pages vides/bloquées (captcha, 200 avec un mur de connexion) sont détectés par des vérifications de contenu, pas uniquement par le code de statut.

Observabilité (minimum)

• Journaliser par étape : lignes lues, lignes écrites, lignes rejetées, durée, début/fin de fenêtre, identifiant d'exécution. Lignes structurées (JSON).
• Affirmer les invariants : lignes sorties par rapport aux lignes entrées dans une tolérance ; zéro ligne sur une fenêtre normalement non vide FAIT ÉCHOUER l'exécution. Un succès silencieux de zéro ligne est l'échec non détecté le plus courant.
• Heartbeat : envoyer un signal à un healthcheck externe en cas de succès et au démarrage ; alerter sur un signal manqué (capture le cron ne se déclenchant pas, hôte down). Un échec envoie également un signal au point de terminaison d'échec.
• Alerter sur la fraîcheur (horodatage max chargé plus ancien que le seuil), pas seulement sur les erreurs de job.
• Conserver une table de journal d'exécution (identifiant d'exécution, fenêtre, statut, comptes) ; c'est aussi votre registre de backfill.

Choisir l'orchestrateur

• Cron (ou timer systemd) + un script : un à quelques jobs indépendants, pas de dépendances inter-jobs, une équipe qui peut lire les logs. Par défaut. Nécessite le heartbeat ci-dessus et un verrou (flock) contre les exécutions simultanées.
• n8n : faible volume, colle entre API SaaS, déclenché par webhook, les non-développeurs modifient les flux. Mauvais choix pour les gros lots, les transformations lourdes ou la revue de code ; gardez la logique lourde dans un script Python qu'il appelle.
• Airflow 3.x : nombreux DAGs, dépendances inter-jobs, backfills de première classe, retries et SLA, UI d'audit. Coût opérationnel réel (scheduler, BD de métadonnées, workers) ; pas rentable pour une poignée de jobs. La migration d'Airflow 2 vers 3 comporte des changements cassants ; vérifier la version avant de copier un extrait. Pour dbt sur Airflow, utilisez Astronomer Cosmos.
• dlt : ingestion d'API ou de base de données où vous souhaitez une inférence de schéma, un état incrémental et des dispositions de merge/replace gérés pour vous. Utilisez-le au lieu d'écrire à la main du code d'extraction paginée + watermark + upsert ; appliquez toujours les règles ci-dessus à sa configuration.
• Conteneurs : secrets sous forme de secrets Compose lus depuis /run/secrets/, jamais de variables d'environnement ou env_file ; lier les ports publiés à 127.0.0.1 ou les garder sur un réseau interne.

Compétences fournisseur

Lorsque la tâche nécessite des spécificités dbt ou Airflow (modèles, tests, création de DAG, opérateurs, configuration Cosmos), n'improvisez pas : utilisez les compétences fournisseur officielles importées via la compétence de-vendors de ce dépôt, et utilisez la version dans references/landscape-2026-11.md uniquement pour vérifier ce qui est actuel.

Checklist de revue

• Clé et grain énoncés ; la ré-exécution de la même fenêtre donne un état identique (testé).
• Suppression et insertion dans une transaction ; watermark validé avec le chargement.
• Lookback présent ; les données tardives et les suppressions ont une réponse explicite.
• Timeouts, retries limités avec backoff, étapes non idempotentes protégées.
• Arrivée brute immuable ; analyse séparable et ré-exécutable.
• Comptes de lignes par étape journalisés ; alertes de zéro ligne et de fraîcheur existantes ; heartbeat configuré.
• La dérive de schéma échoue bruyamment dans la couche typée.
• Choix de l'orchestrateur justifié par le nombre de jobs et les dépendances, pas par habitude.

Squelettes SQL et Python concrets : references/patterns.md. Leçons tirées de l'utilisation réelle : references/gotchas.md.