MapReduce : découper le calcul

Le chapitre précédent a montré comment traiter des données qui ne tiennent pas en mémoire, sur une seule machine, en une seule lecture. Cette solution atteint sa limite quand la lecture elle-même devient trop longue : lire un pétaoctet à 200 mégaoctets par seconde demande deux mois, et aucun algorithme en une passe n'y changera rien.

La seule issue est alors de lire à plusieurs endroits en même temps. C'est ce que fait le motif présenté ici, qui porte le nom de ses deux étapes, map et reduce, et qui structure la quasi-totalité des systèmes de traitement réparti, quel que soit leur habillage.

Le motif, en quatre temps

Avant les trois étapes vient la découpe, et sa règle découle d'un ordre de grandeur. Rassembler 10 téraoctets sur une seule machine de calcul, par un réseau à 1 gigabit par seconde, revient à faire passer $8 \times 10^{13}$ bits à $10^9$ bits par seconde : 80 000 secondes, environ 22 heures, avant le premier calcul. Le programme de traitement, lui, tient en quelques kilooctets et rejoint chaque machine en moins d'une milliseconde. Le motif envoie donc le calcul là où se trouvent les données, et non l'inverse.

Or des données de ce volume sont déjà réparties : un système de fichiers distribué a coupé le fichier en blocs de taille fixe, posés sur de nombreuses machines. La découpe initiale suit ces blocs, et chaque machine traite, autant que possible, ceux que porte son propre disque. Le réseau, dont le coût vient d'être chiffré, ne sert alors presque plus qu'au mélange, ce qui explique la place que ce chapitre lui donne. Le chapitre suivant, Partitionnement et réplication des données, décrit ces blocs et le catalogue qui sait où ils se trouvent.

Le comptage de mots est l'exemple canonique du domaine, et il le mérite : il tient en trois lignes et contient tout le motif.

Les quatre phases se parcourent avec les boutons. La découpe répartit les lignes, sans regarder ce qu'elles contiennent. Le map produit un couple par mot, (mot, 1), sur chaque machine indépendamment. Le mélange rassemble par mot, et l'on voit alors les couleurs se mêler : ce qui vient de la machine 1 rejoint ce qui vient de la machine 3. La réduction additionne.

Le map ne sait rien, et c'est ce qui le rend parallélisable. Le traitement d'une ligne par le map ne dépend d'aucune autre ligne. C'est cette propriété, et elle seule, qui autorise à découper n'importe comment et à traiter en parallèle sans coordination.

Toute logique qui exige de connaître les autres lignes doit donc être repoussée dans le reduce, après le regroupement par clé. Un map qui a besoin d'un compteur global n'est pas un map.

Écrit à la main, le motif ne fait qu'une vingtaine de lignes de Python. Rien n'y est mystérieux, et le voir en entier vaut mieux que de faire confiance à un moteur.

Le mélange est ce qui coûte

Dans la figure ci-dessus, dix-huit couples traversent le réseau pour produire dix résultats. Sur un texte réel, c'est un couple par mot du corpus : des milliards de couples de la forme (le, 1), tous identiques, tous transportés séparément.

L'idée qui corrige cela est simple : réduire d'abord localement, sur chaque machine, avant d'envoyer.

La phase de mélange annonce maintenant quinze couples au lieu de dix-huit : chaque machine envoie un seul (le, 2) au lieu de deux (le, 1), et rien d'autre ne se répète dans une même phrase. Sur trois phrases le gain est anecdotique ; sur un corpus, où le revient des milliers de fois par morceau, il fait la différence entre un traitement qui tient et un traitement qui sature le réseau.

Ce qui se recolle, et ce qui ne se recolle pas

L'affirmation se vérifie sur quatre valeurs.

La machine 1 a reçu 12 et 18, et envoie leur moyenne, 15. La machine 2 a reçu 12, et envoie 12. Le réducteur fait la moyenne de 15 et de 12, soit 13,5. La bonne réponse est 14, et la figure l'affiche à côté.

L'erreur n'est pas dans l'arithmétique : elle est dans ce qui a été transporté. Une moyenne résume deux informations, un total et un effectif, et n'en envoyer qu'une seule revient à traiter un morceau de deux valeurs comme un morceau d'une seule.

Transporter de quoi refaire le calcul, et ne diviser qu'à la fin. Le combineur juste envoie le couple (somme, effectif). Le réducteur additionne les sommes, additionne les effectifs, et divise une seule fois, à la fin.

Ce raisonnement se généralise : une réduction non associative devient calculable par morceaux dès qu'on transporte un état intermédiaire qui, lui, se combine associativement et dans n'importe quel ordre.

Pour la variance, cet état comporte trois nombres. Les accumulateurs du chapitre précédent s'additionnent composante par composante, mais ils gardent leur annulation catastrophique. L'état de Welford, le compte, la moyenne et la somme des écarts au carré, se recolle aussi, par une formule à peine plus longue : avec n = na + nb et d = mb - ma, la moyenne vaut ma + d * nb / n, et M2 = M2a + M2b + d * d * na * nb / n. Pour un écart type, les mêmes. Pour une médiane, aucun état de taille fixe ne convient, ce qui est une autre façon de dire qu'elle n'est pas décomposable.

Le même piège, écrit en Python, sur des données dont le déséquilibre est explicite.

Les morceaux ne sont pas égaux

Le mélange impose que toutes les valeurs d'une clé finissent au même endroit. Quand les données sont déjà rangées par clé, comme une table partitionnée par ville, aucun mélange n'est nécessaire, ce qui évite de tout transporter. Cette économie a un prix, et une clé plus fréquente que les autres en fixe un second. La figure suivante tire trente ventes de trois villes, à graine fixe, et se règle : nombre de machines, combineur, répartition des lignes, poids des clés. Son bilan compte les couples qui traversent le réseau et ceux que reçoit la machine de réduction la plus chargée.

Quand une clé domine, une machine porte presque tout. Le calcul dure ce que dure la plus chargée : ajouter des machines n'y change rien, puisque la clé paris ne se coupe pas en deux.

Ce déséquilibre tient aussi à la façon dont les données sont rangées. Découpées sans regarder la clé, comme un fichier coupé en blocs, et réduites localement avant l'envoi, les ventes se répartissent d'elles-mêmes : chaque machine somme ses propres ventes de paris, et la réduction ne reçoit plus qu'un total par machine, trois en tout à trois machines. C'est ce que fait d'office reduceByKey dans Spark. Le remède ne vaut que pour une réduction qui se recolle : une médiane par ville garde tout son déséquilibre.

Où sont Hadoop et Spark

Le motif présenté ici est antérieur aux outils qui l'ont popularisé, et leur survivra. Il est utile de savoir ce que ces noms recouvrent, à condition de ne pas confondre le motif avec ses mises en œuvre.

Hadoop désigne un ensemble, dont un système de fichiers réparti qui découpe les fichiers en blocs répliqués sur plusieurs machines, et un moteur qui exécute des tâches map et reduce sur ces blocs, en écrivant sur disque entre chaque étape. Cette écriture systématique le rend robuste et lent.

Spark exécute le même motif en enchaînant sans écrire les étapes qui n'exigent pas de mélange, et en gardant en mémoire les résultats qu'on lui demande de garder, ce qui change les ordres de grandeur sur les traitements enchaînés. Le mélange, lui, passe par des fichiers intermédiaires sur disque, comme chez Hadoop. Son interface range ses opérations en transformations, dont map, et en actions, dont reduce : la mécanique du dessous est celle de ce chapitre, mélange compris.

Ce que ces outils ne dispensent pas de savoir. Aucun moteur ne détecte qu'une moyenne a été recollée à tort : il exécute ce qu'on lui décrit et rend un résultat vraisemblable.

Aucun moteur ne répare seul un déséquilibre qui tient au calcul demandé. Certains en absorbent une partie : depuis sa version 3.2, Spark découpe d'office une jointure dont une clé déborde, et une réduction qui se recolle passe par une réduction locale qui allège la clé lourde. Une médiane par clé, un décompte de distincts ou une clé fautive, inconnu ou null, restent à la charge du concepteur.

Aucun moteur ne devine que le calcul demandé aurait pu tenir sur une seule machine en une passe. Un traitement réparti sur des données de quelques gigaoctets est souvent plus lent que la boucle du chapitre précédent, à cause du temps de mise en route et du mélange.

Mettre en pratique

Le mélange est l'étape qui coûte, parce qu'elle fait traverser le réseau. La réduire est affaire de comptage avant d'être affaire de réglage.

La dernière réponse est la définition même du déséquilibre. Ajouter des réducteurs ne change rien tant qu'une seule clé concentre la charge, parce que le temps total est celui du réducteur le plus lent. La solution n'est pas d'ajouter des machines : elle est de réduire localement quand la réduction le permet, sinon de changer la clé.