Le traitement par lots (batch)

Jusqu'ici, ce livre a surtout parlé de requêtes et de réponses : on demande quelque chose, et le système nous répond, idéalement vite, parce qu'un humain attend. Bases de données, caches, index de recherche, serveurs web fonctionnent ainsi. Mais ce style requête/réponse n'est pas la seule manière de bâtir un système. Kleppmann distingue trois familles, qu'il faut garder en tête tout au long de ce chapitre.

Le traitement par lots (batch processing) prend une grande quantité de données en entrée, exécute un travail (job) pour les transformer, et produit des données en sortie. Un job dure souvent de quelques minutes à plusieurs jours ; personne n'attend devant son écran, et on le planifie typiquement à intervalle régulier (une fois par jour, par exemple). Sa mesure de performance n'est pas le temps de réponse, mais le débit (throughput) : combien de temps pour brasser un jeu de données d'une taille donnée. C'est une forme de calcul très ancienne — les machines à cartes perforées Hollerith du recensement américain de 1890 en faisaient déjà — et son étude reste fondatrice pour bâtir des applications fiables et scalables.

La philosophie Unix comme modèle

Avant MapReduce et Hadoop, le meilleur point de départ reste les outils Unix standard. Imaginons un serveur web qui ajoute une ligne à un fichier de log à chaque requête servie. Pour trouver les cinq pages les plus populaires, on peut écrire un pipeline (enchaînement de commandes reliées par des tubes) directement dans un shell.

cat /var/log/nginx/access.log |
  awk '{print $7}' |   # extraire l'URL (7e champ)
  sort             |   # regrouper les URL identiques
  uniq -c          |   # compter les occurrences adjacentes
  sort -r -n       |   # trier par compte décroissant
  head -n 5            # ne garder que les 5 premières

Cette ligne obscure pour le néophyte est incroyablement puissante : elle traite des gigaoctets de logs en quelques secondes, et se modifie aisément (compter les adresses IP plutôt que les URL, exclure les fichiers CSS…). On pourrait écrire le même calcul comme un programme avec une table de hachage en mémoire ; le résultat serait équivalent, mais le flux d'exécution diffère profondément sur un gros fichier.

Trier ou agréger en mémoire ?

Un script qui maintient une table de hachage URL → compteur garde tout son ensemble de travail (working set, la mémoire à accès aléatoire nécessaire) en RAM. Tant que les URL distinctes tiennent en mémoire — disons 1 Go pour un site moyen —, cela marche très bien, même sur un portable. Mais si l'ensemble de travail dépasse la mémoire, l'approche par tri prend l'avantage : elle exploite efficacement le disque. C'est le principe des SSTables et des arbres LSM (chapitre 3) : des morceaux de données triés en mémoire, écrits sur disque, puis fusionnés en un fichier trié plus grand. Le tri-fusion (mergesort) a des motifs d'accès séquentiels idéaux pour le disque. L'outil sort de GNU Coreutils déborde automatiquement sur disque et parallélise sur plusieurs cœurs, ce qui permet à ce simple pipeline de passer à l'échelle sans saturer la mémoire.

Ce qui rend les outils Unix composables

La force d'Unix tient à quelques principes énoncés dès 1978 : « faire qu'un programme fasse une seule chose, et bien » ; s'attendre à ce que la sortie de tout programme devienne l'entrée d'un autre, encore inconnu ; bâtir vite, et jeter sans hésiter les parties maladroites. Cela ressemble, quarante ans plus tard, à l'agilité et au DevOps. sort en est l'exemple parfait : meilleur que la plupart des tris de bibliothèques standard, il reste pourtant inutile seul — il ne devient puissant qu'associé à uniq. Son auteur aurait pu y intégrer uniq ; il a résisté à la tentation.

Trois propriétés rendent cette composition possible.

Propriété	Mécanisme	Bénéfice
Interface uniforme	Tout est un fichier : suite ordonnée d'octets, souvent du texte ASCII en lignes séparées par \n.	N'importe quelle sortie se branche sur n'importe quelle entrée.
Séparation logique / câblage	Les programmes lisent stdin, écrivent stdout ; les tubes relient l'un à l'autre.	L'utilisateur câble les entrées/sorties à sa guise ; couplage lâche.
Transparence	Entrées traitées comme immuables ; on peut inspecter chaque étage (less), écrire un résultat intermédiaire dans un fichier.	Expérimentation et débogage sans danger.

Le revers de la médaille : ces outils ne tournent que sur une seule machine. C'est exactement là qu'interviennent des outils comme Hadoop.

MapReduce et les systèmes de fichiers distribués

MapReduce est un peu comme les outils Unix, mais réparti sur potentiellement des milliers de machines. C'est un outil brutal, à la force brute, mais étonnamment efficace. Un job MapReduce ressemble à un processus Unix : il prend une ou plusieurs entrées, produit une ou plusieurs sorties, ne modifie pas l'entrée, et n'a d'autre effet de bord que d'écrire sa sortie une fois, séquentiellement.

Là où les outils Unix utilisent stdin/stdout, MapReduce lit et écrit des fichiers sur un système de fichiers distribué — HDFS (Hadoop Distributed FileSystem) dans l'implémentation Hadoop, une réimplémentation open source du GFS de Google. HDFS repose sur le principe du sans-partage (shared-nothing) : chaque machine du datacenter a ses propres disques, et HDFS forme conceptuellement un grand système de fichiers utilisant l'espace de toutes. Les blocs sont répliqués sur plusieurs machines pour tolérer les pannes (réplication simple, ou codes correcteurs type Reed-Solomon). À l'écriture de ces lignes, les plus grands déploiements HDFS tournent sur des dizaines de milliers de machines, pour des centaines de pétaoctets.

L'exécution d'un job MapReduce

Le schéma de traitement reproduit exactement notre analyse de logs. On écrit deux fonctions de rappel (callbacks).

• La fonction map est appelée une fois par enregistrement d'entrée ; son rôle est d'en extraire une clé et une valeur (dans l'exemple, awk '{print $7}' est le mapper : l'URL est la clé). Un enregistrement peut produire zéro, une ou plusieurs paires clé-valeur.
• Le framework collecte toutes les paires de même clé et appelle la fonction reduce avec un itérateur sur cette collection de valeurs (uniq -c joue ce rôle).

// Comptage de vues par URL : un seul job MapReduce.
function map(record) {
  const url = record.split(/s+/)[6]; // 7e champ
  emit(url, 1);                       // clé = URL, valeur = 1
}

// Le framework regroupe et trie par clé AVANT d'appeler reduce.
function reduce(url, values) {
  let count = 0;
  for (const v of values) count += v;
  emit(url, count);
}

L'étape cruciale est implicite : entre map et reduce, la sortie des mappers est toujours triée par clé. Ce tri est sans doute l'aspect le plus important de MapReduce. La parallélisation repose sur le partitionnement (chapitre 6) : chaque bloc de fichier en entrée est une partition traitée par une tâche mapper distincte. Le planificateur tente d'exécuter chaque mapper sur une machine qui détient déjà une réplique du bloc — c'est le principe de rapprocher le calcul des données (putting the computation near the data), qui économise des copies réseau.

Le tri ne se fait pas sur une seule machine. Chaque mapper partitionne sa sortie par reducer (via un hash de la clé), l'écrit triée sur son disque local, et les reducers viennent ensuite chercher (fetch) leur partition chez chaque mapper. Ce processus de partitionnement, tri et copie des mappers vers les reducers s'appelle le brassage (shuffle, terme trompeur : il n'y a aucun aléa). Le reducer fusionne les fichiers reçus en préservant l'ordre de tri, puis applique sa logique.

Enchaîner les jobs en workflows

Un seul job MapReduce résout peu de problèmes. Pour obtenir les URL les plus populaires, il faut un second tour de tri, donc un second job dont l'entrée est la sortie du premier. Hadoop ne gère pas nativement ces enchaînements : le câblage se fait implicitement par nom de répertoire HDFS. Les jobs sont vus comme indépendants. La sortie d'un job n'est valide que s'il s'est terminé avec succès (MapReduce jette la sortie partielle d'un job échoué) ; un job aval ne peut donc démarrer qu'une fois ses jobs amont terminés. Pour gérer ces dépendances, on emploie des ordonnanceurs de workflow (Oozie, Azkaban, Luigi, Airflow, Pinball). Les workflows de 50 à 100 jobs sont courants pour bâtir des systèmes de recommandation.

Les jointures en MapReduce

Une jointure (join) est nécessaire dès qu'un traitement doit accéder aux enregistrements des deux côtés d'une association (clé étrangère relationnelle, référence de document, arête de graphe). Or MapReduce n'a pas d'index : recevant des fichiers en entrée, il en lit tout le contenu — ce qu'une base appellerait un full table scan. Pour une requête analytique qui agrège sur un grand nombre d'enregistrements et se parallélise, ce balayage complet est tout à fait raisonnable. Dans le contexte du batch, « jointure » signifie résoudre toutes les occurrences d'une association dans le jeu de données.

Jointures côté reduce : le sort-merge join

Prenons un log d'événements d'activité utilisateur (la table de faits) à joindre avec une base de profils (une dimension), par exemple pour corréler l'âge des visiteurs aux pages vues. L'approche naïve — interroger la base distante pour chaque événement — serait désastreuse : le débit serait limité par le temps d'aller-retour réseau, et le job deviendrait non déterministe, la base distante pouvant changer. Mieux vaut copier la base d'utilisateurs dans le même HDFS (via un processus ETL) et laisser MapReduce rapprocher les enregistrements.

On utilise alors la clé de jointure (ici l'ID utilisateur) comme clé de map : un jeu de mappers parcourt les événements, un autre la base de profils, tous deux émettant l'ID comme clé. Après partitionnement et tri, tous les enregistrements de même ID deviennent adjacents dans l'entrée du reducer. On peut même imposer un tri secondaire (secondary sort) pour que le reducer voie d'abord la date de naissance, puis les événements. Le reducer mémorise la date de naissance dans une variable locale et itère sur les événements : c'est le sort-merge join. Comme il traite tous les enregistrements d'un ID d'un coup, il ne garde qu'un seul profil en mémoire et ne fait aucune requête réseau.

Le même schéma « rapprocher les données liées » sert au regroupement (GROUP BY) et à la sessionisation (collationner les événements d'une session pour A/B testing) : grouper et joindre se ressemblent beaucoup sur MapReduce.

Gérer le déséquilibre (skew)

Le schéma s'effondre s'il existe une quantité énorme de données liées à une seule clé — par exemple une célébrité aux millions d'abonnés dans un réseau social (un linchpin object). Tout converge vers un seul reducer, créant un déséquilibre (skew) : un reducer traite bien plus d'enregistrements que les autres, et tout job suivant attend le plus lent. Parades possibles :

• Skew join (Pig/Hive) : envoyer les enregistrements de la clé brûlante à un reducer aléatoire, et répliquer l'autre côté de la jointure vers tous les reducers.
• Regroupement en deux étapes : un premier job pré-agrège la clé brûlante sur des reducers aléatoires, un second combine les valeurs partielles.

Jointures côté map

Les jointures précédentes opèrent dans les reducers (reduce-side joins) : elles ne supposent rien sur l'entrée, mais le tri, la copie et la fusion coûtent cher. Si l'on peut faire certaines hypothèses sur les données, les map-side joins sont plus rapides : pas de reducer, pas de tri, chaque mapper lit un bloc et écrit un fichier.

Type de jointure	Hypothèse requise	Mécanisme
Sort-merge (reduce-side)	Aucune	Tri + fusion par clé dans le reducer ; cas général.
Broadcast hash (map-side)	Un côté tient en mémoire	Chaque mapper charge le petit jeu en table de hachage, puis balaie le gros.
Partitioned hash (map-side)	Les deux côtés partitionnés à l'identique	Chaque mapper ne charge qu'une partition de chaque entrée.
Map-side merge	Partitionnés et triés de la même façon	Le mapper fusionne incrémentalement, sans tout charger en mémoire.

Le broadcast hash join (le petit jeu est « diffusé » à toutes les partitions du gros) est supporté par Pig (replicated join), Hive (MapJoin), Impala. Le partitioned hash join (bucketed map join dans Hive) suppose un même nombre de partitions, une même clé et une même fonction de hachage des deux côtés — souvent garanti si des jobs antérieurs ont déjà produit ce regroupement. Optimiser ces stratégies exige donc de connaître la disposition physique des données (nombre de partitions, clés de tri), métadonnées maintenues par exemple dans le Hive metastore.

La sortie des workflows de batch

À quoi sert tout ce traitement ? La sortie d'un batch n'est ni du transactionnel (OLTP) ni vraiment de l'analytique : ce n'est généralement pas un rapport, mais une autre structure de données.

• Index de recherche : l'usage originel de MapReduce chez Google était de bâtir l'index de son moteur (un workflow de cinq à dix jobs). Les mappers partitionnent les documents, chaque reducer construit l'index de sa partition, écrit sur HDFS. Ces index sont immuables une fois créés. Pour refléter des changements, on peut soit ré-exécuter tout le workflow (« documents en entrée, index en sortie » : facile à raisonner), soit indexer incrémentalement.
• Stockages clé-valeur : pour des systèmes de ML ou de recommandation, la sortie est souvent une base interrogeable par le service web. Écrire directement dans une base de production depuis un mapper est une mauvaise idée (lenteur des requêtes une à une, risque de submerger la base, effets de bord visibles violant la garantie « tout ou rien »). La bonne solution : construire les fichiers de base à l'intérieur du job, les écrire immuables sur HDFS, puis les charger en masse dans des serveurs en lecture seule (Voldemort, Terrapin, ElephantDB, HBase bulk loading). Voldemort continue de servir l'ancien fichier pendant la copie, puis bascule atomiquement ; en cas de souci, il revient à l'ancien.

Entrée immuable, sortie dérivée

C'est le patron central hérité d'Unix : un programme lit son entrée et écrit sa sortie, sans toucher l'entrée, en remplaçant intégralement l'ancienne sortie, sans effet de bord. On peut donc ré-exécuter à volonté.

Comparaison avec les bases de données MPP

Quand l'article MapReduce parut (2004), ses algorithmes de jointure parallèle existaient depuis plus d'une décennie dans les bases massivement parallèles (MPP, massively parallel processing) comme Gamma, Teradata ou Tandem NonStop SQL. La plus grande différence : une MPP se concentre sur l'exécution parallèle de requêtes SQL analytiques, tandis que MapReduce + système de fichiers distribué offre quelque chose de plus proche d'un système d'exploitation généraliste capable d'exécuter des programmes arbitraires.

Axe	Bases MPP	MapReduce / Hadoop
Stockage	Modèle imposé, modélisation soignée avant l'import.	Octets bruts ; on déverse d'abord, on modélise après (schema-on-read, « data lake »).
Modèle de traitement	SQL déclaratif, monolithique et optimisé ; accessible aux outils d'analystes.	Code arbitraire (ML, NLP, analyse d'images) ; modèles de traitement variés sur un même cluster.
Tolérance aux fautes	Une panne annule souvent toute la requête (acceptable : quelques minutes).	Relance à la granularité d'une tâche ; conçu pour des jobs longs.
Mémoire / disque	Garde un maximum en mémoire (hash joins).	Écrit volontiers sur disque (tolérance, et données trop grosses pour la RAM).

Pourquoi MapReduce est-il si économe en mémoire et si prompt à relancer des tâches ? Parce qu'il fut conçu pour les datacenters mixtes de Google, où services en ligne et batchs partagent les mêmes machines. Les jobs batch tournent à basse priorité et peuvent être préemptés à tout instant pour libérer des ressources à un processus prioritaire. Chez Google, une tâche d'une heure a environ 5 % de risque d'être interrompue — un ordre de grandeur de plus que les pannes matérielles. Avec 100 tâches de 10 minutes, le risque qu'au moins une soit tuée dépasse 50 %. MapReduce tolère donc les terminaisons fréquentes non parce que le matériel serait peu fiable, mais parce que la liberté de tuer arbitrairement des processus améliore l'utilisation du cluster. Là où la préemption est rare (la plupart des ordonnanceurs open source), ces choix de conception ont moins de sens.

Au-delà de MapReduce : les moteurs de dataflow

MapReduce est un excellent outil pédagogique — une abstraction claire sur un système de fichiers distribué — mais il est laborieux à utiliser directement, et son modèle d'exécution souffre de problèmes de performance que des couches d'abstraction (Pig, Hive, Cascading, Crunch) ne corrigent pas.

Le coût de la matérialisation

Chaque job MapReduce écrit son état intermédiaire intégralement sur HDFS : c'est la matérialisation (materialization). Comparée aux tubes Unix qui passent la sortie incrémentalement via un petit tampon, elle a deux défauts :

• Un job ne peut démarrer que lorsque tous ses prédécesseurs sont entièrement terminés. Les tâches traînardes (stragglers) ralentissent alors tout le workflow.
• Les mappers sont souvent redondants : ils relisent un fichier qu'un reducer vient d'écrire, pour le repartitionner. Ce travail pourrait faire partie du reducer précédent.

Opérateurs et graphe de dataflow

Les moteurs de dataflow — Spark, Tez, Flink — traitent tout un workflow comme un seul job plutôt qu'en sous-jobs indépendants. Ils modélisent explicitement le flux de données à travers plusieurs étages. Comme MapReduce, ils appellent une fonction par enregistrement et parallélisent par partitionnement ; mais ces fonctions, appelées opérateurs, ne sont pas contraintes à l'alternance stricte map/reduce. Le moteur offre plusieurs façons de relier la sortie d'un opérateur à l'entrée d'un autre : repartitionner et trier par clé (pour les sort-merge joins), partitionner sans trier (pour les partitioned hash joins, où l'ordre est inutile), ou diffuser à toutes les partitions (broadcast hash join).

Aspect	MapReduce	Moteurs de dataflow (Spark, Tez, Flink)
Unité de travail	Sous-jobs indépendants reliés par répertoires HDFS.	Un seul job : graphe d'opérateurs explicite (DAG).
Tri	Toujours entre map et reduce.	Inséré seulement où il est requis.
État intermédiaire	Matérialisé sur HDFS (répliqué, sur disque).	Gardé en mémoire ou sur disque local ; moins d'E/S.
Démarrage	Attend la fin complète de l'étage précédent.	Un opérateur démarre dès que son entrée est prête (pipelining).
Tolérance aux fautes	Relecture depuis HDFS (durable).	Recalcul de l'état perdu à partir des données amont.
Mappers redondants	Fréquents.	Évités (le travail rejoint l'opérateur précédent).

Ces optimisations rendent les moteurs de dataflow significativement plus rapides, à code de traitement identique (un workflow Pig ou Hive passe de MapReduce à Tez par simple configuration). Pour la tolérance aux fautes, comme l'état intermédiaire n'est plus durable sur HDFS, il est recalculé à partir des données encore disponibles : Spark suit la généalogie des données via les RDD (resilient distributed datasets), Flink fait des points de contrôle (checkpoints) de l'état des opérateurs.

Graphes et traitement itératif

De nombreux algorithmes de graphes — PageRank, fermeture transitive — procèdent en parcourant une arête à la fois, en propageant une information de proche en proche, et en répétant jusqu'à convergence. Or MapReduce ne fait qu'une seule passe sur les données ; ce « répéter jusqu'à terminé » s'implémente alors par itération externe : un ordonnanceur lance une étape, vérifie la condition d'arrêt, relance si besoin. C'est très inefficace : à chaque tour, MapReduce relit tout le jeu d'entrée et réécrit toute la sortie, même si une infime partie du graphe a changé.

Le modèle Pregel

Pour optimiser cela, le modèle parallèle synchrone par lots (bulk synchronous parallel, BSP), popularisé par l'article Pregel de Google, s'est imposé (implémenté par Apache Giraph, l'API GraphX de Spark, l'API Gelly de Flink). L'idée prolonge MapReduce : un sommet peut « envoyer un message » à un autre, typiquement le long d'une arête. À chaque itération, une fonction est appelée pour chaque sommet avec tous les messages reçus — comme un appel à reduce. La différence cruciale : un sommet conserve son état en mémoire d'une itération à l'autre, et ne traite que les nouveaux messages entrants. Là où aucun message ne circule, aucun travail n'est fait. C'est proche du modèle d'acteurs, sauf que l'état et les messages sont durables et tolérants aux pannes, et que la communication procède par tours fixes : tous les messages d'une itération sont délivrés à la suivante, exactement une fois.

La tolérance aux fautes s'obtient par des points de contrôle périodiques de l'état de tous les sommets. La parallélisation se fait selon le principe « penser comme un sommet » (think like a vertex) : un sommet ne connaît pas sa machine, il envoie à un ID de sommet, et le framework partitionne le graphe (souvent par simple hash d'ID). D'où une mise en garde : les algorithmes de graphes engendrent beaucoup de communication inter-machines, et l'état intermédiaire (les messages) dépasse souvent le graphe d'origine. Si le graphe tient en mémoire d'une seule machine, un algorithme mono-machine (voire mono-thread) le surclassera probablement ; le distribué (Pregel) ne devient incontournable que pour les graphes trop gros.

APIs de haut niveau et retour vers le déclaratif

L'infrastructure du batch distribué étant désormais robuste (des pétaoctets sur plus de 10 000 machines), l'attention s'est portée sur le modèle de programmation. Les APIs de haut niveau (Hive, Pig, Cascading, Crunch, et celles intégrées à Spark et Flink) utilisent des briques de style relationnel : joindre, grouper, filtrer, agréger. Elles réduisent le code, permettent un usage interactif (écrire l'analyse incrémentalement dans un shell, à la manière Unix), et améliorent l'efficacité machine.

Spécifier les jointures de façon déclarative laisse le framework analyser les entrées et choisir automatiquement le bon algorithme — Hive, Spark et Flink ont des optimiseurs de requêtes à base de coûts qui réordonnent même les jointures pour minimiser l'état intermédiaire. Mais MapReduce et ses successeurs restent bâtis sur des callbacks : une fonction définie par l'utilisateur peut appeler du code arbitraire, et puiser dans tout l'écosystème de bibliothèques (analyse de langage, d'images, statistiques) — ce qui les distingue durablement des MPP. En incorporant des aspects déclaratifs (stockage en colonnes, exécution vectorisée, génération de code natif), les moteurs de batch finissent par ressembler aux MPP tout en gardant leur flexibilité. À mesure que les MPP deviennent plus programmables, les deux mondes convergent : ce ne sont, au fond, que des systèmes pour stocker et traiter des données.

À retenir

• Trois familles de systèmes coexistent : services (online, temps de réponse), traitement par lots (offline, optimise le débit sur des données bornées), traitement de flux (chapitre 11). Le batch lit une entrée et produit une sortie dérivée, sans modifier l'entrée.
• La philosophie Unix est le modèle : petits outils faisant une seule chose, reliés par des tubes via une interface uniforme (flux d'octets), avec entrées immuables. MapReduce en est la version distribuée sur HDFS.
• MapReduce = map (extraire clé/valeur) → tri/brassage (shuffle, regroupement par clé via hash) → reduce. Le tri implicite est son cœur ; la tolérance aux pannes vient de la relance des tâches échouées, qui relisent leur entrée durable depuis HDFS.
• Les jointures illustrent les algorithmes partitionnés : sort-merge (reduce-side, cas général), broadcast hash et partitioned hash (map-side, plus rapides sous hypothèses). Le déséquilibre (skew) des clés brûlantes se traite par skew join ou regroupement en deux étapes.
• Les moteurs de dataflow (Spark, Tez, Flink) traitent tout le workflow comme un graphe d'opérateurs, gardent l'état intermédiaire en mémoire (moins d'écritures disque), insèrent le tri seulement si nécessaire, et tolèrent les pannes par recalcul — d'où l'importance d'opérateurs déterministes.
• Le traitement de graphes (modèle Pregel / BSP) propage des messages entre sommets en conservant leur état d'une itération à l'autre. Comparées aux MPP, les plateformes Hadoop privilégient le code arbitraire et le schema-on-read ; déclaratif et programmable convergent peu à peu.