Tubes et filtres, et architectures en pipeline

Définition et origine

Dans le style tubes et filtres (pipes and filters), le traitement est une chaîne d’étapes indépendantes. Chaque filtre lit des données sur son entrée, effectue un seul travail sur elles et écrit le résultat sur sa sortie. Chaque tube (pipe) transporte les données dans un seul sens d’un filtre au suivant, généralement avec un tampon entre les deux. La description de Garlan et Shaw (1994) énonce les contraintes qui donnent au style ses propriétés : les filtres doivent être indépendants, ne doivent pas partager d’état avec d’autres filtres et ne doivent pas connaître l’identité des filtres situés à chaque extrémité de leurs tubes. Un filtre ne connaît que le format de ce qu’il lit et de ce qu’il écrit.

Le terrain d’élection le plus connu du style est Unix. Dans une note datée du 11 octobre 1964, Doug McIlroy écrivait qu’il devrait exister des moyens de raccorder les programmes comme un tuyau d’arrosage, en vissant un autre segment quand les données doivent être retravaillées d’une autre manière. Ken Thompson a implémenté l’appel système pipe et les tubes dans le shell de la Version 3 d’Unix en 1973, et la notation | a suivi dans la Version 4. La philosophie Unix qui en est née, écrire des programmes qui font une seule chose et la font bien, et qui collaborent par une interface texte universelle, est le style tubes et filtres appliqué à l’ensemble des outils d’un système d’exploitation. POSA (1996) a catalogué Pipes and Filters comme patron architectural, et Enterprise Integration Patterns (2003) de Hohpe et Woolf l’a appliqué à la messagerie entre systèmes.

Richards et Ford appellent ce même style l’architecture en pipeline, et nomment quatre sortes de filtres. Un producteur (ou source) démarre le flux, en lisant dans un fichier, une base de données ou une file. Un transformateur modifie chaque enregistrement : analyse syntaxique, conversion, enrichissement. Un testeur vérifie une condition et transmet l’enregistrement ou l’écarte, parfois en l’aiguillant vers l’une de plusieurs sorties. Un consommateur (ou puits) termine le flux, en écrivant dans une base de données, un fichier ou sur un écran.

Comment ça marche

Les contraintes produisent directement les avantages. Comme un filtre ne partage aucun état et ne connaît pas ses voisins, il peut être écrit et testé isolément, avec une liste d’enregistrements en entrée et une liste de sorties attendues ; il peut être réutilisé dans d’autres pipelines ; et les filtres peuvent être ajoutés, retirés ou réordonnés en modifiant uniquement la façon dont le pipeline est assemblé. Comme les enregistrements circulent de façon incrémentale, un filtre peut commencer à traiter le premier enregistrement avant que le filtre précédent ait fini le dernier : les étapes s’exécutent donc de façon concurrente, et l’ensemble du pipeline n’a besoin de mémoire que pour quelques enregistrements à la fois, et non pour tout le jeu de données.

Le tube masque la façon dont les données circulent. Dans un shell Unix, c’est un tampon du système d’exploitation entre deux processus ; sous Linux, il contient 64 Kio par défaut, et un écrivain qui le remplit se bloque simplement jusqu’à ce que le lecteur rattrape son retard. Dans un programme Python, ce peut être un générateur, de sorte que chaque filtre tire l’enregistrement suivant du précédent. Dans une plateforme de données, ce peut être une file de messages ou un topic dans un journal comme Apache Kafka, ce qui permet aux filtres de s’exécuter sur des machines différentes et de survivre aux redémarrages. Le code des filtres change à peine d’une solution à l’autre ; les propriétés opérationnelles, elles, changent beaucoup.

La forme stricte connaît des variantes. Un pipeline peut se ramifier (fan-out : un testeur aiguille les enregistrements vers deux consommateurs différents) ou converger (fan-in : deux producteurs alimentent un même transformateur). Il peut être linéaire, ce qui est le plus simple à raisonner, ou former un graphe orienté acyclique d’étapes, comme dans les outils de workflow tels qu’Apache Airflow. À strictement parler, un filtre doté d’un état interne, par exemple un filtre qui compte ou qui trie, rompt la propriété incrémentale, car il doit voir toute l’entrée avant de produire une sortie ; sort dans un pipeline Unix est un tel filtre. Ces filtres sont autorisés, mais ils deviennent les points où la mémoire, la latence et les redémarrages demandent réflexion.

Exemples : où l’on trouve des pipelines

ETL et pipelines de données. Les traitements Extract, transform, load (extraire, transformer, charger) déplacent les données des systèmes opérationnels vers un entrepôt de données : extraire les commandes de la veille, les nettoyer et les convertir, les joindre aux données des restaurants, les charger dans des tables de reporting. Chaque étape est un filtre, et les plateformes de données modernes (Apache Beam, Spark, modèles dbt, services cloud de flux de données) sont en grande partie des outils pour construire et exécuter de tels pipelines à grande échelle.

Compilateurs. Un compilateur classique est un pipeline de phases : l’analyse lexicale transforme des caractères en lexèmes (tokens), l’analyse syntaxique transforme les lexèmes en arbre syntaxique, l’analyse sémantique vérifie les types et les noms, l’optimisation réécrit une représentation intermédiaire, et la génération de code émet du code machine. Chaque phase consomme une représentation et produit la suivante, si bien que les phases peuvent être développées et remplacées séparément ; le projet LLVM, par exemple, permet à de nombreux frontends de langages de partager un même optimiseur et un même ensemble de backends. Les vrais compilateurs assouplissent le style strict, car les phases partagent une table des symboles.

Traitement multimédia. Le transcodage vidéo et le traitement audio enchaînent décodeurs, redimensionneurs, filtres et encodeurs. GStreamer construit explicitement les applications multimédias comme des pipelines d’éléments reliés par des pads, et les graphes de filtres de FFmpeg fonctionnent de la même façon.

Traitement des journaux et des événements. Les expéditeurs de journaux comme Logstash et Fluent Bit se configurent en étapes d’entrée, de filtrage et de sortie : lire les lignes de journal, les découper en champs, écarter le bruit des contrôles de santé, masquer les données personnelles et les envoyer vers un index de recherche. Les données d’observabilité de la plateforme de livraison de repas circulent de cette façon.

Pipelines de caractéristiques et d’inférence en apprentissage automatique. La préparation des données d’entraînement, ainsi que le prétraitement et le post-traitement autour d’un modèle, sont couramment construits comme des pipelines, auxquels revient le chapitre 12.

# Top ten URLs that returned HTTP 500 in today's access log
grep '" 500 ' access.log \
  | awk '{print $7}' \
  | sort \
  | uniq -c \
  | sort -rn \
  | head -10

import json

def read_lines(path):                 # producer
    with open(path) as f:
        for line in f:
            yield line

def parse(lines):                     # transformer
    for line in lines:
        yield json.loads(line)

def only_failed(events):              # tester
    for e in events:
        if e["status"] == "payment_failed":
            yield e

def count_by_provider(events):        # consumer
    counts = {}
    for e in events:
        counts[e["provider"]] = counts.get(e["provider"], 0) + 1
    return counts

events = only_failed(parse(read_lines("orders.jsonl")))
print(count_by_provider(events))

Batch et streaming

La même structure s’exécute selon deux modes. Le traitement par lots traite une entrée bornée, comme toutes les commandes de la veille, s’exécute jusqu’au bout et s’arrête. Il est simple à raisonner : si une exécution échoue, corrigez le problème et relancez-la, à condition que chaque exécution remplace sa sortie au lieu de s’y ajouter. Les versements nocturnes aux restaurants, les factures mensuelles et les chargements de l’entrepôt de données sont des pipelines batch typiques. Leur faiblesse est le délai : les résultats ont l’âge de la dernière exécution.

Le traitement de flux (stream processing) traite une entrée non bornée, en traitant chaque enregistrement peu après son arrivée : positions des livreurs, événements de paiement, clics. Les résultats sont frais, mais de nouvelles questions apparaissent. Un flux ne se termine jamais, donc les agrégats doivent être calculés sur des fenêtres (commandes par restaurant par tranche de cinq minutes) ; des enregistrements peuvent arriver en retard ou dans le désordre ; et le pipeline doit fonctionner en continu, se remettre des plantages sans perdre ni compter deux fois des enregistrements, et faire face aux pics de débit en entrée. Des frameworks comme Kafka Streams, Apache Flink et Apache Beam fournissent des fenêtres, des points de reprise (checkpoints) et des magasins d’état pour ces problèmes. Le modèle de Beam traite le batch comme un cas particulier du streaming, avec une entrée bornée, si bien qu’une même définition de pipeline peut s’exécuter dans les deux modes.

Beaucoup d’organisations utilisent les deux : un flux pour des tableaux de bord opérationnels frais mais approximatifs, et un traitement batch qui recalcule les chiffres exacts pendant la nuit. La section 7.8 examine ce choix du point de vue de l’architecture des données.

Back pressure et gestion des erreurs

Quand un producteur est plus rapide qu’un consommateur, quelque chose doit céder. Les options sont de mettre en tampon (ce qui ne fait que repousser le problème, sauf si le tampon est borné), d’écarter ou d’échantillonner des enregistrements (acceptable pour certaines métriques, jamais pour les paiements), ou d’appliquer un back pressure (contre-pression) : obliger l’étape rapide à ralentir jusqu’à la vitesse de l’étape lente. Les tubes Unix le font automatiquement, car un tube plein bloque l’écrivain. Les conceptions en mode pull, comme les générateurs Python, le font par construction, car rien n’est produit tant que l’étape suivante ne le demande pas. Les pipelines distribués doivent l’intégrer dès la conception : files bornées, consommateurs qui tirent les données à leur propre rythme depuis un journal, ou protocoles comme la spécification Reactive Streams de la JVM, dans laquelle un abonné signale combien d’enregistrements il est prêt à recevoir. La section 9.3 traite en profondeur des files d’attente et du back pressure.

Les erreurs exigent une politique pour chaque filtre, décidée à l’avance. Un enregistrement malformé (un message empoisonné, ou poison message) ne doit pas arrêter le pipeline : aiguillez-le vers une file de lettres mortes (dead-letter queue) ou un fichier d’erreurs avec le motif, comptez-le, et déclenchez une alerte si le taux augmente. Une défaillance transitoire, comme un service de recherche qui dépasse son délai d’expiration, donne lieu à de nouvelles tentatives avec backoff. Un plantage du pipeline exige un point de reprise : les traitements batch sont relancés depuis le début ou depuis un point de reprise (checkpoint), et les frameworks de streaming enregistrent périodiquement leur position et leur état. Comme les redémarrages impliquent que certains enregistrements soient traités deux fois, les consommateurs qui écrivent des résultats doivent être idempotents, par exemple en faisant un upsert sur une clé naturelle plutôt qu’une insertion à l’aveugle. Les erreurs sont le point faible du style : les filtres s’ignorent délibérément les uns les autres, si bien qu’une défaillance découverte à l’étape quatre ne peut pas facilement annuler ce que l’étape deux a déjà fait, et un pipeline qui a besoin de ce type de retour arrière coordonné relève probablement d’un autre style.

Quand l’utiliser, quand l’éviter, et les anti-patterns

Utilisez les tubes et filtres quand le problème est naturellement un flux de données à sens unique à travers des transformations distinctes : ingestion, conversion, enrichissement, validation, agrégation, export. C’est le choix par défaut en ingénierie des données, pour le traitement de fichiers et de contenus multimédias, et pour tout traitement dont les étapes changent plus souvent que le flux global. Il apparaît aussi à l’intérieur d’autres styles : un traitement de reporting nocturne au sein d’un monolithe modulaire, ou un service de traitement de flux au sein d’un système piloté par les événements.

Évitez-le comme structure principale des applications interactives, où les utilisateurs attendent des réponses immédiates et conversationnelles et où le travail n’est pas un flux ; pour une logique qui exige que de nombreuses étapes partagent et modifient le même état ; et pour les processus qui nécessitent un retour arrière coordonné entre étapes, qui appellent une transaction ou une saga.

Anti-patterns courants : le filtre divin, une seule étape qui analyse, enrichit, valide et écrit, ce qui fait perdre les avantages du style ; l’état partagé caché, des filtres qui communiquent par une table de base de données ou une variable globale partagée au lieu du tube, ce qui réintroduit le couplage que le style supprime ; la dérive de format, un producteur qui change silencieusement son format de sortie et casse tous les filtres en aval, ce qu’empêchent des schémas explicites et des vérifications de schéma (voir la section 7.7) ; les tampons non bornés, qui transforment un consommateur lent en plantage par manque de mémoire ; et les puits non idempotents, qui comptent en double après chaque nouvelle tentative.

Faire respecter le style dans le code

Les contraintes du style peuvent être rendues explicites. Donnez à chaque filtre la même signature simple (un itérable d’enregistrements en entrée, un itérable en sortie), pour que la composition soit mécanique. Définissez le format des enregistrements à chaque tube sous forme de schéma (une dataclass, un JSON Schema, une définition Avro ou Protocol Buffers) et validez-le à la frontière. Gardez les filtres exempts d’état mutable partagé : un test d’architecture peut interdire aux modules de filtres d’importer la couche de base de données ou de s’importer entre eux, de sorte que le seul moyen pour les données de passer de l’un à l’autre soit le tube. Testez chaque filtre avec de petites fixtures d’entrée et de sortie, et testez le pipeline assemblé de bout en bout avec un échantillon représentatif, enregistrements malformés compris, pour prouver que la politique d’erreur fonctionne.

Dans les pipelines distribués, surveillez le débit par étape, la profondeur des files et le retard des consommateurs (consumer lag) : une file qui grossit devant une étape est le signe visible d’un goulet d’étranglement ou d’un filtre défaillant, et c’est la première métrique sur laquelle alerter.