6 MapReduce sur une machine

Objectifs de la séance

Comprendre le modèle MapReduce et pourquoi il a été inventé.
Écrire un moteur MapReduce complet en Python, à partir de rien.
Comprendre la phase de shuffle, qui est le cœur du problème.
Paralléliser avec multiprocessing et mesurer le gain.

Diaporama

6.1 Le problème de départ

Vous avez 10 000 fichiers texte et vous voulez compter les occurrences de chaque mot. La solution évidente tient en cinq lignes :

compteur = {}
for nom in fichiers:
    for mot in open(nom, encoding="utf-8").read().split():
        mot = mot.lower()
        compteur[mot] = compteur.get(mot, 0) + 1

Ce programme est correct. Il a deux défauts rédhibitoires à grande échelle :

Il est séquentiel : un seul cœur travaille, les sept autres regardent.
Le dictionnaire compteur doit tenir entièrement en mémoire. Sur un corpus important, il contient des dizaines de millions d’entrées.

Et si l’on veut le faire tourner sur cent machines, il faut tout réécrire.

Pourquoi c’est important

En 2004, deux ingénieurs de Google, Jeffrey Dean et Sanjay Ghemawat, font le constat suivant : chez eux, des centaines de programmes différents traitent d’énormes volumes de données, et tous réécrivent le même code pénible — découper l’entrée, répartir le travail, gérer les pannes, regrouper les résultats. Leur idée : si l’on impose aux programmeurs un modèle très contraint, on peut automatiser tout le reste. C’est l’article MapReduce: Simplified Data Processing on Large Clusters, l’un des plus influents de l’informatique moderne, et l’ancêtre direct d’Hadoop.

6.2 Le corpus de travail

Tous les exercices de cette séance, et ceux des séances 7 et 8, ont besoin de fichiers texte à traiter. Préparons-les une fois pour toutes.

Récupérer quelques textes

Le Projet Gutenberg publie des œuvres du domaine public en texte brut. Ce script en récupère trois, avec la seule bibliothèque standard :

telecharger_corpus.py

"""Telecharge quelques textes du domaine public dans corpus_source/."""
import pathlib
import time
import urllib.request

LIVRES = {
    "miserables":  17489,     # Hugo, Les Miserables I
    "vingt-mille":  5097,     # Verne, Vingt mille lieues sous les mers
    "rouge-noir":    798,     # Stendhal, Le Rouge et le Noir
}
MODELE = "https://www.gutenberg.org/cache/epub/{id}/pg{id}.txt"

dossier = pathlib.Path("corpus_source")
dossier.mkdir(exist_ok=True)

for nom, identifiant in LIVRES.items():
    cible = dossier / f"{nom}.txt"
    if cible.exists():
        print(f"{cible} deja present")
        continue
    url = MODELE.format(id=identifiant)
    with urllib.request.urlopen(url, timeout=30) as reponse:
        cible.write_bytes(reponse.read())
    print(f"{cible} : {cible.stat().st_size} octets")
    time.sleep(3)          # Gutenberg limite les requetes rapprochees

Cela représente environ 2,7 Mo de texte français.

Le sleep(3) n’est pas décoratif

Gutenberg limite les requêtes rapprochées : en enchaînant les téléchargements sans pause, les suivants échouent.

Un seul téléchargement suffit : le NFS

Sur les machines de l’école, votre répertoire personnel (~, c’est-à-dire /cal/homes/votre_login) n’est pas sur le disque de la machine devant laquelle vous êtes assis. Il se trouve sur un serveur de fichiers central, et chaque machine y accède par le réseau, grâce au protocole NFS (Network File System, système de fichiers en réseau). Pour vous, c’est invisible : vous ouvrez vos fichiers comme s’ils étaient locaux.

Conséquence pratique : un fichier créé dans votre répertoire personnel sur la machine tp-1a201-05 est immédiatement visible depuis tp-1a201-18 ou n’importe quelle autre machine de l’école. Le corpus téléchargé une fois est donc disponible partout, inutile de recommencer sur chaque poste.

À l’inverse, /tmp est sur le disque local de chaque machine : ce qui y est écrit n’existe que sur cette machine-là. Nous utiliserons cette différence en séance 7.

Le répertoire personnel d’un camarade ne vous est pas accessible. Si votre enseignant met un corpus commun à disposition, il vous en donnera le chemin : utilisez-le plutôt que de télécharger.

Fabriquer des fichiers à la bonne taille

2,7 Mo ne suffisent pas pour mesurer quoi que ce soit. Le script suivant découpe et recopie ces textes pour produire $N$ fichiers de la taille que vous voulez, sans aucun accès réseau :

fabriquer_corpus.py

"""Fabrique N fichiers de taille voulue a partir de corpus_source/.

    python3 fabriquer_corpus.py 8 1      -> 8 fichiers de 1 Mo dans corpus/
"""
import pathlib
import sys

NB = int(sys.argv[1]) if len(sys.argv) > 1 else 8
TAILLE_MO = float(sys.argv[2]) if len(sys.argv) > 2 else 1.0
OCTETS = int(TAILLE_MO * 1024 * 1024)

sources = sorted(pathlib.Path("corpus_source").glob("*.txt"))
if not sources:
    sys.exit("corpus_source/ est vide : lancez telecharger_corpus.py")

texte = "\n".join(f.read_text(encoding="utf-8", errors="ignore")
                  for f in sources).encode("utf-8")

dossier = pathlib.Path("corpus")
dossier.mkdir(exist_ok=True)
for ancien in dossier.glob("part-*.txt"):    # restes d'un appel precedent
    ancien.unlink()

noms = []
for i in range(NB):
    # Decalage different pour chaque fichier : ils ne sont pas identiques.
    depart = (i * 9973) % len(texte)
    tourne = texte[depart:] + texte[:depart]
    donnees = (tourne * (OCTETS // len(tourne) + 1))[:OCTETS]
    donnees = donnees[:donnees.rfind(b"\n") + 1]   # couper sur une ligne
    chemin = dossier / f"part-{i:02d}.txt"
    chemin.write_bytes(donnees)
    noms.append(str(chemin.resolve()))

pathlib.Path("fichiers.txt").write_text("\n".join(noms) + "\n",
                                        encoding="utf-8")
total = sum((dossier / f"part-{i:02d}.txt").stat().st_size
            for i in range(NB))
print(f"{NB} fichiers dans {dossier}/ ({total / 1024 / 1024:.1f} Mo)")
print(f"fichiers.txt : {len(noms)} chemins absolus")

$ python3 fabriquer_corpus.py 8 1
8 fichiers dans corpus/ (8.0 Mo)
fichiers.txt : 8 chemins absolus

Comment il s’y prend. Il met bout à bout les trois livres (2,6 Mo), puis, pour chaque fichier, commence la lecture un peu plus loin dans ce texte et le répète jusqu’à la taille demandée. Les premières lignes de chaque fichier sont ainsi différentes, mais les fichiers se ressemblent beaucoup : ce sont les mêmes livres, décalés. Pour mesurer des temps de calcul, c’est exactement ce qu’il faut : des fichiers de même taille et de même difficulté. Ne vous étonnez simplement pas de retrouver les mêmes mots, à peu près dans les mêmes proportions, dans tous les fichiers.

Chaque appel remplace le contenu de corpus/ : après fabriquer_corpus.py 32 1 puis fabriquer_corpus.py 8 1, il ne reste que 8 fichiers.

`fichiers.txt`

Le script produit aussi fichiers.txt : un chemin absolu par ligne, la liste des fichiers à traiter.

fichiers.txt

/home/login/bgd701/corpus/part-00.txt
/home/login/bgd701/corpus/part-01.txt

C’est ce fichier que lira le maître en séance 7, exactement comme vivantes.txt lui donne la liste des machines. Les chemins doivent être absolus : le programme qui les ouvrira s’exécutera ailleurs, dans un autre répertoire courant, et parfois sur une autre machine.

Le corpus doit rester sur le NFS

En séance 7, chaque travailleur ouvrira ces fichiers lui-même. Ils doivent donc être visibles depuis toutes les machines, c’est-à-dire dans votre répertoire personnel — pas dans /tmp, qui est local à chaque machine.

6.3 L’ancêtre : compter les mots avec des outils Unix

Avant d’inventer quoi que ce soit, regardons comment un administrateur système résout ce problème depuis les années 1970. Le caractère |, appelé tube (pipe), branche la sortie d’une commande sur l’entrée de la suivante : le texte coule de gauche à droite, transformé à chaque étape.

$ cat corpus/*.txt \
    | grep -oE '[[:alpha:]]+' \
    | awk '{print tolower($0)}' \
    | sort \
    | uniq -c \
    | sort -rn \
    | head -8
  64020 de
  37304 il
  35815 la
  32067 et
  30970 le
  29672 à
  27656 l
  21502 un

Cette ligne de commande fait le même travail que le programme Python précédent, et elle est souvent plus rapide. Détaillons-la :

Étape	Rôle
`cat corpus/*.txt`	lit tous les fichiers d’entrée, l’un après l’autre
`grep -oE '[[:alpha:]]+'`	extrait chaque suite de lettres et l’écrit seule sur sa ligne : un mot par ligne
`awk '{print tolower($0)}'`	passe chaque ligne en minuscules, accents compris
`sort`	trie les lignes, ce qui place côte à côte les mots identiques
`uniq -c`	compte les lignes consécutives identiques
`sort -rn`	trie par effectif décroissant (`-n` : numérique, `-r` : inversé)
`head -8`	garde les 8 premiers

Pourquoi pas tr, que l’on trouve partout sur Internet ?

La recette classique utilise tr -c '[:alpha:]' '\n' et tr '[:upper:]' '[:lower:]'. Sur les machines Linux de l’école, elle détruit les mots accentués : tr y travaille octet par octet, et une lettre comme é occupe deux octets en UTF-8. « L’été, déjà là » devient alors l t d j l. La version ci-dessus donne le même résultat, correct, sur Linux et sur macOS.

Pourquoi les nombres diffèrent un peu de ceux de Python. grep coupe les mots à tout caractère qui n’est pas une lettre, y compris l’apostrophe : qu'il donne deux mots, qu et il. Le programme Python, lui, coupe aux espaces puis retire la ponctuation aux extrémités : qu'il reste un seul mot. D’où 37 304 il pour Unix et 26 219 pour Python sur le même corpus. Les deux méthodes sont cohérentes, elles ne définissent simplement pas le « mot » de la même façon.

À retenir

Regardez bien la quatrième étape. Le sort est le shuffle. C’est lui qui amène côte à côte toutes les occurrences d’une même clé, pour que l’étape suivante puisse les agréger. Les deux premières étapes sont le map ; uniq -c est le reduce.

Pourquoi cela fonctionne si bien

Deux propriétés, héritées de la philosophie Unix, expliquent l’efficacité de cette chaîne.

Une interface uniforme. Chaque programme lit un flux de texte sur son entrée standard et écrit un flux de texte sur sa sortie standard. C’est pauvre, mais c’est universel : n’importe quel outil se branche sur n’importe quel autre. Vous pouvez insérer un grep au milieu sans rien réécrire.

Un sort qui ne tient pas en mémoire. La commande sort d’Unix ne charge pas tout en RAM. Quand les données dépassent la mémoire disponible, elle trie par blocs, écrit chaque bloc trié sur disque, puis fusionne les blocs. C’est le tri par fusion externe. Résultat : sort traite sans broncher un fichier plus gros que la mémoire de la machine — ce que notre dictionnaire Python, lui, ne sait pas faire.

Ce qui bloque, et ce que MapReduce apporte

Cette chaîne s’arrête à la frontière de la machine :

elle utilise un seul disque, un seul jeu de processeurs ;
elle ne survit à aucune panne : si le processus meurt à 90 %, tout est perdu ;
on ne peut pas la répartir sur cent machines.

Pourquoi c’est important

MapReduce est, littéralement, la version distribuée de cette ligne de commande. Même structure en trois temps, même idée d’interface uniforme (des couples clé-valeur au lieu de lignes de texte), même rôle central du tri — mais réparti sur un millier de machines, avec la reprise sur panne en prime. Si vous retenez que MapReduce, c’est map | sort | reduce à l’échelle d’un centre de données, vous avez compris l’essentiel.

Pour aller plus loin

Cette filiation est développée dans Kleppmann & Riccomini, Designing Data-Intensive Applications, chapitre 11 (Batch Processing with Unix Tools).

6.4 Le modèle

Tout le modèle repose sur une seule structure de données : le couple clé-valeur, noté (clé, valeur). La clé dit de quoi l’on parle, la valeur dit ce que l’on en sait. Dans ("chat", 1), la clé est le mot "chat" et la valeur 1 signifie « je l’ai vu une fois ». C’est la même idée qu’un dictionnaire Python, à une différence près : ici, la même clé peut apparaître plusieurs fois, avec des valeurs différentes.

Le programmeur n’écrit que deux fonctions.

map reçoit un morceau de l’entrée (ici, un fichier) et produit une liste de couples (clé, valeur) intermédiaires.
reduce reçoit une clé et toutes les valeurs associées à cette clé, et produit le résultat final pour cette clé.

Entre les deux, le système effectue le shuffle (littéralement : le battage des cartes) : rassembler toutes les valeurs de même clé, d’où qu’elles viennent. C’est le système qui s’en charge, pas vous — mais aujourd’hui, c’est nous le système.

Le comptage de mots dans ce modèle

def fonction_map(nom_fichier, contenu):
    """Emet (mot, 1) pour chaque mot rencontre."""
    for mot in contenu.split():
        mot = mot.strip(".,;:!?\"'()[]").lower()
        if mot:
            yield (mot, 1)

def fonction_reduce(cle, valeurs):
    """Recoit ("chat", [1, 1, 1, 1]) et renvoie 4."""
    return sum(valeurs)

fonction_map découpe le texte aux espaces (split()), retire la ponctuation collée au début ou à la fin du mot (strip(...)), passe en minuscules (lower()) et ignore ce qui reste vide. Ce découpage sera le même en séances 7 et 8 : c’est ce qui permettra de comparer les résultats.

yield : une fonction qui produit au fur et à mesure

fonction_map n’a pas de return mais un yield. Une telle fonction s’appelle un générateur : au lieu de construire toute la liste des résultats puis de la renvoyer d’un coup, elle livre les résultats un par un, au moment où on les demande.

>>> list(fonction_map("f.txt", "Le chat, le chien."))
[('le', 1), ('chat', 1), ('le', 1), ('chien', 1)]

On s’en sert comme d’une liste dans une boucle for, ou on la convertit en liste avec list(...). L’intérêt : sur un fichier d’un million de mots, on n’a jamais en mémoire le million de couples à la fois.

C’est tout. Ces deux fonctions ne savent rien du parallélisme, du réseau ou des pannes. Elles sont identiques que l’on traite trois fichiers sur un portable ou un pétaoctet sur mille machines.

Le déroulement, sur un exemple minuscule

Entrée : deux fichiers, "le chat dort" et "le chien dort".

Phase	Contenu
Map (f1)	`(le,1) (chat,1) (dort,1)`
Map (f2)	`(le,1) (chien,1) (dort,1)`
Shuffle	`le:[1,1]` `chat:[1]` `dort:[1,1]` `chien:[1]`
Reduce	`le:2` `chat:1` `dort:2` `chien:1`

À retenir

La force du modèle vient d’une contrainte : les tâches map sont indépendantes les unes des autres. Chacune peut donc être exécutée n’importe où, dans n’importe quel ordre, et relancée en cas de panne sans conséquence. C’est ce qui rend la distribution et la tolérance aux pannes automatisables.

6.5 Un moteur MapReduce en 25 lignes

Voici le fichier complet. Il contient les deux fonctions du comptage de mots, puis le moteur mapreduce(), qui enchaîne les trois phases pour n’importe quel couple de fonctions qu’on lui donne. Les exercices et les programmes suivants l’importeront : recopiez-le en entier, sous ce nom exact.

mapreduce_simple.py

def fonction_map(nom_fichier, contenu):
    """Emet (mot, 1) pour chaque mot rencontre."""
    for mot in contenu.split():
        mot = mot.strip(".,;:!?\"'()[]").lower()
        if mot:
            yield (mot, 1)

def fonction_reduce(cle, valeurs):
    """Recoit ("chat", [1, 1, 1, 1]) et renvoie 4."""
    return sum(valeurs)

def mapreduce(fichiers, fonction_map, fonction_reduce):
    # ---- Phase MAP ----
    intermediaires = []
    for nom in fichiers:
        with open(nom, encoding="utf-8", errors="ignore") as f:
            contenu = f.read()
        intermediaires.extend(fonction_map(nom, contenu))

    # ---- Phase SHUFFLE : regrouper par cle ----
    groupes = {}
    for cle, valeur in intermediaires:
        groupes.setdefault(cle, []).append(valeur)

    # ---- Phase REDUCE ----
    resultat = {}
    for cle, valeurs in groupes.items():
        resultat[cle] = fonction_reduce(cle, valeurs)
    return resultat

if __name__ == "__main__":
    import sys
    compte = mapreduce(sys.argv[1:], fonction_map, fonction_reduce)
    for mot, n in sorted(compte.items(),
                         key=lambda x: -x[1])[:20]:
        print(f"{n:8d}  {mot}")

Les trois phases, ligne par ligne :

Map. Pour chaque fichier, on lit son contenu et on appelle fonction_map. extend ajoute tous les couples produits à la liste intermediaires.
Shuffle. On parcourt tous les couples et on range chaque valeur dans la liste de sa clé. groupes.setdefault(cle, []) renvoie la liste déjà associée à cle, ou en crée une vide si la clé est nouvelle ; on y ajoute ensuite la valeur. Après cette boucle, groupes["le"] vaut par exemple [1, 1, 1, 1, 1, 1].
Reduce. Pour chaque clé, on appelle fonction_reduce sur sa liste de valeurs.

Le bloc final, exécuté seulement quand on lance le fichier directement, trie les mots par effectif décroissant (key=lambda x: -x[1] : trier selon l’opposé du nombre) et affiche les 20 premiers.

Vous avez écrit un MapReduce. Il fonctionne, il est correct, et il souffre exactement des deux défauts du programme naïf : séquentiel, et tout en mémoire. Corrigeons cela.

6.6 Le shuffle : le vrai problème

Regardez à nouveau la phase de shuffle. Elle suppose que toutes les paires intermédiaires tiennent dans une seule liste, dans une seule mémoire. Avec cent machines, c’est impossible : chacune produit ses paires de son côté.

Il faut donc décider, à l’avance, quelle machine sera responsable de quelle clé. Et il faut que toutes les machines prennent la même décision, sans se concerter.

La formule

$\boxed{\;\texttt{reducteur} = \texttt{hash(cle)} \bmod R\;}$

où $R$ est le nombre de tâches reduce.

Cette formule a exactement la propriété qu’il faut : elle est déterministe, c’est-à-dire qu’elle donne toujours le même résultat pour la même clé, sur n’importe quelle machine. La machine 3 et la machine 47 qui rencontrent toutes deux le mot "chat" calculeront le même numéro de réducteur, sans échanger un seul message.

Vérifié avec la fonction partition ci-dessous, sur un Mac puis sur une machine de l’école, avec $R = 2$ :

Mot	`le`	`chat`	`chien`	`dort`
Réducteur, sur le Mac	1	1	0	1
Réducteur, sur `tp-1a201-18`	1	1	0	1

Toutes les occurrences de le, d’où qu’elles viennent, iront au réducteur 1 ; toutes celles de chien au réducteur 0.

Attention

Nous avons vu en séance 5 pourquoi il faut hashlib et non hash() : la fonction native de Python est salée aléatoirement à chaque démarrage. Ici, l’erreur serait fatale et sournoise — le programme fonctionnerait parfaitement sur une seule machine, et produirait des comptages faux dès qu’on le distribue.

Chaque map écrit $R$ fichiers

Une tâche est une unité de travail confiée à un processus ou à une machine. Ici, une tâche map traite un fichier d’entrée, et une tâche reduce traite une part des clés. On note :

$M$ le nombre de tâches map, donc le nombre de fichiers d’entrée ;
$R$ le nombre de tâches reduce, que l’on choisit librement.

Le schéma est alors le suivant :

chaque tâche map écrit $R$ fichiers, un par réducteur : map-000-part-00, map-000-part-01… Le fichier map-000-part-01 contient les couples produits par la tâche map 0 et destinés au réducteur 1 ;
le réducteur $r$ lit les $M$ fichiers map-*-part-0r et les fusionne.

Cela fait $M \times R$ fichiers intermédiaires — et c’est bien pourquoi le shuffle est la phase la plus coûteuse de tout MapReduce.

mapreduce_partitionne.py

import hashlib
import json
import os

def partition(cle, R):
    empreinte = hashlib.md5(str(cle).encode("utf-8")).hexdigest()
    return int(empreinte, 16) % R

def executer_map(id_map, nom_fichier, fonction_map, R, dossier):
    """Ecrit R fichiers de paires intermediaires."""
    with open(nom_fichier, encoding="utf-8", errors="ignore") as f:
        contenu = f.read()

    # Agregation locale (combiner) : reduit fortement le volume
    seaux = [dict() for _ in range(R)]
    for cle, valeur in fonction_map(nom_fichier, contenu):
        r = partition(cle, R)
        seaux[r][cle] = seaux[r].get(cle, 0) + valeur

    for r in range(R):
        chemin = os.path.join(dossier, f"map-{id_map:03d}-part-{r:02d}")
        with open(chemin, "w", encoding="utf-8") as f:
            for cle, valeur in seaux[r].items():
                f.write(json.dumps([cle, valeur]) + "\n")

def executer_reduce(r, nb_maps, fonction_reduce, dossier):
    """Lit les nb_maps fichiers destines au reducteur r."""
    groupes = {}
    for id_map in range(nb_maps):
        chemin = os.path.join(dossier, f"map-{id_map:03d}-part-{r:02d}")
        if not os.path.exists(chemin):
            continue
        with open(chemin, encoding="utf-8") as f:
            for ligne in f:
                cle, valeur = json.loads(ligne)
                groupes.setdefault(cle, []).append(valeur)

    sortie = os.path.join(dossier, f"resultat-part-{r:02d}")
    with open(sortie, "w", encoding="utf-8") as f:
        for cle, valeurs in groupes.items():
            f.write(json.dumps([cle, fonction_reduce(cle, valeurs)])
                    + "\n")

Le combiner

Remarquez la boucle d’agrégation locale dans executer_map : au lieu d’écrire mille fois ("le", 1), on écrit une fois ("le", 1000).

Cette optimisation s’appelle un combiner. C’est un reduce appliqué localement, par chaque tâche map, avant l’écriture sur disque. Comme le shuffle est la phase la plus coûteuse, c’est l’optimisation la plus rentable de tout MapReduce. Mesuré sur le corpus du cours :

Fichier traité	Sans combiner	Avec combiner	Facteur
1 Mo	2,22 Mo (171 952 couples)	0,31 Mo (18 179 couples)	$\approx 7$
8 Mo	17,61 Mo (1 355 325 couples)	0,55 Mo (31 355 couples)	$\approx 32$

Sans combiner, on écrit un couple par mot du texte ; avec, un couple par mot distinct. Or le nombre de mots distincts croît bien moins vite que la taille du texte : plus le fichier est gros, plus le gain est grand.

Le combiner n’est pas toujours permis

Avec un combiner, une partie des valeurs est réduite d’abord par chaque tâche map, et le résultat est réduit encore une fois par le réducteur. Ce calcul en deux temps doit donner le même résultat que le calcul en un seul temps. C’est vrai quand la fonction reduce est :

associative : on peut regrouper les valeurs comme on veut, $(2 + 3) + 4 = 2 + (3 + 4)$ ;
commutative : on peut les prendre dans n’importe quel ordre, $2 + 3 = 3 + 2$ .

La somme, le comptage, le minimum et le maximum ont ces deux propriétés. La moyenne ne les a pas : la moyenne de $(2, 2, 2)$ et de $(10)$ vaut 4, mais la moyenne des deux moyennes, $(2 + 10) / 2$ , vaut 6. Pour calculer une moyenne, il faut émettre des couples (somme, effectif) et ne diviser qu’à la toute fin. L’exercice 6.6 le montre en détail.

6.7 Paralléliser sur les cœurs de la machine

Les tâches map étant indépendantes, on peut les exécuter en même temps. Deux notions sont nécessaires.

Un cœur est une unité de calcul du processeur : un processeur à 6 cœurs peut exécuter 6 calculs vraiment simultanés. La machine de salle sur laquelle ce chapitre a été mesuré (tp-1a201-18) a un Intel i7-8700 à 6 cœurs physiques, que le système présente comme 12 cœurs logiques : chaque cœur physique peut mener deux calculs de front, en profitant des instants où l’autre attend la mémoire (c’est l’hyper-threading). os.cpu_count() y affiche 12. Sur votre machine, lscpu vous donne ces nombres.
Un processus est un programme en cours d’exécution, avec sa propre mémoire. Deux processus ne partagent aucune variable : ils ne peuvent échanger des données qu’en se les envoyant.

Rappelons la mise en garde de la séance 2 : à cause du GIL, plusieurs threads d’un même programme Python ne calculent jamais en même temps. Pour utiliser plusieurs cœurs, il faut plusieurs processus. C’est ce que fait le module multiprocessing.

multiprocessing.Pool(p) crée un groupe (pool) de $p$ processus ouvriers. pool.map(fonction, liste) distribue les éléments de la liste à ces ouvriers, appelle fonction sur chacun, attend que tous aient fini, et renvoie la liste des résultats. C’est l’équivalent parallèle de [fonction(x) for x in liste].

mapreduce_parallele.py

import getpass
import multiprocessing
import os
import sys
import time

from mapreduce_partitionne import executer_map, executer_reduce
from mapreduce_simple import fonction_map, fonction_reduce

# /tmp est partage par tous les utilisateurs de la machine : on isole
# son travail dans un dossier a son nom.
DOSSIER = f"/tmp/mr-{getpass.getuser()}"
R = 4                      # nombre de reducteurs

def tache_map(args):
    id_map, nom_fichier = args
    executer_map(id_map, nom_fichier, fonction_map, R, DOSSIER)
    return id_map

def tache_reduce(args):
    r, nb_maps = args
    executer_reduce(r, nb_maps, fonction_reduce, DOSSIER)
    return r

if __name__ == "__main__":
    p = int(sys.argv[1])           # nombre de processus
    fichiers = sys.argv[2:]
    nb_maps = len(fichiers)
    os.makedirs(DOSSIER, exist_ok=True)

    debut = time.perf_counter()

    with multiprocessing.Pool(p) as pool:
        pool.map(tache_map, list(enumerate(fichiers)))
        t_map = time.perf_counter()
        pool.map(tache_reduce, [(r, nb_maps) for r in range(R)])

    fin = time.perf_counter()
    print(f"map    : {t_map - debut:.2f} s")
    print(f"reduce : {fin - t_map:.2f} s")
    print(f"total  : {fin - debut:.2f} s")

On le lance en donnant le nombre de processus, puis les fichiers. Sous le shell, $(cat fichiers.txt) est remplacé par le contenu du fichier, donc par la liste des chemins :

$ python3 mapreduce_parallele.py 4 $(cat fichiers.txt)
map    : 1.08 s
reduce : 0.12 s
total  : 1.20 s

Quelques détails du code :

getpass.getuser() renvoie votre login : chaque étudiant a son propre dossier dans /tmp, ce qui évite d’écraser les fichiers d’un camarade connecté à la même machine.
enumerate(fichiers) produit les couples (0, fichier0), (1, fichier1)… : chaque tâche map reçoit ainsi son numéro et son fichier.
Chaque fonction tache_... reçoit un seul argument, un couple, parce que pool.map ne transmet qu’un élément de la liste à la fois. Elle le défait aussitôt en deux variables.

Pourquoi nb_maps est passé en argument

Il serait tentant d’en faire une variable globale calculée dans le if __name__ == "__main__":. Ce serait un bug : un processus ouvrier n’entre jamais dans ce bloc, donc il ne connaîtrait pas cette variable et lèverait NameError (du moins avec les méthodes spawn et forkserver décrites ci-dessous).

Tout ce dont une tâche a besoin doit lui être passé en argument.

Le bloc if __name__ == "__main__": est obligatoire

Pour créer ses processus ouvriers, multiprocessing peut s’y prendre de trois façons, selon le système et la version de Python :

fork : l’ouvrier est une copie exacte du programme en cours. C’est la méthode par défaut sous Linux jusqu’à Python 3.13, donc sur les machines de l’école (Python 3.11) ;
spawn : l’ouvrier est un nouveau Python qui réexécute votre fichier depuis le début pour retrouver vos fonctions. C’est la méthode par défaut sous macOS et Windows ;
forkserver : un intermédiaire des deux, par défaut sous Linux à partir de Python 3.14.

Avec spawn, sans le if __name__ == "__main__":, chaque ouvrier qui réexécute le fichier tombe à son tour sur la création du Pool et tente d’en lancer un. Python l’interdit et lève une erreur, dans chaque ouvrier, sans fin. Testé sur un Mac : en 20 secondes, le programme a affiché 958 fois RuntimeError sans jamais se terminer.

Sur les machines de l’école, avec fork, le même programme fonctionne. L’oubli passe donc inaperçu, jusqu’au jour où vous lancez le code sur un Mac, sous Windows, ou sur une machine passée à Python 3.14. Mettez toujours le bloc.

La barrière entre les phases

Notez qu’il y a deux pool.map successifs, et non un seul. C’est essentiel : aucun réducteur ne peut commencer avant que tous les mappeurs aient fini, sinon il travaillerait sur des données incomplètes.

Ce point de synchronisation obligatoire s’appelle une barrière. Retenez-le : c’est la partie irréductiblement séquentielle de MapReduce, et donc la principale limite à son accélération. Nous la quantifierons en séance 7 avec la loi d’Amdahl.

6.8 MapReduce n’est pas que le comptage de mots

Le modèle s’applique à beaucoup de traitements. Il suffit de changer les deux fonctions.

Problème	map émet	reduce fait
Comptage de mots	`(mot, 1)`	la somme
Distribution des longueurs de mots	`(len(mot), 1)`	la somme
Longueur moyenne des mots	`("longueur", (len(mot), 1))`	somme des longueurs / somme des effectifs
Index inversé	`(mot, fichier)`	la liste des fichiers
Répartition des langues	`(langue, taille)`	la somme
Maximum par catégorie	`(cat, valeur)`	le maximum
Doublons	`(ligne, 1)`	garde si somme $> 1$
Tri distribué	`(cle, ligne)`	renvoie tel quel

Les limites du modèle

MapReduce est mal adapté aux traitements itératifs (algorithmes d’apprentissage, parcours de graphes), car chaque itération impose de tout réécrire sur disque. C’est précisément pour cela que Spark a été créé : garder les données intermédiaires en mémoire entre les itérations. MapReduce reste néanmoins le socle conceptuel de tous ces outils.

6.9 Exercices

Les corrigés se trouvent en Section 6.10.

Exercice 6.1 — Le moteur séquentiel

L’objectif est de faire tourner le moteur mapreduce() et de vérifier qu’il compte juste, d’abord sur un exemple que l’on peut compter à la main, puis sur un vrai corpus.

Étape 1 — Un exemple vérifiable à la main.

Recopiez mapreduce_simple.py en entier, tel qu’il est donné dans le cours (les deux fonctions et le moteur).
Dans le même dossier, créez trois fichiers texte :
- f1.txt contenant le chat dort sur le tapis du salon
- f2.txt contenant le chien dort sur le tapis
- f3.txt contenant le chat et le chien dorment
Avant de lancer quoi que ce soit, comptez à la main, sur papier, combien de fois apparaît chaque mot dans les trois fichiers réunis.
Lancez python3 mapreduce_simple.py f1.txt f2.txt f3.txt et comparez avec votre comptage.

Étape 2 — Le vrai corpus.

Si ce n’est pas déjà fait, préparez le corpus (Section 6.2) : python3 telecharger_corpus.py puis python3 fabriquer_corpus.py 8 1.
Écrivez top20.py, qui lit la liste des fichiers dans fichiers.txt (un chemin par ligne), appelle mapreduce() sur ces fichiers, puis affiche le nombre de mots distincts, le nombre total de mots, et les 20 mots les plus fréquents avec leur effectif.

Indice : collections.Counter(dictionnaire).most_common(20) renvoie directement les 20 couples (mot, effectif) les plus fréquents.

Étape 3 — Comparer avec Unix. Lancez la chaîne de commandes Unix du cours sur les mêmes fichiers. Les effectifs sont-ils identiques à ceux de Python ? Trouvez au moins un mot dont l’effectif diffère, et expliquez pourquoi.

Exercice 6.2 — Changer les fonctions, pas le moteur

L’objectif est de constater qu’avec le même moteur, on calcule des choses très différentes, simplement en changeant les deux fonctions.

Dans un nouveau fichier fonctions_62.py, importez le moteur (from mapreduce_simple import mapreduce) et écrivez trois couples de fonctions map / reduce. Ne modifiez pas une seule ligne de mapreduce().

La distribution des longueurs de mots : combien de mots ont 2 lettres, combien en ont 3, etc. Indice : la clé est la longueur du mot.
Un index inversé : pour chaque mot, la liste des fichiers où il apparaît. C’est ainsi qu’un moteur de recherche retrouve les pages qui contiennent un mot. Indice : la valeur émise est le nom du fichier ; attention à ne pas émettre dix fois le même fichier pour un mot qui y apparaît dix fois.
La longueur moyenne des mots, selon leur première lettre. Indice : lisez l’encadré « Le combiner n’est pas toujours permis » ; que faut-il émettre pour que le calcul reste juste si l’on ajoute un jour un combiner ?

Pour que les trois versions découpent les mots comme fonction_map, écrivez d’abord une petite fonction nettoyer(contenu) qui produit les mots nettoyés, et réutilisez-la dans les trois map.

Testez sur f1.txt, f2.txt et f3.txt, puis vérifiez à la main au moins une valeur de chaque résultat.

Exercice 6.3 — Le partitionnement

L’objectif est de vérifier, par programme, les trois propriétés qui rendent le partitionnement correct.

Recopiez mapreduce_partitionne.py tel qu’il est donné dans le cours.
Écrivez verifier_partition.py, qui :
- prend les 6 premiers fichiers de fichiers.txt ( $M = 6$ ) ;
- lance les 6 tâches map puis les 3 tâches reduce ( $R = 3$ ), dans un dossier de travail à votre nom, par exemple f"/tmp/mr-verif-{getpass.getuser()}", que le programme vide au début (shutil.rmtree(dossier, ignore_errors=True)) ;
- vérifie les trois propriétés ci-dessous avec assert.
Les trois propriétés :
1. il y a exactement $6 \times 3 = 18$ fichiers intermédiaires (glob.glob liste les fichiers dont le nom correspond à un motif) ;
2. un mot donné n’apparaît que dans un seul des 3 fichiers de résultat ;
3. la réunion des 3 fichiers de résultat est identique au dictionnaire renvoyé par le moteur séquentiel de l’exercice 6.1 sur les mêmes 6 fichiers.

assert condition, "message" ne fait rien si la condition est vraie, et arrête le programme en affichant le message si elle est fausse. C’est la façon la plus simple d’écrire une vérification.

Questions. Pourquoi faut-il un dossier à votre nom dans /tmp, et pourquoi faut-il le vider au début ? Pourquoi la propriété b) est-elle indispensable pour que le comptage soit juste ?

Exercice 6.4 — Mesurer l’effet du combiner

L’objectif est de mesurer ce que le combiner fait gagner, et de vérifier qu’il ne change pas le résultat.

Écrivez mapreduce_combiner.py, qui contient une version de executer_map avec un paramètre de plus, combiner. Si combiner vaut True, elle agrège localement comme dans le cours ; s’il vaut False, elle écrit chaque couple tel quel. Dans les deux cas, elle renvoie le nombre de couples écrits sur le disque.
Écrivez mesurer_combiner.py, qui lance cette fonction sur le premier fichier de fichiers.txt, une fois sans combiner et une fois avec, dans deux dossiers différents à votre nom. Pour chaque cas, affichez le nombre de couples écrits, la taille totale des fichiers intermédiaires (os.path.getsize donne la taille d’un fichier en octets), et le nombre de clés du résultat final.
Vérifiez avec assert que les deux résultats finaux sont identiques.

Questions. Quel facteur de réduction obtenez-vous ? Quel lien voyez-vous entre le nombre de couples écrits avec combiner et le nombre de clés du résultat ? Recommencez sur un fichier de 8 Mo (python3 fabriquer_corpus.py 1 8) : le facteur change-t-il, et pourquoi ? Pensez ensuite à relancer python3 fabriquer_corpus.py 8 1 pour retrouver le corpus habituel.

Exercice 6.5 — Parallélisation et accélération

L’objectif est de mesurer le gain réel apporté par plusieurs processus, et de comprendre ses limites. Faites-le sur une machine de l’école, sur laquelle personne d’autre ne travaille (la commande uptime doit afficher une charge proche de 0).

Relevez le nombre de cœurs de la machine : lscpu donne le nombre de cœurs physiques (Core(s) per socket) et de cœurs logiques (CPU(s)).
Recopiez mapreduce_parallele.py, qui prend le nombre de processus en premier argument. Préparez 8 fichiers de 1 Mo : python3 fabriquer_corpus.py 8 1.

Mesurez les temps pour 1 à 8 processus, avec cette boucle :

$ for p in 1 2 3 4 5 6 7 8; do
>   echo "--- p=$p"
>   python3 mapreduce_parallele.py $p $(cat fichiers.txt)
> done

Faites-le deux fois, et notez dans un tableau, pour chaque $p$ , les temps map, reduce et total, puis l’accélération $S(p) = T(1) / T(p)$ , où $T$ est le temps total. Une accélération de 3 signifie que le calcul va 3 fois plus vite qu’avec un seul processus.

Questions.

Les temps pour $p = 4$ , $5$ et $6$ sont presque identiques. Pourquoi ajouter des processus ne sert-il à rien ici ? Indice : il y a 8 tâches map ; combien de tâches chaque processus doit-il traiter à la suite, pour $p = 4$ ? et pour $p = 5$ ?
Le temps reduce cesse de diminuer à partir de $p = 4$ . Pourquoi ?
Atteignez-vous une accélération de 8 avec 8 processus ? Sinon, qu’est-ce qui manque ?
Pourquoi faut-il absolument utiliser le même jeu de fichiers pour toutes les mesures ?

Exercice 6.6 — Un cas qui casse le combiner

L’objectif est de voir de ses yeux pourquoi la moyenne ne supporte pas un combiner naïf, et pourquoi le bug peut passer inaperçu.

Avec le moteur de l’exercice 6.1, calculez la longueur moyenne des mots. Chaque tâche map traite un fichier entier : « moyenner localement » revient donc à calculer la moyenne de chaque fichier.

Version naïve (moyenne_naive.py) : map émet un seul couple par fichier, ("longueur", moyenne_du_fichier) ; reduce fait la moyenne de ces moyennes.
Version correcte (moyenne_correcte.py) : map émet ("longueur", (somme_des_longueurs, nombre_de_mots)) ; reduce additionne les sommes, additionne les nombres de mots, et ne divise qu’à la fin.
Créez court.txt contenant le un de (3 mots de 2 lettres) et long.txt contenant gouverneur (1 mot de 10 lettres). Calculez à la main la vraie moyenne des 4 mots, puis lancez les deux versions sur ces deux fichiers.
Lancez enfin les deux versions sur le corpus : $(cat fichiers.txt).

Questions. Laquelle des deux versions est fausse sur court.txt et long.txt, et pourquoi ? Pourquoi les deux versions donnent-elles presque le même résultat sur le corpus ? Expliquez en trois phrases ce que cela dit du danger de ce bug.

6.10 Corrigés

Exercice 6.1 — Le moteur séquentiel

Étape 1 — L’exemple vérifiable à la main.

Comptage à la main sur les trois fichiers, soit 20 mots en tout : le apparaît 6 fois (2 dans chaque fichier) ; chat, dort, sur, tapis et chien 2 fois chacun ; du, salon, et et dorment 1 fois chacun. On vérifie le total : $6 + 5 \times 2 + 4 \times 1 = 20$ .

$ python3 mapreduce_simple.py f1.txt f2.txt f3.txt
       6  le
       2  chat
       2  dort
       2  sur
       2  tapis
       2  chien
       1  du
       1  salon
       1  et
       1  dorment

Le programme et le comptage à la main coïncident. À égalité d’effectif, l’ordre affiché est celui de la première rencontre : le tri de Python (sorted) laisse dans leur ordre d’origine les éléments égaux, et le dictionnaire conserve l’ordre dans lequel les clés ont été ajoutées.

Remarquez que dort et dorment sont deux mots différents pour le programme : il compare des chaînes de caractères, il ne connaît pas la grammaire.

Étape 2 — Le vrai corpus.

top20.py

import collections

from mapreduce_simple import mapreduce, fonction_map, fonction_reduce

with open("fichiers.txt", encoding="utf-8") as f:
    fichiers = [ligne.strip() for ligne in f if ligne.strip()]

compte = mapreduce(fichiers, fonction_map, fonction_reduce)
print(f"{len(compte)} mots distincts, {sum(compte.values())} occurrences")
for mot, n in collections.Counter(compte).most_common(20):
    print(f"{n:8d}  {mot}")

[ligne.strip() for ligne in f if ligne.strip()] lit le fichier ligne par ligne, retire le retour à la ligne final de chaque chemin, et ignore les lignes vides.
len(compte) est le nombre de clés du dictionnaire, donc de mots distincts ; sum(compte.values()) additionne tous les effectifs, donc compte tous les mots.
collections.Counter accepte un dictionnaire tout fait, et most_common(20) fait le tri à notre place.

Sortie mesurée sur le corpus fabriquer_corpus.py 8 1, identique sur un Mac et sur une machine de l’école :

18810 mots distincts, 1376484 occurrences
   63780  de
   35595  la
   31726  et
   30358  le
   28239  à
   26219  il
   17330  les
   17087  que
   16829  un
   14383  qui
   12598  dans
   12267  en
   11490  ce
   11407  une
   10975  ne
   10960  des
   10873  se
   10051  pas
    9926  lui
    9328  du

Les premiers mots sont tous des mots-outils (articles, prépositions, pronoms) : c’est le cas dans tous les textes, quelle que soit la langue.

Étape 3 — Unix et Python ne comptent pas tout à fait pareil. La chaîne Unix du cours donne de 64 020 fois (Python : 63 780) et surtout il 37 304 fois (Python : 26 219). L’écart vient de la définition du « mot » :

grep -oE '[[:alpha:]]+' coupe à tout caractère qui n’est pas une lettre. qu'il donne donc qu et il, et dit-il donne dit et il.
fonction_map coupe aux espaces, puis retire la ponctuation aux deux extrémités seulement. qu'il et dit-il restent des mots entiers.

On le vérifie : Python compte 7 343 mots collés par une apostrophe et finissant par il (qu'il 6 288 fois, s'il 751 fois…), et 3 580 mots finissant par -il (dit-il 897 fois…). Avec les 26 219 il isolés, on obtient 37 142, soit presque exactement les 37 304 d’Unix.

Aucune des deux méthodes n’est « fausse » ; elles ne définissent pas le mot de la même façon. La leçon est générale : pour comparer deux programmes, il faut qu’ils découpent l’entrée exactement de la même manière. C’est pourquoi les séances 7 et 8 reprennent fonction_map à l’identique : le résultat de top20.py servira de référence pour valider la version distribuée.

Exercice 6.2 — Changer les fonctions

fonctions_62.py

import sys

from mapreduce_simple import mapreduce

SIGNES = ".,;:!?\"'()[]"

def nettoyer(contenu):
    """Les mots du texte, decoupes exactement comme dans fonction_map."""
    for mot in contenu.split():
        mot = mot.strip(SIGNES).lower()
        if mot:
            yield mot

# 1. Distribution des longueurs de mots
def map_longueurs(nom, contenu):
    for mot in nettoyer(contenu):
        yield (len(mot), 1)

def reduce_somme(cle, valeurs):
    return sum(valeurs)

# 2. Index inverse
def map_index(nom, contenu):
    for mot in set(nettoyer(contenu)):      # chaque mot une fois par fichier
        yield (mot, nom)

def reduce_index(cle, valeurs):
    return sorted(valeurs)

# 3. Longueur moyenne par lettre initiale
def map_moyenne(nom, contenu):
    for mot in nettoyer(contenu):
        yield (mot[0], (len(mot), 1))       # (somme, effectif)

def reduce_moyenne(cle, valeurs):
    somme = sum(s for s, _ in valeurs)
    effectif = sum(n for _, n in valeurs)
    return somme / effectif

if __name__ == "__main__":
    fichiers = sys.argv[1:]

    print("1. Longueurs :")
    for longueur, n in sorted(mapreduce(fichiers, map_longueurs,
                                        reduce_somme).items()):
        print(f"   {longueur:2d} lettres : {n} mots")

    print("2. Index inverse :")
    for mot, ou in sorted(mapreduce(fichiers, map_index,
                                    reduce_index).items()):
        print(f"   {mot:8s} -> {', '.join(ou)}")

    print("3. Longueur moyenne par initiale :")
    for lettre, moy in sorted(mapreduce(fichiers, map_moyenne,
                                        reduce_moyenne).items()):
        print(f"   {lettre} : {moy:.2f}")

$ python3 fonctions_62.py f1.txt f2.txt f3.txt
1. Longueurs :
    2 lettres : 8 mots
    3 lettres : 2 mots
    4 lettres : 4 mots
    5 lettres : 5 mots
    7 lettres : 1 mots
2. Index inverse :
   chat     -> f1.txt, f3.txt
   chien    -> f2.txt, f3.txt
   dorment  -> f3.txt
   dort     -> f1.txt, f2.txt
   du       -> f1.txt
   et       -> f3.txt
   le       -> f1.txt, f2.txt, f3.txt
   salon    -> f1.txt
   sur      -> f1.txt, f2.txt
   tapis    -> f1.txt, f2.txt
3. Longueur moyenne par initiale :
   c : 4.50
   d : 4.25
   e : 2.00
   l : 2.00
   s : 3.67
   t : 5.00

1. Les longueurs. La clé est la longueur, la valeur 1 : reduce fait la somme, comme pour le comptage de mots. Vérification à la main pour 2 lettres : le (6 fois), du et et font bien 8 mots. Et le total, $8 + 2 + 4 + 5 + 1 = 20$ , retrouve les 20 mots.

2. L’index inversé. La valeur émise est le nom du fichier. Le set(...) garde chaque mot une seule fois par fichier : sans lui, le, présent deux fois dans f1.txt, aurait la liste [f1.txt, f1.txt, …]. sorted donne une liste dans un ordre stable, ce qui rend le résultat facile à comparer.

3. La moyenne par initiale. On émet le couple (longueur, 1), puis reduce additionne séparément les longueurs et les effectifs avant de diviser une seule fois. Vérification pour c : chat, chat, chien, chien, soit $(4 + 4 + 5 + 5) / 4 = 4{,}50$ .

Aujourd’hui, notre moteur n’a pas de combiner, et émettre directement len(mot) puis faire la moyenne dans reduce donnerait le même résultat. Mais dès qu’on ajoutera un combiner (comme dans mapreduce_partitionne.py), seule la version (somme, effectif) restera juste : les sommes et les effectifs s’additionnent sans erreur, les moyennes non (exercice 6.6).

Exercice 6.3 — Partitionnement

verifier_partition.py

import getpass
import glob
import json
import os
import shutil

from mapreduce_partitionne import executer_map, executer_reduce
from mapreduce_simple import mapreduce, fonction_map, fonction_reduce

DOSSIER = f"/tmp/mr-verif-{getpass.getuser()}"
R = 3

with open("fichiers.txt", encoding="utf-8") as f:
    fichiers = [ligne.strip() for ligne in f if ligne.strip()][:6]

shutil.rmtree(DOSSIER, ignore_errors=True)    # repartir d'un dossier vide
os.makedirs(DOSSIER)
for id_map, nom in enumerate(fichiers):
    executer_map(id_map, nom, fonction_map, R, DOSSIER)
for r in range(R):
    executer_reduce(r, len(fichiers), fonction_reduce, DOSSIER)

# --- 1. M x R fichiers intermediaires ---
intermediaires = glob.glob(os.path.join(DOSSIER, "map-*-part-*"))
assert len(intermediaires) == len(fichiers) * R
print(f"1. {len(fichiers)} x {R} = {len(intermediaires)} fichiers "
      f"intermediaires")

# --- 2. chaque cle dans un seul fichier de resultat ---
fusion = {}
for r in range(R):
    chemin = os.path.join(DOSSIER, f"resultat-part-{r:02d}")
    with open(chemin, encoding="utf-8") as f:
        for ligne in f:
            cle, valeur = json.loads(ligne)
            assert cle not in fusion, f"{cle} vu dans deux reducteurs"
            fusion[cle] = valeur
print(f"2. {len(fusion)} cles, chacune dans un seul fichier de resultat")

# --- 3. meme resultat que le moteur sequentiel ---
reference = mapreduce(fichiers, fonction_map, fonction_reduce)
assert fusion == reference
print(f"3. identique au moteur sequentiel "
      f"({sum(fusion.values())} occurrences)")

Sortie, identique sur un Mac et sur une machine de l’école :

$ python3 verifier_partition.py
1. 6 x 3 = 18 fichiers intermediaires
2. 18661 cles, chacune dans un seul fichier de resultat
3. identique au moteur sequentiel (1032265 occurrences)

Le programme, partie par partie.

glob.glob(".../map-*-part-*") renvoie la liste des fichiers dont le nom correspond au motif, * remplaçant n’importe quelle suite de caractères. On vérifie qu’il y en a $M \times R = 18$ .
On lit les 3 fichiers de résultat et on range chaque mot dans le dictionnaire fusion. Avant de ranger un mot, l’assert vérifie qu’il n’y est pas déjà : s’il y était, c’est qu’il figure dans deux fichiers de résultat.
On compare fusion au résultat du moteur séquentiel. fusion == reference compare deux dictionnaires en entier : mêmes clés, et même valeur pour chaque clé. Un seul mot mal compté suffirait à faire échouer la vérification.

Pourquoi un dossier à votre nom, vidé au début. /tmp est partagé par tous les utilisateurs d’une machine. Avec un nom commun comme /tmp/mr-verif, le premier étudiant crée le dossier, et le second reçoit PermissionError en essayant d’y écrire (le même piège qu’en séance 2). Le vider au début, avec shutil.rmtree, évite un autre piège : si vous relancez après avoir changé $R$ , les fichiers de l’exécution précédente sont encore là, et le décompte de la propriété 1 est faux.

Pourquoi la propriété 2 est indispensable. Si le mot chat était traité par deux réducteurs, chacun ne verrait qu’une partie de ses occurrences et produirait un comptage partiel, par exemple chat: 12 d’un côté et chat: 30 de l’autre. Personne ne ferait la somme finale, et le résultat serait faux. La formule hash(cle) % R l’empêche : toutes les occurrences d’une même clé, quelle que soit la tâche map qui les a produites, arrivent au même réducteur.

Exercice 6.4 — Effet du combiner

Le executer_map du cours agrège d’office : pour comparer, il faut d’abord rendre le combiner optionnel. Une seule fonction, deux branches, et elle renvoie le nombre de paires écrites.

mapreduce_combiner.py

import json
import os

from mapreduce_partitionne import partition

def executer_map(id_map, nom_fichier, fonction_map, R, dossier,
                 combiner=True):
    """Comme dans le cours, mais le combiner devient optionnel.

    Renvoie le nombre de paires reellement ecrites sur disque.
    """
    with open(nom_fichier, encoding="utf-8", errors="ignore") as f:
        contenu = f.read()

    if combiner:
        seaux = [dict() for _ in range(R)]
        for cle, valeur in fonction_map(nom_fichier, contenu):
            r = partition(cle, R)
            seaux[r][cle] = seaux[r].get(cle, 0) + valeur
        paires = [list(seau.items()) for seau in seaux]
    else:
        paires = [[] for _ in range(R)]
        for cle, valeur in fonction_map(nom_fichier, contenu):
            paires[partition(cle, R)].append((cle, valeur))

    ecrites = 0
    for r in range(R):
        chemin = os.path.join(dossier, f"map-{id_map:03d}-part-{r:02d}")
        with open(chemin, "w", encoding="utf-8") as f:
            for cle, valeur in paires[r]:
                f.write(json.dumps([cle, valeur]) + "\n")
                ecrites += 1
    return ecrites

Le relevé lance les deux variantes sur le même fichier, chacune dans son propre dossier à votre nom, mesure les couples et les octets, puis compare les résultats finaux.

mesurer_combiner.py

import getpass
import glob
import json
import os
import shutil

from mapreduce_combiner import executer_map
from mapreduce_partitionne import executer_reduce
from mapreduce_simple import fonction_map, fonction_reduce

R = 3
with open("fichiers.txt", encoding="utf-8") as f:
    fichiers = [ligne.strip() for ligne in f if ligne.strip()][:1]

resultats = []
for combiner in (False, True):
    dossier = f"/tmp/mr-combiner-{getpass.getuser()}-{int(combiner)}"
    shutil.rmtree(dossier, ignore_errors=True)
    os.makedirs(dossier)
    paires = sum(executer_map(i, nom, fonction_map, R, dossier, combiner)
                 for i, nom in enumerate(fichiers))
    octets = sum(os.path.getsize(chemin) for chemin
                 in glob.glob(os.path.join(dossier, "map-*-part-*")))

    for r in range(R):
        executer_reduce(r, len(fichiers), fonction_reduce, dossier)
    final = {}
    for r in range(R):
        chemin = os.path.join(dossier, f"resultat-part-{r:02d}")
        with open(chemin, encoding="utf-8") as f:
            for ligne in f:
                cle, valeur = json.loads(ligne)
                final[cle] = valeur
    resultats.append(final)

    print(f"combiner={str(combiner):5s} : {paires:7d} paires, "
          f"{octets / 1024 / 1024:5.2f} Mo, {len(final)} cles")

assert resultats[0] == resultats[1]
print("resultat final identique dans les deux cas")

Sorties mesurées, identiques sur un Mac et sur une machine de l’école :

$ python3 fabriquer_corpus.py 8 1 ; python3 mesurer_combiner.py
combiner=False :  171952 paires,  2.22 Mo, 18179 cles
combiner=True  :   18179 paires,  0.31 Mo, 18179 cles
resultat final identique dans les deux cas

$ python3 fabriquer_corpus.py 1 8 ; python3 mesurer_combiner.py
combiner=False : 1355325 paires, 17.61 Mo, 31355 cles
combiner=True  :   31355 paires,  0.55 Mo, 31355 cles
resultat final identique dans les deux cas

Le facteur de réduction. Sur 1 Mo, le volume intermédiaire passe de 2,22 Mo à 0,31 Mo, soit un facteur 7 environ. Sur 8 Mo, de 17,61 Mo à 0,55 Mo, soit un facteur 32.

Le lien entre couples et clés. Sans combiner, on écrit un couple par mot du texte : 171 952 couples pour 171 952 mots. Avec combiner, un couple par mot distinct : 18 179 couples, exactement le nombre de clés du résultat.

Pourquoi le facteur grandit avec la taille. Multiplier la taille du texte par 8 multiplie le nombre de mots par 8 (171 952 → 1 355 325), mais le nombre de mots distincts n’augmente que de 72 % (18 179 → 31 355) : on retrouve sans cesse les mêmes mots, et les mots nouveaux deviennent de plus en plus rares. Le combiner, qui écrit un couple par mot distinct, est donc d’autant plus efficace que les données sont volumineuses. C’est justement le cas du Big Data.

Le résultat final est strictement identique : le combiner ne change que le volume transféré, jamais la valeur calculée, parce que la somme est associative et commutative.

Exercice 6.5 — Accélération

Mesures réalisées sur tp-1a201-18, une machine de salle libre : Intel i7-8700, 6 cœurs physiques, 12 logiques, Python 3.11. Huit fichiers de 1 Mo ( $M = 8$ ), quatre réducteurs ( $R = 4$ ). Trois séries ont été lancées ; elles diffèrent de moins de 0,02 s, on donne la première.

$p$	map (s)	reduce (s)	total $T(p)$ (s)	$S(p) = T(1)/T(p)$
1	4,28	0,48	4,75	1,00
2	2,17	0,23	2,41	1,97
3	1,61	0,23	1,84	2,58
4	1,08	0,12	1,20	3,96
5	1,09	0,12	1,21	3,93
6	1,08	0,12	1,20	3,96
7	0,97	0,12	1,09	4,36
8	0,55	0,12	0,67	7,09

a) Des paliers, pas une courbe. Une tâche map traite 1 Mo en 0,54 s environ. Les 8 tâches se répartissent entre les $p$ processus, et un processus qui a fini prend la tâche suivante. Le temps de la phase map est donc fixé par le nombre de tâches que le processus le plus chargé doit traiter à la suite, c’est-à-dire le nombre de « vagues » :

$p$	Répartition des 8 tâches	Vagues	Map prévu	Map mesuré
1	8	8	$8 \times 0{,}54 = 4{,}3$ s	4,28 s
2	4 + 4	4	2,2 s	2,17 s
3	3 + 3 + 2	3	1,6 s	1,61 s
4	2 + 2 + 2 + 2	2	1,1 s	1,08 s
5	2 + 2 + 2 + 1 + 1	2	1,1 s	1,09 s
6	2 + 2 + 1 + 1 + 1 + 1	2	1,1 s	1,08 s
8	1 + 1 + … + 1	1	0,54 s	0,55 s

Avec 5 ou 6 processus, il reste toujours au moins un processus qui doit traiter deux tâches l’une après l’autre : la phase dure autant qu’avec 4. Les processus en plus finissent tôt et attendent. Le nombre de vagues vaut $\lceil 8 / p \rceil$ , c’est-à-dire 8 divisé par $p$ arrondi à l’entier supérieur.

$p = 7$ est un peu plus rapide que prévu (0,97 s au lieu de 1,1 s) : la huitième tâche démarre dès que le premier processus est libre, et les sept autres n’ont pas toutes commencé au même instant. C’est un effet de second ordre, sans importance pour la conclusion.

b) Le reduce plafonne à 4. Il n’y a que $R = 4$ tâches reduce. La même règle des vagues s’applique : 4 vagues avec 1 processus (0,48 s), 2 vagues avec 2 ou 3 processus (0,23 s), et 1 seule dès 4 processus (0,12 s). Au-delà, les processus supplémentaires n’ont rien à faire pendant cette phase.

c) Pas tout à fait 8. Avec 8 processus, on obtient 7,09 et non 8. La phase map, elle, est presque parfaite : 0,55 s, contre $4{,}28 / 8 = 0{,}535$ s. Ce qui manque :

le reduce, qui ne peut pas aller plus de 4 fois plus vite, faute de tâches : il pèse 0,12 s sur 0,67 s, soit 18 % du temps total ;
la barrière : le reduce ne peut commencer qu’une fois toutes les tâches map finies ;
le coût fixe de création des processus et de transmission des arguments.

La leçon : l’accélération est limitée par le nombre de tâches autant que par le nombre de processeurs. Pour bien utiliser 12 cœurs, il faut au moins 12 tâches, et idéalement beaucoup plus. C’est l’idée de la loi d’Amdahl, que nous formaliserons en séance 7.

d) Le même jeu de fichiers. L’accélération compare des temps pour le même travail. Si le volume changeait d’une mesure à l’autre, on mélangerait deux effets (plus de processus, et moins ou plus de données), et le rapport $T(1)/T(p)$ n’aurait plus aucun sens.

Pour aller plus loin : 32 fichiers et 12 processus

Sur la même machine, avec 32 fichiers de 1 Mo (fabriquer_corpus.py 32 1) : $T(1) = 18{,}7$ s, $T(6) = 3{,}6$ s et $T(12) = 1{,}8$ s, soit une accélération de 10 avec 12 processus. Pour ce travail, les 12 cœurs logiques sont presque aussi utiles que des cœurs physiques, ce qui n’est pas le cas pour tous les calculs : seule la mesure permet de le savoir.

Exercice 6.6 — Le combiner qui casse

Les deux versions n’utilisent que le moteur de l’exercice 6.1 : seules les deux fonctions changent. Chaque tâche map traite un fichier, donc « moyenner localement » revient à émettre une moyenne par fichier.

moyenne_naive.py

"""Version naive : la moyenne est calculee des la phase map."""
import sys

from mapreduce_simple import mapreduce

SIGNES = ".,;:!?\"'()[]"

def fonction_map(nom_fichier, contenu):
    longueurs = [len(m.strip(SIGNES)) for m in contenu.split()]
    longueurs = [n for n in longueurs if n]
    yield ("longueur", sum(longueurs) / len(longueurs))

def fonction_reduce(cle, valeurs):
    return sum(valeurs) / len(valeurs)      # moyenne des moyennes : FAUX

if __name__ == "__main__":
    print(mapreduce(sys.argv[1:], fonction_map, fonction_reduce))

moyenne_correcte.py

"""Version correcte : on transporte le couple (somme, effectif)."""
import sys

from mapreduce_simple import mapreduce

SIGNES = ".,;:!?\"'()[]"

def fonction_map(nom_fichier, contenu):
    somme = effectif = 0
    for mot in contenu.split():
        mot = mot.strip(SIGNES)
        if mot:
            somme += len(mot)
            effectif += 1
    yield ("longueur", (somme, effectif))

def fonction_reduce(cle, valeurs):
    somme = sum(s for s, _ in valeurs)
    effectif = sum(n for _, n in valeurs)
    return somme / effectif                 # une seule division, a la fin

if __name__ == "__main__":
    print(mapreduce(sys.argv[1:], fonction_map, fonction_reduce))

Pour rendre l’erreur visible, il faut des fichiers déséquilibrés : court.txt contient le un de (3 mots de 2 lettres) et long.txt contient gouverneur (1 mot de 10 lettres).

$ python3 moyenne_naive.py court.txt long.txt
{'longueur': 6.0}
$ python3 moyenne_correcte.py court.txt long.txt
{'longueur': 4.0}

Vraie moyenne, à la main : $(2 + 2 + 2 + 10) / 4 = 16 / 4 = 4$ .
Version naïve : la moyenne de court.txt vaut 2, celle de long.txt vaut 10, et la moyenne des moyennes vaut $(2 + 10) / 2 = 6$ .
Version correcte : map émet (6, 3) pour court.txt et (10, 1) pour long.txt ; reduce calcule $(6 + 10) / (3 + 1) = 4$ .

La version naïve est fausse : elle donne le même poids à un fichier de 3 mots et à un fichier d’un seul mot.

Sur le corpus, en revanche :

$ python3 moyenne_naive.py $(cat fichiers.txt)
{'longueur': 4.737425038857935}
$ python3 moyenne_correcte.py $(cat fichiers.txt)
{'longueur': 4.737424481505052}

Pourquoi le corpus de la séance ne révèle pas le bug

Les deux résultats ne diffèrent qu’à la septième décimale. Les 8 fichiers de fabriquer_corpus.py ont tous presque le même nombre de mots (171 952, 172 007, 172 100…) : les poids que la version naïve oublie sont presque égaux, et l’erreur s’annule presque. Un test sur ce corpus « montrerait » que la version naïve est juste.

Un bug qui ne se manifeste que sur des données déséquilibrées est bien pire qu’un bug systématique : il passe tous les tests, puis se réveille en production, le jour où un fichier est beaucoup plus gros que les autres.

En trois phrases. La moyenne n’est pas associative : la moyenne des moyennes perd l’information du nombre d’éléments de chaque groupe, et donne le même poids à un groupe de 3 éléments et à un groupe d’un seul. La correction consiste à transporter ce nombre avec la somme, sous forme de couple (somme, effectif), couple que l’on peut additionner sans erreur dans n’importe quel ordre. On ne divise qu’une seule fois, à la toute fin.

6.11 Ce qu’il faut retenir

L’essentiel de la séance

MapReduce est la version distribuée de map | sort | reduce : le sort d’Unix, c’est le shuffle.
MapReduce = le programmeur écrit map et reduce, le système fait tout le reste.
Trois phases : map $\rightarrow$ shuffle $\rightarrow$ reduce.
Les tâches map sont indépendantes : c’est ce qui rend la distribution et la reprise sur panne possibles.
Le shuffle répartit les clés par hash(cle) % R, formule déterministe qui évite toute concertation entre machines.
$M$ maps $\times$ $R$ reduces $= M \times R$ fichiers intermédiaires : le shuffle est la phase la plus coûteuse.
Le combiner (agrégation locale) est l’optimisation la plus rentable — mais seulement si reduce est associative et commutative.
La barrière entre map et reduce est obligatoire : elle est la part séquentielle du calcul.
L’accélération est limitée par le nombre de tâches : avec 8 tâches, 5 ou 6 processus ne font pas mieux que 4. Il faut plus de tâches que de processus.
En Python : multiprocessing pour le calcul, threading pour l’attente. Toujours le bloc if __name__ == "__main__":, et tout ce dont une tâche a besoin passé en argument.

6.1 Le problème de départ

6.2 Le corpus de travail

Récupérer quelques textes

Fabriquer des fichiers à la bonne taille

fichiers.txt

6.3 L’ancêtre : compter les mots avec des outils Unix

Pourquoi cela fonctionne si bien

Ce qui bloque, et ce que MapReduce apporte

6.4 Le modèle

Le comptage de mots dans ce modèle

Le déroulement, sur un exemple minuscule

6.5 Un moteur MapReduce en 25 lignes

6.6 Le shuffle : le vrai problème

La formule

Chaque map écrit RR fichiers

Le combiner

6.7 Paralléliser sur les cœurs de la machine

La barrière entre les phases

6.8 MapReduce n’est pas que le comptage de mots

6.9 Exercices

6.10 Corrigés

6.11 Ce qu’il faut retenir

`fichiers.txt`

Chaque map écrit $R$ fichiers