Compter les mots en une ligne

cat corpus/*.txt \
    | grep -oE '[[:alpha:]]+' \
    | awk '{print tolower($0)}' \
    | sort \
    | uniq -c \
    | sort -rn | head -8

Attention aux accents

La recette classique, tr -c '[:alpha:]' '\n', détruit les accents sous Linux : tr y travaille octet par octet, et é occupe deux octets.

« L’été, déjà là » → l t d j l

Ce qui bloque

• Un disque, un jeu de processeurs
• Aucune reprise sur panne : mort à 90 %, tout est perdu
• Impossible à répartir sur cent machines

Des couples clé-valeur

• ("chat", 1) : la clé dit de quoi on parle, la valeur ce qu’on en sait
• map : un morceau de l’entrée → des couples (clé, valeur)
• shuffle : rassembler toutes les valeurs de même clé (le système)
• reduce : une clé et toutes ses valeurs → le résultat

Le comptage de mots

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)

yield : un générateur qui livre les couples un par un, sans tout garder en mémoire.

Sur un exemple minuscule

Le moteur

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

Changer les fonctions, pas le moteur

Qui traite quel mot ?

Avec cent machines, les couples sont éparpillés. Il faut décider à l’avance quel réducteur traite quelle clé, sans se concerter.

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

$M \times R$ fichiers

• $M$ tâches map (une par fichier), $R$ tâches reduce
• Chaque map écrit $R$ fichiers, un par réducteur
• Le réducteur $r$ lit les $M$ fichiers qui lui sont destinés

Le combiner

Au lieu d’écrire mille fois ("le", 1), écrire une fois ("le", 1000).

Seulement si reduce le permet

Le combiner réduit en deux temps. Il faut que ce soit sans effet :

• associative : $(2 + 3) + 4 = 2 + (3 + 4)$
• commutative : $2 + 3 = 3 + 2$

Cœurs et processus

• Un cœur : une unité de calcul. Machine de salle mesurée : i7-8700, 6 physiques, 12 logiques
• Un processus : un programme en cours, avec sa propre mémoire
• GIL : les threads Python ne calculent jamais en même temps → il faut des processus

multiprocessing.Pool(p) : $p$ ouvriers. pool.map(f, liste) = un [f(x) for x in liste] en parallèle, qui attend tout le monde.

Le moteur parallèle

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

    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)])

Deux règles impératives

• if __name__ == "__main__": obligatoire. Sur Mac (spawn), sans lui : 958 RuntimeError en 20 s, jamais fini. À l’école (fork, Python 3.11) : ça marche… jusqu’au jour où l’on change de machine
• Tout passer en argument. Un ouvrier n’entre jamais dans le bloc __main__ : une variable globale calculée là lui est inconnue

Mesuré sur une machine de salle

8 fichiers de 1 Mo, $R = 4$, tp-1a201-18

Des vagues de tâches

Une tâche map = 0,54 s. Le processus le plus chargé fixe le temps.

Ce qui empêche d’atteindre 8

• Le reduce : seulement $R = 4$ tâches, donc au plus 4 fois plus vite
• La barrière entre map et reduce
• Le coût de création des processus

Les exercices

• 6.1 — Le moteur. Vérifier à la main, puis 18 810 mots distincts, puis comparer avec Unix.
• 6.2 — D’autres fonctions. Longueurs, index inversé, moyenne par initiale.
• 6.3 — Le partitionnement. Trois propriétés vérifiées par assert.
• 6.4 — Le combiner. Mesurer le gain, sur 1 Mo puis 8 Mo.
• 6.5 — L’accélération. Sur une machine de l’école, 1 à 8 processus.
• 6.6 — La moyenne. Un bug invisible sur le corpus.

Ce qu’il faut retenir

• MapReduce = map | sort | reduce à grande échelle ; le sort, c’est le shuffle
• On écrit map et reduce, le système fait le reste
• hash(cle) % R : déterministe, sans concertation
• $M \times R$ fichiers : le shuffle est la phase la plus coûteuse
• Le combiner divise le volume, si reduce est associative
• La barrière est la part séquentielle
• Plus de tâches que de processus, sinon les cœurs attendent