7 MapReduce distribué et performance

Objectifs de la séance

Faire tourner le MapReduce de la séance 6 sur plusieurs machines de l’école.
Partager proprement des machines communes : un port à son nom, un dossier à son nom, une connexion SSH à la fois, et un nettoyage qui fonctionne.
Comprendre pourquoi les fichiers intermédiaires vont sur le disque local (/tmp), et le vérifier par la mesure.
Mesurer, tracer une courbe d’accélération et l’interpréter avec la loi d’Amdahl.
Survivre à la panne d’une machine, et prouver que le résultat ne change pas.

Diaporama

7.1 L’architecture

Nous avons tous les morceaux. Il reste à les assembler :

Séance	Brique	Rôle ici
1	sockets, `bind`, `connect`	les machines se parlent
2	threads, timeouts, `vivantes.txt`	servir plusieurs demandes ; savoir quelles machines utiliser
3	`protocole.py`	échanger des messages JSON complets
5	horloge monotone, détection de panne	mesurer des durées ; repérer une machine morte
6	map / shuffle / reduce, `fichiers.txt`	le calcul lui-même, et les données

Le système comporte trois programmes.

Le travailleur (un par machine). C’est un serveur, au sens de la séance 1 : il écoute sur un port et attend des ordres. Il sait faire quatre choses : exécuter un map, exécuter un reduce, servir ses fichiers intermédiaires aux autres travailleurs, et rendre son résultat.

Le maître (un seul). C’est un client : il connaît la liste des fichiers à traiter (fichiers.txt) et la liste des travailleurs (travailleurs.txt). Il distribue les tâches, attend les réponses et recommence en cas d’échec. Il ne calcule rien.

Le déployeur. Un petit script qui démarre les travailleurs par SSH sur les machines de vivantes.txt, vérifie qu’ils répondent, et écrit travailleurs.txt. Il sait aussi tout arrêter et tout effacer.

Vocabulaire

La littérature dit master/workers, parfois encore master/slaves dans les textes anciens. On dit aujourd’hui main ou coordinateur et workers. Dans Hadoop, ce sont le ResourceManager et les NodeManagers.

Une tâche est une unité de travail : « fais le map du fichier 7 », ou « fais le reduce de la partition 3 ». Un mappeur est un travailleur en train d’exécuter un map, un réducteur un travailleur en train d’exécuter un reduce. Ici, chaque travailleur joue les deux rôles, l’un après l’autre.

Les fichiers de la séance

Tout se trouve dans un seul dossier, ~/bgd701, sur votre répertoire personnel. Celui-ci est partagé par toutes les machines de l’école (c’est le NFS, détaillé plus bas) : un fichier posé là est visible partout.

Fichier	D’où il vient	Rôle
`protocole.py`	séance 3	`envoyer_message`, `recevoir_message`
`mapreduce_simple.py`	séance 6, exercice 6.1	le moteur séquentiel, pour vérifier
`fabriquer_corpus.py`	séance 6 (Section 6.2)	fabrique `corpus/` et `fichiers.txt`
`vivantes.txt`	séance 2, exercice 2.4	les machines allumées et libres
`travailleur.py`	ce chapitre	le serveur qui calcule
`deployer.py`	ce chapitre	lance et arrête les travailleurs ; écrit `travailleurs.txt`
`maitre.py`	ce chapitre	distribue les tâches et rassemble le résultat

Le déroulement dans le temps

Étape	Ce qui se passe
0	Le déployeur lance un travailleur sur chaque machine, vérifie qu’il répond (`PING`), et écrit `travailleurs.txt`
1	Le maître envoie une tâche `MAP` à chaque travailleur
2	Chaque travailleur lit son fichier, applique `map`, écrit $R$ fichiers sur son disque local
3	Le travailleur répond `OK`, le maître lui envoie la tâche suivante
4	Barrière : le maître attend que tous les maps soient finis
5	Il envoie les $R$ tâches `REDUCE`, en indiquant où se trouve chaque fichier intermédiaire
6	Chaque réducteur va lire à distance sa partition chez chaque mappeur (`LIRE_PARTITION`)
7	Il additionne, et écrit son résultat sur son disque local
8	Le maître collecte les $R$ résultats (`RESULTAT`) et les fusionne

L’étape 6 est le shuffle distribué. C’est la vraie nouveauté par rapport à la séance 6 : les fichiers intermédiaires ne sont plus dans un même dossier, ils sont dispersés sur $N$ machines. Avec $M = 32$ fichiers et $R = 8$ réducteurs, chaque réducteur fait 32 demandes, une par map : $32 \times 8 = 256$ demandes en tout.

M \times R

: un maximum ou un compte exact ?

Dans notre code, c’est un compte exact. Chaque map écrit ses $R$ fichiers, même vides, et chaque réducteur les demande tous, même quand ils sont vides. Vérifié sur 8 fichiers, 4 travailleurs et $R = 8$ : 64 demandes, dont 5 pour des partitions vides (un fichier ne contenait que trois mots différents).

Toutes ne traversent pas le réseau. Quand le réducteur est aussi le mappeur, la demande reste sur la même machine. Avec $p$ machines, cela arrive environ une fois sur $p$ : 16 demandes sur 64 avec 4 machines. Pour 32 fichiers et 16 machines, environ $256 \times (1 - 1/16) = 240$ demandes passent par le réseau.

Un système optimisé en ferait moins, et $M \times R$ devient alors un maximum : ne pas demander les partitions vides (il suffit que le map signale lesquelles le sont), ou regrouper en une seule demande tout ce qu’une même machine possède pour un réducteur. Hadoop fait ce regroupement.

Un cas extrême, mesuré. 30 fichiers qui ne contiennent qu’un seul mot, chat, 1 000 fois en tout, avec $R = 8$ . Le mot tombe dans la partition 5. Notre code écrit quand même $30 \times 8 = 240$ fichiers intermédiaires et fait 240 demandes : 210 rapportent un dictionnaire vide, et seules 30 transportent des données, toutes vers le réducteur 5. Un système optimisé ne ferait que ces 30 transferts. Mais le vrai défaut de ce cas est ailleurs : le réducteur 5 fait tout le travail et les 7 autres attendent. Quand une clé domine les données, le reduce n’est plus parallèle du tout, quel que soit le nombre de machines : c’est un straggler fabriqué par les données elles-mêmes.

Faut-il vraiment lancer les $R$ reduces ? Non : ici, un seul suffit. Il faut autant de reduces que de partitions non vides, soit au plus le plus petit de $R$ et du nombre de clés distinctes. $R$ doit être fixé avant les maps, qui en ont besoin pour calculer hash(cle) % R. Mais après la barrière, le maître pourrait savoir quelles partitions sont vides, si chaque map le signalait dans sa réponse OK, et ne lancer que les reduces utiles. Hadoop lance les $R$ reduces quoi qu’il arrive ; Spark, dans son mode adaptatif, regarde la taille des partitions après les maps et regroupe les petites. Une limite demeure : une clé ne se découpe pas entre plusieurs réducteurs. Ce qui sauve ce cas, c’est le combiner : chaque map envoie ("chat", 33) au lieu de 33 couples ("chat", 1).

Enfin, ce qui coûte vraiment, c’est le volume en octets, et lui dépend des données : beaucoup de mots différents, de grosses partitions.

7.2 Où écrire les fichiers intermédiaires ?

C’est le point le plus important de la séance, et l’erreur la plus fréquente.

Votre répertoire personnel n’est pas sur la machine

Sur les machines de l’école, votre répertoire personnel (~/) est monté par le réseau : c’est un NFS (Network File System, voir la séance 6). Un serveur central héberge les fichiers de tout le monde.

Chemin	Où sont réellement les octets	Visible depuis
`~/donnees.txt`	sur le serveur NFS, à travers le réseau	toutes les machines
`/tmp/donnees.txt`	sur le disque local de la machine	cette machine seulement

Ce que coûte une écriture sur le NFS

Mesure faite sur tp-1a201-18 avec le script de l’exercice 7.2 : on écrit 64 petits fichiers JSON (3,1 Mo en tout, la taille des fichiers intermédiaires de huit maps), d’abord dans /tmp, puis dans ~.

Dossier	Durée, essai 1	Durée, essai 2
`/tmp` (disque local)	0,30 s	0,30 s
`~` (NFS)	1,06 s	2,60 s

Deux leçons. Le NFS est plusieurs fois plus lent. Et il est irrégulier : la même mesure varie du simple au double, parce que le serveur est partagé avec toute l’école et que sa charge change d’une minute à l’autre. Le disque local, lui, ne dépend que de votre machine.

Pourquoi. Quand un programme ferme un fichier sur le NFS, le système envoie les données au serveur et attend sa confirmation avant de rendre la main : c’est ce qui garantit qu’une autre machine qui ouvre le fichier juste après verra la bonne version. Sur un disque local, la fermeture rend la main dès que les données sont en mémoire ; le système les écrira sur le disque un peu plus tard, sans faire attendre le programme.

Le vrai problème : l’échelle

Attention

Avec $N$ machines qui écrivent dans /tmp, on a $N$ disques qui travaillent en même temps : la capacité grandit avec le nombre de machines. Avec $N$ machines qui écrivent dans ~/, on a $N$ machines qui écrivent sur le même serveur : la capacité reste celle d’un seul serveur, partagée entre vous, vos camarades, et tous les utilisateurs de l’école.

Le disque local passe à l’échelle ; le NFS, non. À cent machines, vous auriez transformé cent machines en une seule, et ralenti le répertoire personnel de tout le monde.

À retenir

Les fichiers intermédiaires vont sur le disque local, dans /tmp. C’est exactement ce que dit la figure 1 de l’article de Google : intermediate files (on local disks).

Le répertoire personnel NFS sert aux trois autres usages, et à eux seuls :

déployer le programme — une fois ;
lire les données d’entrée — chaque travailleur ouvre lui-même les fichiers de fichiers.txt, qui doivent donc être visibles partout ;
récupérer le résultat final — une fois.

Ce qui distingue les deux, c’est le volume : les intermédiaires sont $M \times R$ fichiers, écrits puis relus, le plus souvent par d’autres machines ; l’entrée et la sortie sont lues ou écrites une seule fois.

Prenez l’habitude d’un dossier qui porte votre nom : /tmp est commun à tous les utilisateurs d’une machine (séance 2).

import getpass, os
DOSSIER = f"/tmp/bgd701-{getpass.getuser()}"
os.makedirs(DOSSIER, exist_ok=True)

getpass.getuser() renvoie votre identifiant de connexion (votre login).

Explorer les disques d’une machine

/tmp n’est pas forcément le seul espace local disponible, ni le plus grand. La commande df -h liste les systèmes de fichiers montés et leur taille ; mount indique lesquels viennent du réseau (leur source contient un :, par exemple serveur:/home). Cela vaut la peine de regarder avant de lancer un gros calcul.

7.3 Partager les machines avec les autres

Pendant la séance, une trentaine de binômes déploient en même temps, et souvent sur les mêmes machines : tout le monde a obtenu à peu près le même vivantes.txt, trié par nom. Sans précaution, vos programmes se marchent dessus. Quatre règles évitent tous les problèmes que nous avons rencontrés en préparant cette séance.

Règle 1 : un port à votre nom

Supposons que tout le monde utilise le port 50007, comme dans les séances précédentes. Le premier travailleur lancé sur tp-1a201-05 ouvre le port. Le second, celui de votre camarade, meurt aussitôt :

OSError: [Errno 98] Address already in use

Ce message part dans le fichier journal, sur la machine distante : personne ne le voit. La commande ssh qui a lancé le travailleur, elle, a réussi. Lors de nos essais, un déployeur qui se contentait de vérifier le succès de ssh affichait « 3/3 machines » alors qu’un des trois travailleurs était mort au démarrage. Pire : le maître de votre camarade enverra ses tâches à votre travailleur, puisque c’est lui qui répond sur ce port.

La parade : chaque utilisateur a un numéro unique, son uid (user identifier), le même sur toutes les machines de l’école puisque les comptes sont centralisés. La commande id -u l’affiche. On en déduit un port qui n’appartient qu’à vous :

PORT_PERSONNEL = 40000 + os.getuid() % 20000

% est le reste de la division : le port tombe toujours entre 40000 et 59999, loin des ports des services habituels (22 pour SSH, 80 pour le Web…). Pour l’uid 11114, cela donne le port 51114.

Règle 2 : vérifier que c’est votre travailleur qui répond

Le succès de la commande ssh ne prouve rien : elle rend la main dès que le travailleur est lancé en arrière-plan, avant de savoir s’il a pu ouvrir son port. Le seul test fiable est de lui parler.

Notre travailleur répond au message PING par un PONG qui contient le nom de son propriétaire :

{"type": "PONG", "utilisateur": "login"}

Le déployeur ne garde une machine que si le PONG porte votre login. Deux comptes peuvent, rarement, tomber sur le même port (deux uid qui diffèrent d’un multiple de 20 000) : le PING le détecte, et le déployeur passe simplement à une autre machine.

Règle 3 : une connexion SSH à la fois

L’école limite le nombre de sessions SSH simultanées d’un même utilisateur : au-delà, les connexions sont refusées. C’est ce qu’annonçait la séance 2. Notre déployeur ouvre donc une connexion, lance la commande, referme, attend que le travailleur réponde, puis passe à la machine suivante. Jamais deux à la fois.

C’est plus lent qu’en parallèle : 19 s pour 16 machines dans nos mesures. Mais c’est un coût fixe, payé une fois par séance : les travailleurs restent ensuite en place pour tous les calculs.

Règle 4 : nettoyer — et vérifier que le nettoyage a marché

Un travailleur lancé avec nohup ne s’arrête jamais tout seul. Si vous partez sans l’arrêter, il occupe la machine et votre port jusqu’au redémarrage. Et les fichiers de /tmp restent.

La commande qui vient naturellement est :

$ ssh tp-1a201-05.enst.fr "pkill -f travailleur.py ; rm -rf /tmp/bgd701-\$USER"

Elle ne marche pas, et c’est un piège très instructif. pkill -f motif tue tous les processus dont la ligne de commande contient le motif. Or, sur la machine distante, ssh exécute votre commande dans un shell, dont la ligne de commande est :

zsh -c pkill -f travailleur.py ; rm -rf /tmp/bgd701-$USER

Elle contient travailleur.py ! pkill tue donc le travailleur… et le shell qui l’a lancé. Le rm qui suivait n’est jamais exécuté, et ssh renvoie le code d’erreur 255. Lors de nos essais, 265 fichiers étaient restés dans /tmp, sans aucun message visible.

La bonne commande :

$ ssh tp-1a201-05.enst.fr "pkill -u \$USER -f '[t]ravailleur.py' ; rm -rf /tmp/bgd701-\$USER"

Élément	Rôle
`-u \$USER`	seulement vos processus, jamais ceux des autres
`[t]ravailleur.py`	une expression régulière : `[t]` signifie « la lettre t ». Le motif reconnaît `travailleur.py` dans la ligne du travailleur, mais pas dans celle du shell, qui contient `[t]ravailleur.py`, avec les crochets
`'...'`	les apostrophes sont obligatoires : sans elles, le shell des machines de l’école (zsh) prend `[t]ravailleur.py` pour un motif de noms de fichiers, ne trouve aucun fichier, affiche `zsh: no matches found` et n’exécute rien
`\$USER`	le `\` empêche votre shell de remplacer `$USER` : c’est la machine distante qui le fera

Vérifiez le nettoyage avec un test qui ne peut pas se tromper

pgrep -f travailleur.py, lancé par ssh, souffre exactement du même mal : il trouve son propre shell, et vous croirez qu’il reste un travailleur. Le test fiable consiste à essayer de se connecter au port : si personne ne répond, le travailleur est bien arrêté.

À retenir

Écrivez le nettoyage en même temps que le déploiement, jamais après, et vérifiez qu’il a réellement fonctionné. Sinon, vous laisserez des processus orphelins sur les machines de l’école, qui gêneront tout le monde — y compris vous, au TP suivant.

7.4 Déployer sur $N$ machines

Depuis quelle machine travailler ?

Ouvrez une session sur une machine de l’école, ssh login@tp-1a201-XX.enst.fr, puis cd ~/bgd701. C’est de là que vous lancerez le déployeur et le maître. Comme le dossier est sur le NFS, vous pouvez aussi éditer les fichiers depuis n’importe quelle autre machine.

À la main : un `scp`, puis un `ssh` par machine

Si vous avez écrit vos programmes sur votre propre ordinateur, il faut les copier vers l’école. Comme votre répertoire personnel est partagé, une seule copie suffit pour qu’ils soient visibles sur toutes les machines. Si vous travaillez déjà dans ~/bgd701 sur une machine de l’école, il n’y a rien à copier.

# 1. UNE seule copie vers le NFS (depuis votre ordinateur)
$ scp travailleur.py deployer.py maitre.py login@tp-1a201-01.enst.fr:~/bgd701/

# 2. Lancer un travailleur, sur UNE machine
$ ssh tp-1a201-02.enst.fr "cd ~/bgd701 && \
      nohup python3 travailleur.py \
      > /tmp/bgd701-travailleur-\$USER.log 2>&1 &"

Chaque morceau de la deuxième commande a son rôle (séance 2) :

Élément	Rôle
`cd ~/bgd701`	se placer là où sont les programmes
`nohup`	no hangup : ignorer le signal envoyé quand la connexion se ferme
`> fichier.log`	envoyer l’affichage (la sortie standard) dans un fichier
`2>&1`	envoyer aussi les erreurs (sortie n°2) au même endroit que la sortie n°1
`&`	lancer en arrière-plan : le shell rend la main sans attendre

Sans la redirection, ssh ne rend jamais la main

ssh attend que le programme distant ait fermé sa sortie. Un travailleur en arrière-plan qui écrit encore sur la sortie de la connexion la garde ouverte : ssh reste bloqué indéfiniment. La redirection vers un fichier libère la connexion.

Un seul scp

Faire $N$ scp au lieu d’un seul est une erreur classique : cela copie $N$ fois le même fichier sur le même serveur NFS, ce qui est à la fois inutile et lent.

Le script de déploiement

deployer.py

"""Deploie (ou arrete) les travailleurs sur les machines de vivantes.txt.

    python3 deployer.py deployer 4     lance 4 travailleurs
    python3 deployer.py nettoyer       les arrete et efface /tmp

Produit travailleurs.txt : une ligne "machine:port" par travailleur qui a
reellement repondu. C'est ce fichier que lit le maitre.
"""
import getpass
import os
import random
import socket
import subprocess
import sys
import time

from protocole import envoyer_message, recevoir_message

DOSSIER_NFS = "~/bgd701"                      # la ou est travailleur.py
PORT = 40000 + os.getuid() % 20000            # le meme que le travailleur
MOI = getpass.getuser()


def ssh(machine, commande):
    """Une connexion SSH, une commande, puis on referme."""
    try:
        r = subprocess.run(["ssh", "-o", "BatchMode=yes",
                            "-o", "ConnectTimeout=5", machine, commande],
                           timeout=20, capture_output=True, text=True)
        return r.returncode == 0
    except subprocess.TimeoutExpired:
        return False


def repond(machine, delai=5.0):
    """Vrai si NOTRE travailleur repond sur cette machine."""
    fin = time.monotonic() + delai
    while time.monotonic() < fin:
        try:
            with socket.create_connection((machine, PORT), timeout=2) as s:
                envoyer_message(s, {"type": "PING"})
                reponse = recevoir_message(s)
            return reponse.get("utilisateur") == MOI
        except (OSError, ConnectionError, ValueError):
            time.sleep(0.5)                    # pas encore demarre
    return False


def deployer(machines, nombre):
    commande = (f"cd {DOSSIER_NFS} && nohup python3 travailleur.py {PORT} "
                f"> /tmp/bgd701-travailleur-{MOI}.log 2>&1 &")
    retenus = []
    for machine in machines:                   # UNE connexion a la fois
        if len(retenus) == nombre:
            break
        if ssh(machine, commande) and repond(machine):
            retenus.append(f"{machine}:{PORT}")
            print(f"  ok      {machine}")
        else:
            print(f"  echec   {machine}")
    with open("travailleurs.txt", "w") as f:
        f.write("\n".join(retenus) + "\n")
    print(f"{len(retenus)} travailleur(s) dans travailleurs.txt "
          f"(port {PORT})")


def nettoyer():
    machines = [l.strip().rsplit(":", 1)[0]
                for l in open("travailleurs.txt") if l.strip()]
    # [t] : sinon pkill tue aussi le shell distant, dont la ligne de
    # commande contient "travailleur.py", et le rm n'est jamais execute.
    commande = (f"pkill -u {MOI} -f '[t]ravailleur.py' ; "
                f"rm -rf /tmp/bgd701-{MOI} /tmp/bgd701-travailleur-{MOI}.log")
    for machine in machines:                   # UNE connexion a la fois
        print(f"  {'ok   ' if ssh(machine, commande) else 'echec'}  {machine}")


if __name__ == "__main__":
    action = sys.argv[1] if len(sys.argv) > 1 else "deployer"
    if action == "deployer":
        nombre = int(sys.argv[2]) if len(sys.argv) > 2 else 4
        machines = [l.strip() for l in open("vivantes.txt") if l.strip()]
        random.shuffle(machines)       # ne pas tous prendre les memes
        deployer(machines, nombre)
    else:
        nettoyer()

Ce qu’il faut remarquer.

ssh() lance la commande ssh du système grâce à subprocess.run. BatchMode=yes interdit toute question (mot de passe, confirmation) : si la clé ne suffit pas, on échoue tout de suite au lieu de rester bloqué. ConnectTimeout=5 et timeout=20 bornent l’attente.
repond() interroge le travailleur pendant au plus 5 secondes, le temps qu’il démarre. time.monotonic() est l’horloge monotone de la séance 5.
deployer() parcourt les machines une par une et s’arrête dès qu’il en a assez. Une machine qui échoue est simplement sautée.
random.shuffle mélange la liste : si tout le monde prenait les quatre premières machines de vivantes.txt, elles seraient surchargées.
travailleurs.txt contient des adresses complètes, machine:port. C’est ce fichier, et non vivantes.txt, que lit le maître.
nettoyer() relit travailleurs.txt pour savoir où passer. Ne modifiez pas ce fichier à la main, sinon des travailleurs seront oubliés.

Ce que l’on voit à l’écran (le port dépend de votre uid) :

$ python3 deployer.py deployer 4
  ok      tp-1a201-10.enst.fr
  ok      tp-1a201-27.enst.fr
  ok      tp-1a201-31.enst.fr
  ok      tp-1a201-13.enst.fr
4 travailleur(s) dans travailleurs.txt (port 51114)

$ cat travailleurs.txt
tp-1a201-10.enst.fr:51114
tp-1a201-27.enst.fr:51114
tp-1a201-31.enst.fr:51114
tp-1a201-13.enst.fr:51114

7.5 Le travailleur

travailleur.py

"""Travailleur du MapReduce distribue (seance 7).

    python3 travailleur.py            ecoute sur votre port personnel
    python3 travailleur.py 45678      ou sur le port indique
"""
import getpass
import hashlib
import json
import os
import socket
import sys
import threading

from protocole import envoyer_message, recevoir_message

# Chaque utilisateur a un numero unique (uid), le meme sur toutes les
# machines de l'ecole : on en deduit un port qui n'appartient qu'a vous.
PORT_PERSONNEL = 40000 + os.getuid() % 20000

DOSSIER = f"/tmp/bgd701-{getpass.getuser()}"   # disque LOCAL, a votre nom
os.makedirs(DOSSIER, exist_ok=True)


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


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


def ecrire_atomique(chemin, objet):
    """Ecrit dans un fichier .tmp puis le renomme : jamais a moitie ecrit."""
    with open(chemin + ".tmp", "w", encoding="utf-8") as f:
        json.dump(objet, f)
    os.replace(chemin + ".tmp", chemin)


def faire_map(requete):
    id_map, R = requete["id_map"], requete["R"]
    with open(requete["fichier"], encoding="utf-8", errors="ignore") as f:
        contenu = f.read()

    seaux = [dict() for _ in range(R)]          # combiner : un dict par r
    for cle, valeur in fonction_map(requete["fichier"], contenu):
        r = partition(cle, R)
        seaux[r][cle] = seaux[r].get(cle, 0) + valeur

    for r in range(R):
        ecrire_atomique(f"{DOSSIER}/map-{id_map:04d}-part-{r:02d}",
                        seaux[r])
    return {"type": "OK", "id_map": id_map}


def lire_partition(requete):
    """Sert une partition a un reducteur distant."""
    chemin = (f"{DOSSIER}/map-{requete['id_map']:04d}"
              f"-part-{requete['r']:02d}")
    if not os.path.exists(chemin):
        return {"type": "ERREUR", "message": f"partition absente : {chemin}"}
    with open(chemin, encoding="utf-8") as f:
        return {"type": "OK", "donnees": json.load(f)}


def faire_reduce(requete):
    """Va chercher la partition r chez chaque mappeur, puis additionne."""
    r = requete["r"]
    total = {}
    for travailleur, id_map in requete["sources"]:
        machine, port = travailleur.rsplit(":", 1)
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(20.0)
            s.connect((machine, int(port)))
            envoyer_message(s, {"type": "LIRE_PARTITION",
                                "id_map": id_map, "r": r})
            reponse = recevoir_message(s)
            s.close()
        except (OSError, ConnectionError, ValueError) as e:
            return {"type": "ERREUR", "message": str(e),
                    "machine": travailleur}
        if reponse.get("type") != "OK":
            return {"type": "ERREUR", "message": reponse.get("message"),
                    "machine": travailleur}
        for cle, valeur in reponse["donnees"].items():
            total[cle] = total.get(cle, 0) + valeur

    ecrire_atomique(f"{DOSSIER}/resultat-part-{r:02d}", total)
    return {"type": "OK", "r": r, "cles": len(total)}


def lire_resultat(requete):
    with open(f"{DOSSIER}/resultat-part-{requete['r']:02d}",
              encoding="utf-8") as f:
        return {"type": "OK", "donnees": json.load(f)}


def traiter(requete):
    t = requete.get("type")
    if t == "PING":
        return {"type": "PONG", "utilisateur": getpass.getuser()}
    if t == "MAP":
        return faire_map(requete)
    if t == "REDUCE":
        return faire_reduce(requete)
    if t == "LIRE_PARTITION":
        return lire_partition(requete)
    if t == "RESULTAT":
        return lire_resultat(requete)
    return {"type": "ERREUR", "message": f"type inconnu : {t}"}


def servir(connexion, adresse):
    try:
        while True:
            requete = recevoir_message(connexion)
            try:
                reponse = traiter(requete)
            except (OSError, KeyError, ValueError) as e:
                reponse = {"type": "ERREUR", "message": str(e)}
            envoyer_message(connexion, reponse)
    except (ConnectionError, OSError, ValueError):
        pass                     # le client est parti : rien a faire
    finally:
        connexion.close()


if __name__ == "__main__":
    port = int(sys.argv[1]) if len(sys.argv) > 1 else PORT_PERSONNEL
    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind(("0.0.0.0", port))
    srv.listen(64)
    print(f"Travailleur de {getpass.getuser()} pret sur le port {port}",
          flush=True)
    while True:
        c, a = srv.accept()
        threading.Thread(target=servir, args=(c, a), daemon=True).start()

Lecture guidée

Ce qui vient de la séance 6. partition et fonction_map sont identiques à celles de la séance 6, au caractère près : c’est ce qui permettra de comparer le résultat distribué au moteur séquentiel. Le combiner aussi : chaque map additionne déjà les 1 d’un même mot avant d’écrire.

faire_map. Lit le fichier d’entrée (sur le NFS), applique fonction_map, range chaque couple dans le seau de sa partition, et écrit $R$ fichiers dans /tmp : map-0007-part-03 contient la partition 3 du map 7.

ecrire_atomique. Une opération est atomique quand elle se fait en entier ou pas du tout, sans état intermédiaire visible. Écrire un fichier ne l’est pas : un lecteur peut tomber sur un fichier à moitié écrit. Renommer l’est : os.replace() remplace le nom en une seule opération du système. D’où le motif : écrire dans chemin.tmp, puis renommer en chemin. À tout instant, le fichier final est soit absent, soit complet. Si le travailleur meurt en pleine écriture, un réducteur ne lira jamais un fichier tronqué. C’est une technique standard, que l’on retrouve dans l’article de Google.

lire_partition. Sert une partition à un réducteur. Si le fichier n’existe pas, elle répond ERREUR, et surtout pas un dictionnaire vide : un dictionnaire vide donnerait un résultat faux (des mots manquants) sans aucun message. Une erreur visible vaut mieux qu’un résultat faux silencieux.

faire_reduce. C’est le shuffle distribué. La requête contient sources, la liste des couples (travailleur, id_map) : « le map 7 a été fait par tp-1a201-10.enst.fr:51114 ». Le réducteur r se connecte à chacun et demande sa partition r. Si un mappeur ne répond pas, il renvoie ERREUR avec le nom de la machine fautive : le maître saura ainsi qui est mort.

servir. Une connexion peut porter plusieurs requêtes à la suite. L’erreur d’une requête (fichier absent, clé manquante) produit une réponse ERREUR, sans tuer la connexion ni le travailleur.

Le programme principal. bind(("0.0.0.0", port)) : écouter sur toutes les interfaces réseau, sans quoi les autres machines ne pourraient pas se connecter (séance 1). Puis un thread par connexion (séance 2).

Un travailleur n’utilise qu’un cœur

Le travailleur sert ses requêtes avec des threads. À cause du GIL (séance 2), un seul thread exécute du Python à un instant donné : chaque travailleur n’utilise donc qu’un des 12 cœurs logiques de sa machine.

Le prix est mesurable. Sur les 32 fichiers de 1 Mo, 16 machines font le calcul en 2,02 s ; en séance 6, une seule machine avec 12 processus le faisait en 1,8 s. Pour bien faire, il faudrait plusieurs travailleurs par machine, ou un multiprocessing.Pool dans chaque travailleur. C’est ce que font les vrais systèmes : Hadoop exécute plusieurs tâches en même temps sur chaque machine.

7.6 Le maître

maitre.py

"""Maitre du MapReduce distribue (seance 7).

    python3 maitre.py
      travailleurs.txt : une ligne "machine:port" par travailleur
      fichiers.txt     : un chemin absolu par ligne (seance 6)
"""
import socket
import time
from concurrent.futures import ThreadPoolExecutor

from protocole import envoyer_message, recevoir_message

R = 8                       # nombre de reducteurs


def demander(travailleur, requete, delai=60.0):
    """Envoie une requete a "machine:port" et renvoie la reponse."""
    machine, port = travailleur.rsplit(":", 1)
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(delai)
    try:
        s.connect((machine, int(port)))
        envoyer_message(s, requete)
        return recevoir_message(s)
    finally:
        s.close()


def executer_lot(travailleurs, taches):
    """Lance une serie de maps ; renvoie (reussies, echecs).

    taches : liste de (id_map, fichier). La k-ieme tache va au
    travailleur numero k % len(travailleurs).
    """
    def executer(k_et_tache):
        k, (id_map, fichier) = k_et_tache
        travailleur = travailleurs[k % len(travailleurs)]
        try:
            r = demander(travailleur, {"type": "MAP", "id_map": id_map,
                                       "fichier": fichier, "R": R})
            if r.get("type") == "OK":
                return ("ok", travailleur, id_map)
        except (OSError, ConnectionError, ValueError):
            pass
        return ("echec", travailleur, id_map)

    reussies, echecs = [], []
    with ThreadPoolExecutor(max_workers=len(travailleurs)) as pool:
        for etat, travailleur, id_map in pool.map(executer,
                                                  enumerate(taches)):
            (reussies if etat == "ok" else echecs).append(
                (travailleur, id_map))
    return reussies, echecs


def phase_map(travailleurs, taches, essais_max=3):
    """Execute les maps, en relancant ailleurs ceux qui echouent.

    Renvoie (sources, vivants) : sources = [(travailleur, id_map)].
    """
    a_faire = list(taches)
    sources = []
    vivants = list(travailleurs)

    for essai in range(essais_max):
        if not a_faire or not vivants:
            break
        reussies, echecs = executer_lot(vivants, a_faire)
        sources.extend(reussies)

        morts = {t for t, _ in echecs}
        vivants = [t for t in vivants if t not in morts]
        perdus = {i for _, i in echecs}
        a_faire = [(i, f) for i, f in a_faire if i in perdus]
        if a_faire:
            print(f"  map, essai {essai + 1} : {len(a_faire)} tache(s) a "
                  f"relancer, {len(vivants)} travailleur(s) restant(s)")

    if a_faire:
        raise RuntimeError(f"{len(a_faire)} map(s) impossibles a terminer")
    return sources, vivants


def phase_reduce(travailleurs, sources):
    """Lance les R reduces ; renvoie l'ensemble des travailleurs morts."""
    def executer(r):
        travailleur = travailleurs[r % len(travailleurs)]
        try:
            reponse = demander(travailleur, {"type": "REDUCE", "r": r,
                                             "sources": sources},
                               delai=300.0)
        except (OSError, ConnectionError, ValueError):
            return travailleur                 # le reducteur est mort
        if reponse.get("type") == "OK":
            return None
        return reponse.get("machine")          # un mappeur est injoignable

    # Une requete par reducteur : ils travaillent tous en meme temps.
    with ThreadPoolExecutor(max_workers=R) as pool:
        return {mort for mort in pool.map(executer, range(R)) if mort}


def collecter_resultats(travailleurs):
    """Recupere les R parties ; renvoie (resultat, travailleurs morts)."""
    total, morts = {}, set()
    for r in range(R):
        travailleur = travailleurs[r % len(travailleurs)]  # comme au reduce
        try:
            reponse = demander(travailleur, {"type": "RESULTAT", "r": r})
        except (OSError, ConnectionError, ValueError):
            morts.add(travailleur)
            continue
        if reponse.get("type") != "OK":
            raise RuntimeError(f"partition {r} illisible sur {travailleur}")
        total.update(reponse["donnees"])   # chaque cle est dans un seul r
    return total, morts


def mapreduce_distribue(travailleurs, fichiers):
    taches = list(enumerate(fichiers))
    sources, vivants = phase_map(travailleurs, taches)

    # --- BARRIERE : tous les maps sont finis ---
    while True:
        if not vivants:
            raise RuntimeError("plus aucun travailleur")
        morts = phase_reduce(vivants, sources)
        if not morts:
            resultat, morts = collecter_resultats(vivants)
            if not morts:
                return resultat, vivants
        # Tout ce qui etait sur le disque local d'un mort est PERDU :
        # ses maps termines et ses resultats de reduce. On refait ses
        # maps ailleurs, puis on recommence tout le reduce.
        vivants = [t for t in vivants if t not in morts]
        perdus = [i for t, i in sources if t in morts]
        sources = [(t, i) for t, i in sources if t not in morts]
        print(f"  apres la barriere : {len(morts)} travailleur(s) mort(s), "
              f"{len(perdus)} map(s) a refaire")
        refaits, vivants = phase_map(vivants,
                                     [(i, fichiers[i]) for i in perdus])
        sources.extend(refaits)


if __name__ == "__main__":
    travailleurs = [l.strip() for l in open("travailleurs.txt") if l.strip()]
    fichiers = [l.strip() for l in open("fichiers.txt") if l.strip()]

    debut = time.perf_counter()
    resultat, vivants = mapreduce_distribue(travailleurs, fichiers)
    duree = time.perf_counter() - debut

    print(f"TOTAL : {duree:.2f} s")
    print(f"{len(resultat)} mots distincts, "
          f"{len(vivants)}/{len(travailleurs)} travailleurs utilises")

Lecture guidée

Le maître paraît long ; il est fait de six petites fonctions, chacune avec un seul rôle.

demander envoie une requête à un travailleur et rend la réponse. rsplit(":", 1) coupe l’adresse au dernier : : "tp-1a201-10.enst.fr:51114" donne la machine et le port. Le finally ferme la connexion dans tous les cas, même en cas d’erreur. Le delai est le timeout de la séance 2 : 60 secondes par défaut.

executer_lot envoie une série de maps, en parallèle grâce à un ThreadPoolExecutor (séance 2) : la $k$ -ième tâche va au travailleur $k \bmod n$ , où $n$ est le nombre de travailleurs. Avec 32 tâches et 4 travailleurs, chacun en reçoit 8. Elle renvoie deux listes : les maps réussis et les maps en échec, chacun sous forme (travailleur, id_map).

phase_map appelle executer_lot, puis recommence avec les seules tâches en échec, en écartant les travailleurs qui ont échoué. Au plus essais_max fois. Elle renvoie sources, la liste de qui a fait quel map, indispensable aux réducteurs.

phase_reduce lance les $R$ reduces en même temps et renvoie l’ensemble des travailleurs morts (vide si tout va bien). Un mort se révèle de deux façons : le réducteur lui-même ne répond pas, ou bien il répond ERREUR en désignant un mappeur injoignable.

collecter_resultats récupère les $R$ résultats. Le réducteur r est travailleurs[r % n], la même règle qu’au reduce : c’est ainsi que le maître sait où chercher, sans rien mémoriser. Un hachage envoie chaque mot dans une seule partition, donc update ne peut écraser aucune valeur.

mapreduce_distribue enchaîne le tout, et gère les pannes qui surviennent après la barrière ; nous y revenons plus bas.

Lancer le tout :

$ python3 maitre.py
TOTAL : 1.56 s
18810 mots distincts, 4/4 travailleurs utilises

7.7 Mesurer

Chronométrer chaque phase séparément

Un temps total ne vous apprend rien. Mesurez au minimum :

Phase	Ce qu’elle révèle si elle domine
Déploiement	le coût fixe du système, indépendant des données
Lecture des fichiers	vous êtes limité par les entrées/sorties
Calcul (map)	vous êtes limité par le processeur : le bon cas
Écriture des intermédiaires	le disque est lent — ou c’est le NFS
Shuffle et reduce	vous êtes limité par le réseau, ou $R$ est trop petit
Attente à la barrière	une machine lente bloque toutes les autres

import time
debut = time.perf_counter()
...
duree = time.perf_counter() - debut

Utilisez perf_counter() et non time() : c’est une horloge monotone (séance 5), insensible aux réglages de l’heure du système.

L’accélération et l’efficacité

$S(p) = \frac{T(1)}{T(p)} \qquad\qquad E(p) = \frac{S(p)}{p}$

$T(1)$ est le temps avec une seule machine, $T(p)$ avec $p$ machines. $S(p)$ est l’accélération (speedup) : $S = 7$ signifie 7 fois plus vite qu’avec une machine. $E(p)$ est l’efficacité : la part de chaque machine réellement utile. $E = 100\,\%$ est l’idéal, où $p$ machines vont $p$ fois plus vite.

Attention

La règle absolue de la mesure : tous les points d’une courbe d’accélération doivent porter sur exactement le même jeu de données. Si vous mesurez 10 fichiers avec 1 machine et 100 fichiers avec 10 machines, votre courbe ne veut rien dire. C’est l’erreur la plus fréquente, et elle invalide totalement le résultat.

La seule exception tolérable

On peut parfois extrapoler le point de référence $T(1)$ lorsque la mesure est trop longue : si 100 Mo prennent une minute sur une machine, on peut supposer, sous une hypothèse forte de linéarité, que 1 Go en prendrait dix. Cette hypothèse n’est acceptable que sur une seule machine. Dès qu’il y a deux machines ou plus, elle est fausse : il faut mesurer, jamais extrapoler.

Nos mesures

Mesures réelles, faites avec le script de l’exercice 7.4 : 32 fichiers de 1 Mo, $R = 8$ , 16 travailleurs déployés une fois, dont on utilise les $p$ premiers. Trois mesures par point, on garde la médiane (la valeur du milieu, insensible à une mesure aberrante). Le résultat a été vérifié identique à chaque fois.

$p$	$T(p)$ (s)	$S(p)$	$E(p)$
1	18,38	1,00	100 %
2	9,47	1,94	97 %
4	5,03	3,65	91 %
8	2,97	6,19	77 %
16	2,02	9,10	57 %

L’accélération continue de croître, mais l’efficacité s’effondre : à 16 machines, près de la moitié du temps machine est perdu. Trois raisons, que vous avez déjà rencontrées en séance 6 :

Les vagues. Un map dure environ 0,6 s. Avec 16 travailleurs, chacun fait $\lceil 32/16 \rceil = 2$ maps à la suite ; avec 32 machines ce serait 1, et au-delà, plus rien à gagner sur le map.
$R = 8$ . Il n’y a que 8 tâches reduce : au-delà de 8 machines, les autres n’ont rien à faire pendant cette phase.
Les coûts fixes : connexions, barrière, collecte séquentielle. Ils ne diminuent pas quand on ajoute des machines.

Et le déploiement ?

Déployer 16 travailleurs a pris 19 s, presque dix fois la durée du calcul (2,02 s). Si l’on redéployait à chaque calcul, 16 machines iraient moins vite qu’une seule ! C’est pourquoi les systèmes réels (Hadoop, Spark) gardent leurs travailleurs démarrés en permanence, et c’est pourquoi nous mesurons $T(p)$ travailleurs déjà lancés.

7.8 La loi d’Amdahl

Un problème pour commencer : processeur ou disque ?

Votre entreprise dispose d’un gros serveur de calcul (plusieurs processeurs, des disques de stockage). On veut le rendre plus rapide, et deux améliorations sont possibles :

Amélioration	Gain sur le composant	Coût
Changer les processeurs	50 % plus rapides	10 000 €
Changer les disques	150 % plus rapides	7 000 €

On a mesuré que les calculs passent 70 % de leur temps dans les processeurs, et 30 % à lire et écrire sur les disques.

Quel composant faut-il améliorer pour avoir la meilleure performance ?
Lequel donne le meilleur rapport performance / coût ?
Ce rapport est-il toujours le bon critère ?

L’intuition dit « le disque : 150 %, c’est trois fois mieux que 50 % ». Faisons le calcul.

Le raisonnement, en minutes

« 50 % plus rapide » veut dire que le composant fait en une seconde 1,5 fois plus de travail : le même travail lui prend $1/1\mathord{,}5$ du temps. « 150 % plus rapide » veut dire 2,5 fois plus de travail par seconde : le temps est divisé par 2,5. (Et non « le temps diminue de 150 % », ce qui n’aurait aucun sens.)

Prenons un calcul qui dure 60 minutes : 42 minutes de processeur (70 %), 18 minutes de disque (30 %).

	Processeur	Disque	Total	Accélération
Aujourd’hui	42 min	18 min	60 min	—
Nouveaux processeurs	$42 / 1\mathord{,}5 = 28$ min	18 min	46 min	$60 / 46 = 1\mathord{,}30$
Nouveaux disques	42 min	$18 / 2\mathord{,}5 = 7\mathord{,}2$ min	49,2 min	$60 / 49\mathord{,}2 = 1\mathord{,}22$

L’amélioration ne touche que le temps passé dans le composant amélioré ; le reste ne bouge pas. Le disque va 2,5 fois plus vite, mais il ne pesait que 18 minutes : on gagne moins qu’avec un processeur à peine plus rapide, qui pesait 42 minutes.

La formule générale

Appelons $a$ la part du temps concernée par l’amélioration (ici 0,7 ou 0,3), et $k$ le facteur d’accélération de cette partie (1,5 ou 2,5). Le temps après amélioration est :

$T_{\text{après}} = (1-a)\,T_{\text{avant}} + \frac{a\,T_{\text{avant}}}{k} \qquad\text{donc}\qquad S = \frac{T_{\text{avant}}}{T_{\text{après}}} = \frac{1}{(1-a) + \dfrac{a}{k}}$

C’est la loi d’Amdahl, du nom de Gene Amdahl, qui l’a énoncée en 1967. Sur notre problème :

Amélioration	$a$	$k$	$S$	Gain	Coût par % de gain
Processeurs	0,7	1,5	$\dfrac{1}{0\mathord{,}3 + 0\mathord{,}7/1\mathord{,}5} = 1\mathord{,}304$	+30,4 %	$10\,000 / 30\mathord{,}4 \approx$ 329 €
Disques	0,3	2,5	$\dfrac{1}{0\mathord{,}7 + 0\mathord{,}3/2\mathord{,}5} = 1\mathord{,}220$	+22,0 %	$7\,000 / 22\mathord{,}0 \approx$ 319 €

Avec des gains arrondis (30 % et 22 %), on trouve 333 € et 318 € : la conclusion est la même.

Le plafond. Même un composant infiniment rapide ( $k \to \infty$ ) ne supprime que sa propre part du temps : $S$ plafonne à $1/(1-a)$ . Un disque infiniment rapide ne donnerait que $1/0\mathord{,}7 = 1\mathord{,}43$ , soit +43 % ; des processeurs infiniment rapides, $1/0\mathord{,}3 = 3\mathord{,}33$ .

Les réponses

1. La meilleure performance : les processeurs (+30 % contre +22 %). Ce n’est pas le composant le plus amélioré qui compte, c’est le composant le plus utilisé.

2. Le meilleur rapport performance / coût : les disques, de peu (319 € par point de pourcentage, contre 329 €).

3. Ce rapport n’est pas toujours le bon critère.

Un objectif à tenir. Si le calcul de 60 minutes doit passer sous les 45 minutes, aucune des deux options ne suffit seule (46 et 49,2 min) : il faut les deux, soit $42/1\mathord{,}5 + 18/2\mathord{,}5 = 35\mathord{,}2$ minutes, pour 17 000 €.
Un budget. Avec 8 000 €, la question ne se pose pas : seuls les disques sont possibles.
Les gains ne s’additionnent pas. Les deux ensemble donnent $60 / 35\mathord{,}2 = 1\mathord{,}70$ , soit +70 %, et non $30 + 22 = 52$ % : un « coût par % » calculé option par option ne permet pas de comparer des combinaisons.
Le profil change avec le travail. Les 70 % / 30 % valent pour ces calculs. Un autre programme, qui lit davantage, inverserait la conclusion. D’où la règle : mesurer le profil avant d’acheter, comme à l’exercice 7.3.
D’autres critères comptent aussi : consommation électrique, fiabilité, délai d’installation.

À retenir

Pour accélérer un système, on améliore ce qui pèse le plus dans le temps, pas ce qui s’améliore le plus. Et on le sait en mesurant.

Appliquée à notre système

À l’exercice 7.3, nous avons mesuré où passe le temps des maps : lecture 15 %, calcul 79 %, écriture 6 %. Même une écriture infiniment rapide ne donnerait que $1/(1 - 0\mathord{,}06) = 1\mathord{,}064$ , soit +6 %. C’est pourquoi, à l’exercice 7.2, passer de /tmp au NFS change peu le total quand le serveur NFS est peu chargé, alors que chaque écriture y est plusieurs fois plus lente. Le calcul, lui, pèse 79 % : c’est là qu’il faut agir.

Le parallélisme : un cas particulier

Répartir un calcul sur $p$ machines, c’est accélérer d’un facteur $k = p$ la partie qui se partage. Tout programme contient pourtant une part que l’on ne peut pas partager : lire la liste des fichiers, attendre la barrière, collecter le résultat. Appelons $f$ cette fraction séquentielle du temps passé sur une machine. La partie améliorée est $a = 1 - f$ , et la formule générale devient :

$S(p) = \frac{1}{f + \dfrac{1-f}{p}} \qquad\text{et}\qquad S(\infty) = \frac{1}{f}$

Quand $p$ devient très grand, $(1-f)/p$ tend vers 0 : il ne reste que la part séquentielle, et l’accélération plafonne à $1/f$ . C’est le plafond de l’exemple : un disque infiniment rapide ne supprimait que sa part.

Ce que cela veut dire concrètement

$f$	$S(10)$	$S(100)$	$S(1000)$	plafond
1 %	9,2	50,3	91,0	100
5 %	6,9	16,8	19,6	20
10 %	5,3	9,2	9,9	10
30 %	2,7	3,3	3,3	3,3

À retenir

Avec seulement 5 % de code séquentiel, mille machines ne vous apporteront jamais plus qu’un facteur 20. Ajouter des machines ne sert à rien passé un certain point : il faut réduire $f$ . C’est la raison pour laquelle l’optimisation la plus rentable est presque toujours de supprimer une synchronisation, pas d’acheter du matériel.

Estimer $f$ à partir de vos mesures

En isolant $f$ dans la formule d’Amdahl, une seule mesure $S(p)$ suffit :

$f = \frac{\dfrac{1}{S(p)} - \dfrac{1}{p}}{1 - \dfrac{1}{p}}$

Sur nos mesures, avec $S(16) = 9\mathord{,}10$ :

$f = \frac{1/9\mathord{,}10 - 1/16}{1 - 1/16} = \frac{0\mathord{,}1099 - 0\mathord{,}0625}{0\mathord{,}9375} \approx 0\mathord{,}051$

soit 5 % de séquentiel, et un plafond d’environ $1/0\mathord{,}051 \approx 20$ , quel que soit le nombre de machines ajoutées.

f

n’est pas une constante

Le même calcul sur les autres points donne $f \approx 0\mathord{,}031$ ( $p = 2$ ), $0\mathord{,}032$ ( $p = 4$ ), $0\mathord{,}042$ ( $p = 8$ ) et $0\mathord{,}051$ ( $p = 16$ ). Si la loi d’Amdahl décrivait parfaitement notre système, on trouverait la même valeur partout. Elle augmente avec $p$ , parce que les vagues et la limite $R = 8$ pèsent de plus en plus : ce sont des pertes qui grandissent avec le nombre de machines, et que le modèle ne prévoit pas. Amdahl est un modèle : il donne le bon ordre de grandeur, pas un chiffre exact.

Pourquoi la courbe peut redescendre

Sur un vrai système, l’accélération finit souvent par diminuer au-delà d’un certain nombre de machines. Amdahl prédit un plateau, pas une baisse. Les causes de la baisse sont ailleurs :

le coût du déploiement croît avec $N$ ;
le shuffle fait $M \times R$ demandes, dont la plupart traversent le réseau ;
la barrière : on attend toujours la machine la plus lente (straggler) ;
une ressource partagée sature (le réseau, le serveur NFS).

Le straggler

L’article de Google appelle straggler (le traînard) une machine anormalement lente sur l’une des dernières tâches : un disque défaillant, une machine chargée par d’autres programmes. Leur parade : quand le calcul approche de sa fin, le maître relance les dernières tâches en cours en double sur d’autres machines, et garde le premier résultat qui arrive. Ce sont les backup tasks, ou exécution spéculative. Sur leur programme de tri (section 5.4 de l’article), désactiver ce mécanisme allonge la durée totale de 44 % : 1 283 s au lieu de 891 s, à cause de 5 tâches reduce qui finissent 300 s après toutes les autres.

7.9 Survivre à une panne

Avec cent machines, il y en aura toujours une pour tomber, être redémarrée par un étudiant, ou saturer. Le système doit continuer.

Deux façons de mourir

Une panne n’est pas toujours visible de la même façon. Nos mesures, sur 4 machines et 32 fichiers (exercice 7.5) :

Ce qui arrive	Comment le maître l’apprend	Délai mesuré
Le processus du travailleur est tué, la machine vit	le système de la machine refuse ou coupe aussitôt la connexion	immédiat : calcul fini en 7,4 s au lieu de 6,1 s
La machine est figée, débranchée, ou le réseau est coupé	plus rien ne répond ; seul le timeout le révèle	60 s (timeout du map) + 20 s (timeout du shuffle) : 88,7 s

C’est le dilemme de la tolérance, vu en séance 5 : de l’extérieur, un silence ne dit pas si la machine est morte ou seulement lente. Le timeout est le seul outil, et il coûte cher : trop court, on déclare morte une machine simplement lente ; trop long, on attend longtemps une machine vraiment morte.

Réagir : relancer la tâche ailleurs

Le principe de MapReduce est d’une grande simplicité : on refait la tâche ailleurs. C’est possible parce que les tâches sont idempotentes : les refaire une, deux ou dix fois donne exactement le même résultat. Un map lit un fichier qui ne change pas et applique une fonction sans effet de bord ; un reduce réécrit son résultat par ecrire_atomique. Rien n’est compté deux fois.

Situation	Que faire	Où, dans notre code
Un map en cours échoue	le relancer sur une autre machine	`phase_map`, essais suivants
Un map terminé, machine morte	le refaire quand même : ses fichiers étaient sur le disque local de la machine morte	`mapreduce_distribue`, liste `perdus`
Un reduce échoue	refaire les maps perdus, puis tous les reduces	`mapreduce_distribue`, boucle
Un reduce terminé, machine morte avant la collecte	le refaire : chez nous, son résultat aussi est en local	`collecter_resultats` signale le mort
Le maître meurt	tout est perdu : c’est le point unique de défaillance	—

Attention

La deuxième ligne surprend toujours. Un map terminé doit être refait si sa machine meurt, parce que son résultat était écrit en local. Mesuré (exercice 7.5) : une machine tuée juste après la barrière, et ses 8 maps, pourtant terminés, sont refaits.

La quatrième ligne est différente chez Google. Leurs reduces écrivent leur résultat dans un système de fichiers global et répliqué (GFS) : un reduce terminé n’a jamais besoin d’être refait. Les nôtres l’écrivent dans /tmp ; tant que le maître ne l’a pas collecté, il est perdu avec sa machine.

Dans notre code. phase_map gère les pannes pendant les maps : les tâches en échec sont relancées sur les survivants. La boucle while True de mapreduce_distribue gère les pannes après la barrière : si un travailleur meurt pendant le reduce ou avant la collecte, on le retire, on refait ailleurs les maps qu’il avait terminés (liste perdus), puis on relance le reduce. Le résultat final est identique au calcul sans panne : c’est ce que l’exercice 7.5 vous fait vérifier.

7.10 Après MapReduce

Vous venez d’écrire, à la main, ce que Hadoop automatise. Deux évolutions majeures ont suivi, et il est utile de savoir les situer.

Les moteurs de flot de données : Spark

La faiblesse structurelle de MapReduce est visible dans votre propre code : tout passe par le disque. Chaque map écrit ses partitions, chaque reduce les relit. Pour un traitement en une passe, c’est acceptable. Pour un algorithme itératif — un apprentissage automatique, un parcours de graphe, qui répètent cent fois le même calcul — on paie cent fois ce coût.

Les moteurs comme Spark conservent les données intermédiaires en mémoire entre les étapes, et ne matérialisent sur disque que ce qui est nécessaire. Ils raisonnent sur le graphe complet des opérations avant de l’exécuter, ce qui leur permet d’optimiser (fusionner deux étapes, éviter un shuffle inutile). Le modèle de programmation, lui, reste très proche : des transformations sur des collections partitionnées.

Du lot au flux : le traitement en continu

MapReduce traite un lot (batch) : une entrée finie, connue d’avance, un début et une fin. Beaucoup de besoins réels ne fonctionnent pas ainsi : les événements arrivent en permanence et il n’y a pas de « fin ».

	Traitement par lots	Traitement en continu
Entrée	finie, bornée	infinie, non bornée
Résultat	produit une fois, à la fin	mis à jour en permanence
Latence	minutes à heures	millisecondes à secondes
Reprise	relancer la tâche	rejouer depuis un point de reprise
Outils	MapReduce, Spark	Kafka Streams, Flink

Le changement de mentalité est réel : un comptage de mots en continu ne produit jamais un résultat définitif, mais une suite de mises à jour (chat vaut 3, puis 4, puis 7…). Il faut alors décider quand on considère un résultat comme publiable — ce qui ramène directement à la question des horloges vue en séance 5.

Pour aller plus loin

Kleppmann & Riccomini, Designing Data-Intensive Applications, chapitres 11 (Batch Processing) et 12 (Stream Processing).

7.11 Exercices

Les corrigés se trouvent à la Section 7.12.

Avant de commencer : les règles du terrain partagé

Tout se fait depuis une session sur une machine de l’école, dans ~/bgd701.
Pas de boucle qui lance des ssh en parallèle : deployer.py les fait un par un.
Nettoyez (python3 deployer.py nettoyer) avant de partir, et à chaque fois qu’un exercice le demande.
Gardez les expériences petites : les tailles indiquées suffisent.

Exercice 7.1 — Faire tourner le système

L’objectif est d’obtenir le même résultat qu’en séance 6, mais calculé par quatre machines.

1. Rassembler les fichiers. Dans ~/bgd701, il vous faut :

de la séance 3 : protocole.py ;
de la séance 6 : mapreduce_simple.py, fabriquer_corpus.py et le dossier corpus_source/ ;
de la séance 2 : vivantes.txt, produit par machines_libres.py. Si votre fichier date d’une autre séance, refaites-le : les machines allumées changent ;
de ce chapitre : travailleur.py, deployer.py, maitre.py.

Vérifiez avec ls.

2. Fabriquer les données : 8 fichiers de 1 Mo, et fichiers.txt.

$ python3 fabriquer_corpus.py 8 1

3. Trouver votre port. Lancez id -u, puis calculez $40000 + \text{uid} \bmod 20000$ . Vérifiez avec python3 -c "import deployer; print(deployer.PORT)".

4. Déployer 4 travailleurs, puis regarder travailleurs.txt :

$ python3 deployer.py deployer 4
$ cat travailleurs.txt

5. Calculer : python3 maitre.py.

6. Vérifier. Écrivez verifier.py, qui calcule le résultat distribué et le résultat du moteur séquentiel de la séance 6 sur les mêmes fichiers, compare les deux dictionnaires et affiche les deux durées.

7. Ranger : python3 deployer.py nettoyer, puis vérifiez que plus aucun travailleur ne répond, sans utiliser pgrep (pourquoi ?).

Questions.

Le résultat distribué est-il identique au résultat séquentiel ? Pourquoi cette comparaison est-elle le seul moyen sérieux de savoir que votre système est correct ?
Le calcul distribué sur 4 machines est-il 4 fois plus rapide que le séquentiel ? Proposez une explication.

Exercice 7.2 — NFS contre disque local

A. L’écriture seule. Le script suivant écrit 64 petits fichiers JSON dans /tmp, puis dans votre répertoire personnel, et les efface aussitôt.

ecrire_64.py

"""Ecrit 64 petits fichiers JSON dans /tmp, puis dans ~ (exercice 7.2)."""
import getpass
import json
import os
import shutil
import time

donnees = {f"mot{i}": i for i in range(3000)}    # ~ 50 Ko, comme une partition
dossiers = [("/tmp  (disque local)", f"/tmp/bgd701-essai-{getpass.getuser()}"),
            ("~     (NFS)", os.path.expanduser("~/bgd701-essai"))]

for nom, dossier in dossiers:
    os.makedirs(dossier, exist_ok=True)
    debut = time.perf_counter()
    for k in range(64):
        with open(f"{dossier}/f{k}.tmp", "w", encoding="utf-8") as f:
            json.dump(donnees, f)
        os.replace(f"{dossier}/f{k}.tmp", f"{dossier}/f{k}")
    duree = time.perf_counter() - debut
    taille = sum(os.path.getsize(f"{dossier}/f{k}") for k in range(64))
    shutil.rmtree(dossier)                        # on nettoie aussitot
    print(f"{nom} : 64 fichiers, {taille / 1e6:.1f} Mo en {duree:.2f} s")

Lancez-le deux fois. Notez les quatre durées.

B. Le système complet. Avec 2 machines et 4 fichiers de 2 Mo (python3 fabriquer_corpus.py 4 2) :

déployez 2 travailleurs, lancez maitre.py, notez le TOTAL, nettoyez ;
faites une copie de sauvegarde, cp travailleur.py travailleur_tmp.py, puis remplacez dans travailleur.py la ligne DOSSIER = ... par DOSSIER = os.path.expanduser("~/bgd701-tmp") ;
redéployez 2 travailleurs, relancez maitre.py, notez le TOTAL, nettoyez ;
restaurez : cp travailleur_tmp.py travailleur.py ;
effacez ~/bgd701-tmp (nettoyer ne s’en occupe pas : pourquoi ?), puis vérifiez une minute plus tard, avec ls -d ~/bgd701-tmp, qu’il a bien disparu.

Questions.

Quel rapport de durées trouvez-vous en A ? Et en B ? Pourquoi l’écart est-il plus faible, ou plus variable, en B ?
Les deux travailleurs de la version NFS écrivent dans le même dossier. Pourquoi n’y a-t-il pas de conflit ?
Expliquez en trois phrases pourquoi, à cent machines, la version NFS serait une catastrophe, même si l’écart vous paraît modeste ici.

Gardez l’expérience petite

Ne soyez pas tenté de la refaire avec 8 machines et 16 fichiers de 5 Mo : vous écririez alors des centaines de mégaoctets sur le serveur NFS, et vous le ralentiriez pour toute l’école, vous compris. On nettoie ~/bgd701-tmp juste après.

Exercice 7.3 — Décomposer le temps

On veut savoir où passe le temps dans un map : lecture du fichier, calcul, ou écriture des partitions.

1. Instrumenter le travailleur. Dans faire_map, relevez time.perf_counter() avant la lecture (t0), après la lecture (t1), après le calcul des seaux (t2) et après l’écriture (t3). Ajoutez à la réponse OK les trois durées t_lecture, t_calcul, t_ecriture. N’oubliez pas import time.

2. Récupérer les durées côté maître, sans modifier maitre.py. Les réponses passent toutes par maitre.demander. Écrivez profil.py, qui remplace cette fonction par un espion : une fonction qui appelle la vraie demander, garde au passage les réponses aux requêtes MAP, et les renvoie sans rien changer.

import maitre

reponses = []
demander = maitre.demander          # la vraie fonction, mise de cote

def espion(travailleur, requete, delai=60.0):
    reponse = demander(travailleur, requete, delai)
    if requete["type"] == "MAP":
        reponses.append(reponse)
    return reponse

maitre.demander = espion            # le maitre utilisera l'espion

Ensuite, profil.py appelle lui-même phase_map, phase_reduce et collecter_resultats en chronométrant chacune, puis affiche :

la somme de chaque durée sur tous les maps, et son pourcentage ;
la durée de chaque phase vue du maître.

3. Mesurer avec 32 fichiers de 1 Mo (fabriquer_corpus.py 32 1) et 4 travailleurs, déployés avec le travailleur instrumenté.

Questions. Où est le goulot d’étranglement ? Justifiez par les chiffres, pas par l’intuition. Retrouvez la durée de la phase map, vue du maître, à partir de la durée moyenne d’un map.

Exercice 7.4 — La courbe d’accélération

Avec un jeu de données fixe — python3 fabriquer_corpus.py 32 1, soit 32 fichiers de 1 Mo — mesurez le temps total pour $p = 1, 2, 4, 8, 16$ machines. Faites trois mesures par point et gardez la médiane.

Méthode. Ne redéployez pas pour chaque $p$ : déployez une fois 16 travailleurs, puis, pour chaque $p$ , passez au maître les $p$ premières lignes de travailleurs.txt (travailleurs[:p]). Écrivez pour cela acceleration.py, qui importe mapreduce_distribue depuis maitre.py. Chronométrez aussi le déploiement.

Tabulez $T(p)$ , $S(p) = T(1)/T(p)$ et $E(p) = S(p)/p$ .
Estimez $f$ avec la formule du cours, pour chaque $p$ . Trouvez-vous la même valeur ? Pourquoi ?
Quel est votre plafond théorique d’accélération ?
Pourquoi l’efficacité baisse-t-elle ? Appuyez-vous sur la notion de vagues (séance 6), sur la valeur de $R$ , et sur l’exercice 7.3.
Comparez le temps du déploiement à $T(16)$ . Qu’en concluez-vous ?

Si toute la salle mesure en même temps

Vos travailleurs peuvent partager une machine avec ceux d’un camarade : les mesures deviennent bruitées. C’est normal ; la médiane de trois mesures limite les dégâts, et notez-le dans votre analyse.

Exercice 7.5 — Tuer une machine

On va provoquer des pannes au bon moment, puis vérifier que le résultat ne change pas. Le script suivant s’en charge : il tue (ou fige) le travailleur de la deuxième ligne de travailleurs.txt, à l’instant choisi, puis compare le résultat au moteur séquentiel.

panne.py

"""Provoque une panne pendant un calcul, puis verifie le resultat (ex. 7.5).

    python3 panne.py sans-relance     tue un travailleur a t = 2 s, sans relance
    python3 panne.py pendant-map      tue un travailleur a t = 2 s
    python3 panne.py apres-map        le tue juste apres la barriere
    python3 panne.py avant-collecte   le tue juste avant la collecte
    python3 panne.py figer            le fige (SIGSTOP) a t = 2 s
"""
import subprocess
import sys
import threading
import time

import maitre
from mapreduce_simple import mapreduce, fonction_map, fonction_reduce

moment = sys.argv[1]
travailleurs = [l.strip() for l in open("travailleurs.txt") if l.strip()]
fichiers = [l.strip() for l in open("fichiers.txt") if l.strip()]
victime = travailleurs[1].rsplit(":", 1)[0]       # la deuxieme machine
debut = time.perf_counter()


def frapper():
    signal = "-STOP" if moment == "figer" else "-TERM"
    subprocess.run(["ssh", "-o", "BatchMode=yes", victime,
                    f"pkill {signal} -u $USER -f '[t]ravailleur.py'"])
    print(f"   >>> t = {time.perf_counter() - debut:5.1f} s : travailleur "
          f"{'FIGE' if moment == 'figer' else 'TUE'} sur {victime}")


def precede_d_une_panne(fonction):
    """Renvoie la meme fonction, mais qui frappe lors de son 1er appel."""
    deja = []

    def espion(*args):
        if not deja:
            deja.append(True)
            frapper()
        return fonction(*args)
    return espion


if moment == "sans-relance":
    phase_map = maitre.phase_map
    maitre.phase_map = lambda t, taches: phase_map(t, taches, essais_max=1)
if moment in ("sans-relance", "pendant-map", "figer"):
    threading.Timer(2.0, frapper).start()
elif moment == "apres-map":
    maitre.phase_reduce = precede_d_une_panne(maitre.phase_reduce)
elif moment == "avant-collecte":
    maitre.collecter_resultats = precede_d_une_panne(
        maitre.collecter_resultats)

distribue, vivants = maitre.mapreduce_distribue(travailleurs, fichiers)
print(f"t = {time.perf_counter() - debut:5.1f} s : termine avec "
      f"{len(vivants)} travailleur(s) sur {len(travailleurs)}")
reference = mapreduce(fichiers, fonction_map, fonction_reduce)
print("identique au moteur sequentiel :", distribue == reference)

Deux techniques à reconnaître : threading.Timer(2.0, frapper) exécute frapper dans 2 secondes, pendant que le calcul tourne ; et precede_d_une_panne est un espion, comme dans l’exercice 7.3, qui frappe juste avant le premier appel de la fonction espionnée.

Préparation : 32 fichiers de 1 Mo, pour que la phase map dure environ 5 s. Avant chaque question, repartez de zéro avec 4 travailleurs neufs :

$ python3 deployer.py nettoyer ; python3 deployer.py deployer 4

python3 panne.py sans-relance : sans mécanisme de reprise, que se passe-t-il ? Le maître attend-il 60 s, la durée de son timeout ? Pourquoi ?
python3 panne.py pendant-map : le calcul aboutit-il ? Le résultat est-il identique ? Expliquez les deux lignes affichées par le maître.
python3 panne.py apres-map : combien de maps sont refaits ? Pourquoi ceux-là, alors qu’ils étaient terminés ?
python3 panne.py avant-collecte : que doit faire le maître ? Que se passerait-il chez Google, dont les reduces écrivent dans un système de fichiers global ?
python3 panne.py figer : comparez la durée totale à celle de la question 2. D’où viennent les secondes en plus ? Puis dégelez la victime avant de nettoyer : ssh MACHINE "pkill -CONT -u \$USER -f '[t]ravailleur.py'".

Exercice 7.6 — Un autre traitement

Sans modifier le maître ni le protocole, faites calculer à votre système les trois traitements de l’exercice 6.2 : la répartition des longueurs de mots, l’index inversé — pour chaque mot, la liste des fichiers où il apparaît —, et la longueur moyenne par initiale.

L’idéal : réutiliser telles quelles les fonctions de fonctions_62.py.

Repérez d’abord toutes les lignes du travailleur qui supposent que l’on compte des mots. Il y en a plus qu’on ne croit : regardez de près le combiner de faire_map, et l’agrégation de faire_reduce.
Rendez le travailleur générique : il importe fonction_map et fonction_reduce depuis un fichier traitement.py, et appelle fonction_reduce(cle, valeurs) comme le moteur de la séance 6. Pour changer de traitement, il suffit alors de modifier traitement.py et de redéployer.
Vérifiez chaque traitement contre le moteur séquentiel. L’un des trois « échoue » : lequel, et pourquoi ? Est-ce une vraie erreur ?
Mesurez la taille totale des fichiers intermédiaires (map-*), sur le comptage de mots, avec le travailleur du cours puis avec le travailleur générique. Qu’avez-vous perdu ? Pourquoi ne peut-on pas simplement le remettre pour la moyenne ?

Exercice 7.7 — Quel composant améliorer ?

C’est le problème « processeur ou disque » du cours, sur votre système. À l’exercice 7.3, le temps des maps se répartissait ainsi : lecture 15 %, calcul 79 %, écriture 6 %. Supposons que tout le calcul ait ce profil. Deux achats sont possibles :

Option	Effet	Coût
A. Nouveaux processeurs	calcul 100 % plus rapide	12 000 €
B. Disques SSD	lectures et écritures 400 % plus rapides	3 000 €

Traduisez « 100 % plus rapide » et « 400 % plus rapide » en facteurs $k$ . Quelle est la part $a$ du temps concernée par chaque option ?
Calculez l’accélération de chaque option. Laquelle donne la meilleure performance ? Le meilleur rapport performance / coût ?
Sur une machine, le calcul dure 18,38 s (exercice 7.4). Il doit passer sous 12 s. Que faut-il acheter ? Faites le calcul en secondes, comme le tableau en minutes du cours.
Quel est le plafond de l’option B, avec des disques infiniment rapides ?

7.12 Corrigés

Exercice 7.1 — Faire tourner le système

verifier.py

"""Compare le resultat distribue au moteur sequentiel (exercice 7.1)."""
import time

from maitre import mapreduce_distribue
from mapreduce_simple import mapreduce, fonction_map, fonction_reduce

travailleurs = [l.strip() for l in open("travailleurs.txt") if l.strip()]
fichiers = [l.strip() for l in open("fichiers.txt") if l.strip()]

debut = time.perf_counter()
distribue, vivants = mapreduce_distribue(travailleurs, fichiers)
print(f"distribue  : {time.perf_counter() - debut:5.2f} s "
      f"({len(vivants)} travailleurs)")

debut = time.perf_counter()
reference = mapreduce(fichiers, fonction_map, fonction_reduce)
print(f"sequentiel : {time.perf_counter() - debut:5.2f} s (1 processus)")

assert distribue == reference, "resultats differents !"
print(f"OK : {len(reference)} mots distincts, resultats identiques")

Sortie réelle, sur tp-1a201-18, avec 4 travailleurs :

$ python3 verifier.py
distribue  :  1.37 s (4 travailleurs)
sequentiel :  2.07 s (1 processus)
OK : 18810 mots distincts, resultats identiques

$ python3 deployer.py nettoyer
  ok     tp-1a201-10.enst.fr
  ok     tp-1a201-27.enst.fr
  ok     tp-1a201-31.enst.fr
  ok     tp-1a201-13.enst.fr

Pour vérifier l’arrêt, on tente une connexion au port de chaque machine ; si aucune ne réussit, plus aucun travailleur n’écoute :

import socket
for ligne in open("travailleurs.txt"):
    machine, port = ligne.strip().rsplit(":", 1)
    try:
        socket.create_connection((machine, int(port)), timeout=2).close()
        print("ENCORE VIVANT :", machine)
    except OSError:
        print("arrete :", machine)

pgrep -f travailleur.py, lancé par ssh, trouverait son propre shell, dont la ligne de commande contient le motif : c’est le piège du nettoyage, à l’envers.

a) Oui, à l’identique : 18 810 mots, mêmes valeurs. Sans ce test, vous savez seulement que votre système ne plante pas ; vous ne savez pas qu’il est juste. Un mot oublié dans une partition, une partition lue deux fois, et le résultat serait faux sans le moindre message.

b) Non : 1,37 s contre 2,07 s, soit 1,5 fois plus vite seulement avec 4 machines. Le travail est minuscule (8 Mo) : chaque travailleur fait 2 maps d’environ 0,6 s, puis viennent le shuffle ( $8 \times 8 = 64$ demandes, dont 48 par le réseau), la collecte et les connexions, qui ne se divisent pas. Plus le travail est petit, plus ces coûts fixes pèsent. L’exercice 7.4 le quantifie.

Exercice 7.2 — NFS contre disque local

A. Deux lancements successifs de ecrire_64.py :

/tmp  (disque local) : 64 fichiers, 3.1 Mo en 0.30 s
~     (NFS) : 64 fichiers, 3.1 Mo en 1.06 s
/tmp  (disque local) : 64 fichiers, 3.1 Mo en 0.30 s
~     (NFS) : 64 fichiers, 3.1 Mo en 2.60 s

B. Le système complet, 2 machines, 4 fichiers de 2 Mo :

`DOSSIER`	`TOTAL`, campagne 1	`TOTAL`, campagne 2
`/tmp` (disque local)	2,38 s	3,25 s
`~/bgd701-tmp` (NFS)	2,44 s	9,91 s

a) En A, le NFS est 3,5 à 9 fois plus lent, et sa durée varie d’un essai à l’autre, alors que /tmp ne bouge pas. En B, l’écart va de presque rien (2,44 contre 2,38 s) à un facteur 3 (9,91 contre 3,25 s), selon le moment.

Deux effets se combinent. D’abord, dans le système complet, l’écriture n’est qu’une petite partie du travail : le calcul domine (exercice 7.3). Rendre l’écriture plusieurs fois plus lente ne change donc presque rien tant que le serveur NFS est peu chargé. Ensuite, le serveur est partagé : quand d’autres l’utilisent au même moment, chaque écriture attend son tour, et l’écriture devient dominante. Votre programme dépend alors de ce que font les autres. C’est déjà une raison suffisante pour ne pas l’utiliser.

Le piège du nettoyage. Lors de nos essais, rm -rf ~/bgd701-tmp, lancé juste après le calcul, n’a rien effacé, sans aucun message : 40 fichiers étaient encore là. Le dossier avait été créé par deux autres machines ; celle qui lançait rm avait encore en mémoire, dans son cache NFS, l’information « ce dossier n’existe pas ». Relancé un peu plus tard, le même rm a fonctionné. Avec le NFS, ce que voit une machine peut être en retard sur ce qu’a fait une autre : d’où la vérification demandée une minute après.

nettoyer n’efface que /tmp/bgd701-login sur chaque machine : il ne sait rien de ~/bgd701-tmp, qui n’existe que dans cet exercice.

b) Les noms de fichiers sont uniques : map-0007-part-03 n’est écrit que par le travailleur qui a fait le map 7, et resultat-part-03 par le seul réducteur 3. Deux écrivains ne touchent jamais le même fichier. Le renommage atomique protège en plus les lecteurs.

c) Avec /tmp, chaque machine ajoutée apporte son propre disque : la capacité d’écriture grandit avec le nombre de machines. Avec le NFS, toutes écrivent sur un seul serveur, dont la capacité ne bouge pas : à cent machines, chacune n’en reçoit plus qu’un centième, et l’écriture devient le goulot de tout le calcul. Et ce serveur est celui de toute l’école : vous ralentissez aussi tous les autres.

Exercice 7.3 — Décomposer le temps

Le travailleur instrumenté : seule faire_map change, plus import time en tête de fichier.

def faire_map(requete):
    id_map, R = requete["id_map"], requete["R"]
    t0 = time.perf_counter()
    with open(requete["fichier"], encoding="utf-8", errors="ignore") as f:
        contenu = f.read()
    t1 = time.perf_counter()

    seaux = [dict() for _ in range(R)]          # combiner : un dict par r
    for cle, valeur in fonction_map(requete["fichier"], contenu):
        r = partition(cle, R)
        seaux[r][cle] = seaux[r].get(cle, 0) + valeur
    t2 = time.perf_counter()

    for r in range(R):
        ecrire_atomique(f"{DOSSIER}/map-{id_map:04d}-part-{r:02d}",
                        seaux[r])
    t3 = time.perf_counter()
    return {"type": "OK", "id_map": id_map, "t_lecture": t1 - t0,
            "t_calcul": t2 - t1, "t_ecriture": t3 - t2}

Le profil, côté maître :

profil.py

"""Ou passe le temps des maps ? (exercice 7.3)

A lancer avec le travailleur instrumente, qui renvoie t_lecture,
t_calcul et t_ecriture dans sa reponse OK.
"""
import time

import maitre

travailleurs = [l.strip() for l in open("travailleurs.txt") if l.strip()]
fichiers = [l.strip() for l in open("fichiers.txt") if l.strip()]

# Un espion : la meme fonction demander, qui garde en plus les reponses MAP.
reponses = []
demander = maitre.demander


def espion(travailleur, requete, delai=60.0):
    reponse = demander(travailleur, requete, delai)
    if requete["type"] == "MAP":
        reponses.append(reponse)
    return reponse


maitre.demander = espion

debut = time.perf_counter()
sources, vivants = maitre.phase_map(travailleurs, list(enumerate(fichiers)))
t_map = time.perf_counter() - debut
debut = time.perf_counter()
maitre.phase_reduce(vivants, sources)
t_reduce = time.perf_counter() - debut
debut = time.perf_counter()
maitre.collecter_resultats(vivants)
t_collecte = time.perf_counter() - debut

print(f"Dans les {len(reponses)} maps (temps additionnes) :")
phases = ("t_lecture", "t_calcul", "t_ecriture")
totaux = {p: sum(r[p] for r in reponses) for p in phases}
cumul = sum(totaux.values())
for p in phases:
    print(f"  {p:<11} {totaux[p]:6.2f} s  {totaux[p] / cumul:4.0%}")
print(f"Vu du maitre ({len(travailleurs)} travailleurs) :")
print(f"  map {t_map:5.2f} s   reduce {t_reduce:5.2f} s   "
      f"collecte {t_collecte:5.2f} s   total "
      f"{t_map + t_reduce + t_collecte:5.2f} s")

L’espion fonctionne parce que executer_lot cherche demander dans le module maitre au moment de l’appel : en remplaçant maitre.demander, on change ce qu’elle trouve, sans toucher au fichier.

Sortie réelle, 32 fichiers de 1 Mo, 4 travailleurs :

Dans les 32 maps (temps additionnes) :
  t_lecture     2.85 s   15%
  t_calcul     15.56 s   79%
  t_ecriture    1.20 s    6%
Vu du maitre (4 travailleurs) :
  map  5.11 s   reduce  0.90 s   collecte  0.11 s   total  6.12 s

Le goulot est le calcul : 79 % du temps des maps. C’est le bon régime : on est limité par le processeur, et ajouter des machines aide. L’écriture dans /tmp ne pèse que 6 %.

Retrouver la phase map. Un map dure en moyenne $(2\mathord{,}85 + 15\mathord{,}56 + 1\mathord{,}20) / 32 \approx 0\mathord{,}61$ s. Chaque travailleur en fait $32 / 4 = 8$ à la suite, soit 8 vagues : $8 \times 0\mathord{,}61 \approx 4\mathord{,}9$ s, pour 5,11 s mesurées. Le reste est le temps des connexions et de l’attente du map le plus lent.

La même mesure avec DOSSIER sur le NFS (faite une fois par l’enseignant, pour ne pas multiplier les écritures sur le serveur) donnait une écriture de 8,86 s, soit 36 % du temps des maps, contre 6 % ici, et un total de 8,24 s au lieu de 6,12 s. L’écriture était devenue presque dix fois plus longue.

Ces durées sont cumulées, pas écoulées

La somme des trois phases (19,6 s) dépasse de loin la durée de la phase map (5,11 s), puisque les quatre travailleurs œuvrent en parallèle. C’est bien ce qu’on veut ici : on cherche où part le travail, pas combien de temps on a attendu.

La lecture varie d’une mesure à l’autre

Selon les essais, la lecture totale a varié de 0,8 s à 3,9 s. Les fichiers d’entrée sont sur le NFS : si une machine les a lus récemment, ils sont encore dans sa mémoire (le cache), et la lecture est presque instantanée. Une seule mesure ne suffit jamais.

Exercice 7.4 — Courbe d’accélération

acceleration.py

"""Courbe d'acceleration (exercice 7.4).

    python3 deployer.py deployer 16      # une seule fois
    python3 acceleration.py              # p = 1, 2, 4, 8, 16
"""
import statistics
import time

from maitre import mapreduce_distribue

travailleurs = [l.strip() for l in open("travailleurs.txt") if l.strip()]
fichiers = [l.strip() for l in open("fichiers.txt") if l.strip()]

reference = None
T = {}
for p in (1, 2, 4, 8, 16):
    durees = []
    for essai in range(3):                       # 3 mesures par point
        debut = time.perf_counter()
        resultat, _ = mapreduce_distribue(travailleurs[:p], fichiers)
        durees.append(time.perf_counter() - debut)
        if reference is None:
            reference = resultat
        assert resultat == reference             # MEME resultat partout
    T[p] = statistics.median(durees)
    print(f"p={p:2d}  mesures {' '.join(f'{d:6.2f}' for d in durees)}"
          f"  mediane T={T[p]:6.2f} s  S={T[1] / T[p]:5.2f}", flush=True)

Sortie réelle (le déploiement des 16 travailleurs a pris 19 s) :

p= 1  mesures  20.41  18.38  18.29  mediane T= 18.38 s  S= 1.00
p= 2  mesures   9.66   9.47   9.35  mediane T=  9.47 s  S= 1.94
p= 4  mesures   6.06   5.03   4.86  mediane T=  5.03 s  S= 3.66
p= 8  mesures   3.17   2.97   2.77  mediane T=  2.97 s  S= 6.19
p=16  mesures   2.05   2.02   2.02  mediane T=  2.02 s  S= 9.09

Remarquez la première mesure de $p = 1$ (20,41 s), plus lente que les deux suivantes : c’est l’effet du cache, les fichiers n’avaient pas encore été lus. La médiane l’écarte.

1. Le tableau, avec $S$ calculé à partir des médianes arrondies :

$p$	$T(p)$ (s)	$S(p)$	$E(p)$	$f$ estimé	plafond $1/f$
1	18,38	1,00	100 %	—	—
2	9,47	1,94	97 %	0,031	32
4	5,03	3,65	91 %	0,032	31
8	2,97	6,19	77 %	0,042	24
16	2,02	9,10	57 %	0,051	20

2. Pour $p = 16$ : $f = (1/9\mathord{,}10 - 1/16) / (1 - 1/16) \approx 0\mathord{,}051$ . Les autres points donnent des valeurs différentes, de 0,031 à 0,051 : $f$ augmente avec $p$ . La loi d’Amdahl suppose une part séquentielle fixe et une part parallèle qui se divise parfaitement. Ici, une partie des pertes grandit avec $p$ (question 4), et le modèle la compte dans $f$ .

3. Avec $f \approx 0\mathord{,}05$ , le plafond est $1/f \approx 20$ . Au vu de la tendance de $f$ , le plafond réel est sans doute plus bas.

4. L’efficacité passe de 97 % à 57 % :

les vagues : un map dure environ 0,6 s (exercice 7.3). Avec $p$ travailleurs, chacun enchaîne $\lceil 32/p \rceil$ maps : 32 vagues pour $p = 1$ , 2 seulement pour $p = 16$ . Le map ne peut plus descendre sous 0,6 s, quelle que soit la valeur de $p$ au-delà de 32 ;
$R = 8$ : au-delà de 8 machines, le reduce n’accélère plus, faute de tâches. Il pèse 0,9 s avec 4 machines (exercice 7.3), soit déjà 15 % du total ; sa part grandit quand le map raccourcit ;
les coûts fixes : la collecte, séquentielle, et l’établissement des connexions.

La courbe ne redescend pas encore à 16 machines, mais elle s’aplatit nettement.

5. Le déploiement (19 s) dure presque dix fois $T(16)$ . Déploiement compris, 16 machines mettraient environ 21 s, plus qu’une seule machine sans déploiement (18,4 s) ! Distribuer n’est rentable que si le déploiement est amorti sur beaucoup de calculs, ou si le calcul est beaucoup plus gros. Et notre système gâche 11 cœurs sur 12 par machine (GIL) : en séance 6, une machine avec 12 processus faisait le même travail en 1,8 s.

Exercice 7.5 — Tuer une machine

Sorties réelles, 32 fichiers de 1 Mo, 4 travailleurs neufs à chaque fois. Sans panne, le calcul dure environ 6,1 s (exercice 7.3).

1. Sans relance.

   >>> t =   2.5 s : travailleur TUE sur tp-1a201-28.enst.fr
  map, essai 1 : 4 tache(s) a relancer, 3 travailleur(s) restant(s)
RuntimeError: 4 map(s) impossibles a terminer

Le calcul échoue, mais le maître n’attend pas 60 s. Seul le processus est mort ; la machine, elle, fonctionne. Son système répond aussitôt à toute nouvelle connexion par un refus (Connection refused), et coupe celles qui étaient en cours : le maître l’apprend en quelques millisecondes. Le timeout ne sert que lorsque plus rien ne répond (question 5).

2. Pendant le map, avec relance.

   >>> t =   2.4 s : travailleur TUE sur tp-1a201-29.enst.fr
  map, essai 1 : 4 tache(s) a relancer, 3 travailleur(s) restant(s)
  apres la barriere : 1 travailleur(s) mort(s), 4 map(s) a refaire
t =   7.4 s : termine avec 3 travailleur(s) sur 4
identique au moteur sequentiel : True

Le résultat est identique, en 7,4 s au lieu de 6,1 s. Deux lignes :

essai 1 : la victime devait faire 8 maps. Elle en avait terminé 4 quand elle est morte ; les 4 autres ont échoué, et phase_map les a relancés sur les 3 survivants ;
après la barrière : les 4 maps que la victime avait terminés sont dans sources. Au reduce, les réducteurs ne peuvent pas lire ses partitions : il faut refaire ces 4 maps ailleurs, puis relancer le reduce.

C’est possible parce qu’un map est idempotent : le refaire produit exactement les mêmes paires.

3. Après la barrière.

   >>> t =   5.3 s : travailleur TUE sur tp-1a201-32.enst.fr
  apres la barriere : 1 travailleur(s) mort(s), 8 map(s) a refaire
t =   7.9 s : termine avec 3 travailleur(s) sur 4
identique au moteur sequentiel : True

Les 8 maps de la victime étaient tous terminés, et ils sont tous refaits : leurs fichiers intermédiaires étaient sur le disque local de la machine morte, donc inaccessibles. C’est la ligne la plus surprenante du tableau du cours, mesurée.

4. Avant la collecte.

   >>> t =   5.5 s : travailleur TUE sur tp-1a201-34.enst.fr
  apres la barriere : 1 travailleur(s) mort(s), 8 map(s) a refaire
t =   8.3 s : termine avec 3 travailleur(s) sur 4
identique au moteur sequentiel : True

Les reduces de la victime étaient terminés, mais leurs résultats étaient dans son /tmp. Le maître doit donc refaire ses 8 maps, puis tout le reduce, avant de collecter. Chez Google, les reduces écrivent dans GFS, un système de fichiers global et répliqué : un reduce terminé survit à sa machine, et il n’y aurait rien à refaire pour lui.

5. Machine figée.

   >>> t =   2.4 s : travailleur FIGE sur tp-1a201-37.enst.fr
  map, essai 1 : 4 tache(s) a relancer, 3 travailleur(s) restant(s)
  apres la barriere : 1 travailleur(s) mort(s), 4 map(s) a refaire
t =  88.7 s : termine avec 3 travailleur(s) sur 4
identique au moteur sequentiel : True

Même déroulement qu’à la question 2, mais 88,7 s au lieu de 7,4 s. Un processus figé par SIGSTOP ne répond plus, mais le système de sa machine accepte encore les connexions (elles s’empilent dans la file d’attente de listen). Rien n’est refusé : seul le timeout peut révéler la panne. On paie donc 60 s au map (le delai de demander), puis 20 s au shuffle, quand les réducteurs tentent de lire chez la victime (le settimeout(20.0) de faire_reduce). C’est le cas d’une machine débranchée ou d’un réseau coupé, le plus coûteux.

Exercice 7.6 — Un autre traitement

1. Les lignes qui supposent un comptage. Le maître, le protocole, le déploiement, le shuffle et la tolérance aux pannes n’en contiennent aucune. Dans le travailleur, il y en a trois :

fonction_map elle-même ;
le combiner de faire_map : seaux[r][cle] = seaux[r].get(cle, 0) + valeur, qui additionne ;
l’agrégation de faire_reduce : total[cle] = total.get(cle, 0) + valeur, qui additionne aussi.

Avec des noms de fichiers comme valeurs (index inversé), le + lèverait TypeError. Avec des moyennes, il additionnerait des choses qu’il ne faut pas additionner.

2. Le travailleur générique. Trois changements par rapport au travailleur du cours :

from traitement import fonction_map, fonction_reduce   # 1. en tete

# 2. dans faire_map : on regroupe les valeurs en listes, sans additionner
    seaux = [dict() for _ in range(R)]          # cle -> liste des valeurs
    for cle, valeur in fonction_map(requete["fichier"], contenu):
        seaux[partition(cle, R)].setdefault(cle, []).append(valeur)

# 3. dans faire_reduce : on concatene les listes, puis fonction_reduce
    groupes = {}
    ...
        for cle, valeurs in reponse["donnees"].items():
            groupes.setdefault(cle, []).extend(valeurs)

    total = {cle: fonction_reduce(cle, valeurs)
             for cle, valeurs in groupes.items()}

et on supprime la fonction_map du travailleur. Le réducteur reçoit désormais, comme en séance 6, la liste complète des valeurs de chaque clé.

Pour choisir le traitement, traitement.py tient en deux lignes :

traitement.py

from fonctions_62 import map_longueurs as fonction_map
from fonctions_62 import reduce_somme as fonction_reduce

fonctions_62.py, le corrigé de l’exercice 6.2, est utilisé sans aucune modification. On change de traitement en éditant ces deux lignes, puis en redéployant (nettoyer, puis deployer 4) : chaque travailleur lit traitement.py au démarrage.

3. La vérification, pour chaque traitement, avec les deux fonctions importées de fonctions_62.py :

distribue, _ = maitre.mapreduce_distribue(travailleurs, fichiers)
reference = mapreduce(fichiers, fonction_map, fonction_reduce)
en_chaines = {str(cle): valeur for cle, valeur in reference.items()}
print(len(distribue), distribue == reference, distribue == en_chaines)

Résultats réels, 8 fichiers de 1 Mo, 4 travailleurs :

Traitement	Clés	`== reference`	`== en_chaines`
comptage	18 810	`True`	`True`
longueurs	27	`False`	`True`
index	18 810	`True`	`True`
moyenne	55	`True`	`True`

Les longueurs semblent fausses. Le résultat distribué contient ('10', 34533), le séquentiel (10, 34533) : la clé est devenue une chaîne. Les partitions voyagent en JSON, et JSON n’accepte que des chaînes comme clés de dictionnaire : json.dumps({10: 1}) donne '{"10": 1}'. Les valeurs sont justes ; seul leur type a changé. Une fois les clés de la référence converties en chaînes, les deux résultats sont identiques. Le remède : émettre directement str(len(mot)), ou reconvertir à la fin.

L’index inversé et la moyenne passent tels quels, bien que leurs valeurs soient des noms de fichiers et des couples : le partitionnement utilise str(cle), identique sur toutes les machines, et les couples deviennent des listes JSON, ce qui ne gêne pas reduce_moyenne.

4. Le combiner perdu. Sur le comptage de mots, 8 fichiers de 1 Mo : 2,4 Mo de fichiers intermédiaires avec le travailleur du cours, 6,4 Mo avec le travailleur générique. Le générique envoie un 1 par occurrence au lieu d’un total par mot : 2,7 fois plus de données à écrire et à transférer.

On ne peut pas remettre le combiner « pour tout le monde ». Combiner, c’est appliquer la réduction en avance, sur un morceau des valeurs. C’est permis pour une somme, car elle est associative et commutative. Pour une moyenne, c’est faux : la moyenne des moyennes n’est pas la moyenne, c’est exactement le piège de l’exercice 6.6.

Ce que cela dit du modèle. La logique métier tient en quelques lignes, face à une infrastructure qu’on n’a pas eu à toucher. Mais l’exercice montre aussi ce qu’un vrai moteur doit rendre paramétrable : tout ce qu’il suppose du type des valeurs, et l’existence ou non d’un combiner. Hadoop a une classe Combiner optionnelle, distincte du Reducer ; Spark distingue reduceByKey, qui combine localement, de groupByKey, qui ne le fait pas.

Exercice 7.7 — Quel composant améliorer ?

1. « 100 % plus rapide » : deux fois plus de travail par seconde, donc $k = 2$ . « 400 % plus rapide » : cinq fois plus, donc $k = 5$ . L’option A touche le calcul, $a = 0\mathord{,}79$ ; l’option B touche la lecture et l’écriture, $a = 0\mathord{,}15 + 0\mathord{,}06 = 0\mathord{,}21$ .

Option	$S = 1 / \big((1-a) + a/k\big)$	Gain	Coût par % de gain
A	$1 / (0\mathord{,}21 + 0\mathord{,}79/2) = 1 / 0\mathord{,}605 = 1\mathord{,}65$	+65 %	$12\,000 / 65 \approx$ 184 €
B	$1 / (0\mathord{,}79 + 0\mathord{,}21/5) = 1 / 0\mathord{,}832 = 1\mathord{,}20$	+20 %	$3\,000 / 20 \approx$ 149 €

Meilleure performance : A, parce que le calcul pèse 79 % du temps. Meilleur rapport performance / coût : B, quatre fois moins cher.

3. Les 18,38 s se décomposent en $0\mathord{,}79 \times 18\mathord{,}38 = 14\mathord{,}52$ s de calcul et $18\mathord{,}38 - 14\mathord{,}52 = 3\mathord{,}86$ s de lectures et écritures.

Option	Calcul	Lecture et écriture	Total
Aujourd’hui	14,52 s	3,86 s	18,38 s
A	$14\mathord{,}52 / 2 = 7\mathord{,}26$ s	3,86 s	11,12 s
B	14,52 s	$3\mathord{,}86 / 5 = 0\mathord{,}77$ s	15,29 s

Seule l’option A passe sous 12 s. Le meilleur rapport performance / coût (B) ne permet pas d’atteindre l’objectif : c’est la question 3 du cours.

4. $1 / (1 - 0\mathord{,}21) = 1 / 0\mathord{,}79 \approx 1\mathord{,}27$ : même avec des disques infiniment rapides, au mieux +27 %, soit 14,52 s. Le calcul resterait là, intact.

7.13 Ce qu’il faut retenir

L’essentiel de la séance

Un maître distribue et surveille ; il ne calcule pas. Les travailleurs calculent. Un déployeur les lance et les arrête.
Les fichiers intermédiaires vont sur le disque local (/tmp), jamais dans le répertoire personnel NFS : le disque local passe à l’échelle, le NFS non.
Machines partagées : un port à votre nom (uid), un dossier à votre nom, un PING qui vérifie le propriétaire, une connexion SSH à la fois.
Déploiement : un seul scp, puis un ssh par machine. Le nettoyage s’écrit en même temps, et se vérifie : pkill -u $USER -f '[t]ravailleur.py'.
L’écriture atomique (fichier .tmp puis os.replace) évite les fichiers à moitié écrits.
Mesurer chaque phase séparément, avec perf_counter(), plusieurs fois.
Une courbe d’accélération n’a de sens qu’à jeu de données constant.
Loi d’Amdahl : $S(p) = 1/(f + (1-f)/p)$ , plafond $1/f$ . Réduire $f$ vaut mieux qu’ajouter des machines.
Tolérance aux pannes = refaire la tâche ailleurs, ce qui exige des tâches idempotentes. Tout ce qui était sur le disque local d’une machine morte est perdu, même terminé.
Un processus mort se voit tout de suite ; une machine figée ne se révèle que par le timeout.
Toujours vérifier le résultat contre un calcul de référence.

7.1 L’architecture

Les fichiers de la séance

Le déroulement dans le temps

7.2 Où écrire les fichiers intermédiaires ?

Votre répertoire personnel n’est pas sur la machine

Ce que coûte une écriture sur le NFS

Le vrai problème : l’échelle

7.3 Partager les machines avec les autres

Règle 1 : un port à votre nom

Règle 2 : vérifier que c’est votre travailleur qui répond

Règle 3 : une connexion SSH à la fois

Règle 4 : nettoyer — et vérifier que le nettoyage a marché

7.4 Déployer sur NN machines

À la main : un scp, puis un ssh par machine

Le script de déploiement

7.5 Le travailleur

Lecture guidée

7.6 Le maître

Lecture guidée

7.7 Mesurer

Chronométrer chaque phase séparément

L’accélération et l’efficacité

Nos mesures

7.8 La loi d’Amdahl

Un problème pour commencer : processeur ou disque ?

Le raisonnement, en minutes

La formule générale

Les réponses

Appliquée à notre système

Le parallélisme : un cas particulier

Ce que cela veut dire concrètement

Estimer ff à partir de vos mesures

Pourquoi la courbe peut redescendre

7.9 Survivre à une panne

Deux façons de mourir

Réagir : relancer la tâche ailleurs

7.10 Après MapReduce

Les moteurs de flot de données : Spark

Du lot au flux : le traitement en continu

7.11 Exercices

7.12 Corrigés

7.13 Ce qu’il faut retenir

7.4 Déployer sur $N$ machines

À la main : un `scp`, puis un `ssh` par machine

Estimer $f$ à partir de vos mesures