5 Architectures des systèmes de données

Objectifs de la séance

Savoir mesurer honnêtement la performance d’un système (percentiles plutôt que moyenne).
Comprendre les deux grandes techniques de répartition : réplication et partitionnement.
Comprendre pourquoi détecter une panne est un problème fondamentalement difficile, et comment on évite d’avoir deux leaders (split brain).
Savoir ce que dit vraiment le théorème CAP.
Situer ces idées dans les outils réels (HDFS, Kafka, bases de données distribuées).

Références

Cette séance s’inspire de l’ouvrage de référence du domaine : Martin Kleppmann et Chris Riccomini, Designing Data-Intensive Applications, O’Reilly. Nous n’en abordons que les idées principales, sans entrer dans le détail : l’objectif est de vous donner le vocabulaire et les bons réflexes.

Diaporama

Les diapositives de cette séance

5.1 Trois qualités attendues d’un système

Fiabilité — Le système continue à fonctionner correctement même quand quelque chose va mal (panne matérielle, bug logiciel, erreur humaine).
Scalabilité — Quand la charge augmente, il existe un moyen raisonnable d’y faire face.
Maintenabilité — D’autres personnes que vous peuvent le comprendre, le faire évoluer, le réparer.

Ces trois mots reviendront constamment. Retenez surtout que la scalabilité n’est pas une propriété binaire : la bonne question n’est jamais « est-ce que ce système est scalable ? » mais « si la charge est multipliée par 10, que faut-il changer ? ».

5.2 Deux mondes : opérationnel et analytique

Avant d’entrer dans les mécanismes, il faut savoir distinguer deux familles de systèmes, car elles n’ont ni les mêmes contraintes ni les mêmes solutions. C’est la première distinction que pose la littérature du domaine, et elle structure tout le vocabulaire du Big Data.

On les désigne par deux acronymes que vous rencontrerez partout :

OLTP, OnLine Transaction Processing : le traitement transactionnel en ligne. C’est le système qui fait tourner l’application au quotidien — enregistrer une commande, lire une fiche client.
OLAP, OnLine Analytical Processing : le traitement analytique en ligne. C’est le système qui répond aux questions d’ensemble — des tendances, des totaux, des classements.

Le mot transaction y désigne simplement un groupe d’opérations liées (débiter un compte et créditer l’autre), pas nécessairement une opération financière.

	Opérationnel (OLTP)	Analytique (OLAP)
Question type	« quelle est l’adresse du client 4712 ? »	« quel est le chiffre d’affaires par région en mars ? »
Accès	quelques enregistrements par clé	des millions de lignes, quelques colonnes
Écriture	permanente, par petites touches	par lots massifs (import périodique)
Utilisateur	l’application, en direct	un analyste, un tableau de bord
Critère	latence très faible	débit très élevé
Exemple	PostgreSQL, MySQL	entrepôt de données, Hadoop, Spark

On sépare généralement les deux physiquement : une longue requête analytique qui balaie dix ans d’historique ralentirait le site de vente. Les données sont donc copiées périodiquement du monde opérationnel vers un entrepôt de données (data warehouse), par un processus appelé ETL : Extract (extraire), Transform (nettoyer, harmoniser), Load (charger).

Pourquoi l’analytique stocke en colonnes

Une table de ventes a 100 colonnes. La requête « total des ventes par région » n’en utilise que deux. Un stockage classique, par lignes, oblige pourtant à lire les 100 colonnes de chaque ligne : on transfère 50 fois plus d’octets que nécessaire.

Le stockage en colonnes range ensemble toutes les valeurs d’une même colonne. On ne lit alors que les deux colonnes utiles. Deux bénéfices s’ajoutent : les valeurs d’une même colonne se ressemblent, donc elles se compressent très bien ; et le processeur peut les traiter en série, sans sauter d’une zone mémoire à l’autre.

Ce que vous rencontrerez

C’est exactement ce que font les formats Parquet et ORC, omniprésents dans l’écosystème Hadoop et Spark. Quand on vous dira « convertis ce CSV en Parquet », il s’agira de passer d’un stockage par lignes à un stockage par colonnes. Voir Kleppmann & Riccomini, Designing Data-Intensive Applications, chapitres 1 et 4.

Pourquoi c’est important

Le MapReduce que nous écrirons en séances 6 et 7 est un outil du monde analytique : il balaie tout le jeu de données pour produire un résumé. Il serait absurde de l’utiliser pour répondre à « quelle est l’adresse du client 4712 ? ». Savoir dans quel monde on se trouve est la première décision d’architecture.

5.3 Mesurer la performance honnêtement

Latence et débit

Latence — le temps que met une opération. Se mesure en millisecondes.
Débit — le nombre d’opérations par seconde. Se mesure en requêtes/s, ou en Mo/s.

Ce ne sont pas la même chose, et améliorer l’un peut dégrader l’autre. Un train est un excellent moyen de transport en débit (beaucoup de passagers par heure), et médiocre en latence (il ne part pas quand vous arrivez).

Pourquoi la moyenne ment

Supposons que votre service réponde en 10 ms pour 999 requêtes sur 1000, et en 5 secondes pour la millième.

$\text{moyenne} = \frac{999 \times 10 + 1 \times 5000}{1000} \approx 15 \text{ ms}$

Sur le tableau de bord, tout va bien : 15 ms de moyenne. Pourtant, un utilisateur sur mille attend cinq secondes. S’il visite dix pages, il a environ 1 % de chances de subir ce délai — et c’est probablement votre meilleur client, celui qui a le plus de données.

À retenir

On ne mesure jamais un système distribué avec une moyenne. On utilise des percentiles :

p50 (médiane) : la moitié des requêtes sont plus rapides.
p95 : 95 % des requêtes sont plus rapides ; 1 sur 20 est plus lente.
p99 : 1 requête sur 100 est plus lente.
p99,9 : 1 sur 1000. C’est celle que regardent les grandes plateformes.

Calculer des percentiles

def percentile(valeurs, p):
    """p entre 0 et 100. Renvoie le percentile p de la liste."""
    triees = sorted(valeurs)
    if not triees:
        return None
    indice = int(round((p / 100.0) * (len(triees) - 1)))
    return triees[indice]

mesures = [12, 11, 13, 10, 12, 14, 11, 950, 12, 13]
print("p50  :", percentile(mesures, 50))    # 12
print("p95  :", percentile(mesures, 95))    # 950
print("moy  :", sum(mesures) / len(mesures))  # 105.8

La queue de distribution. Si l’on trace le nombre de requêtes en fonction de leur durée, on obtient une bosse autour des valeurs habituelles, puis une longue traîne fine vers la droite : les rares requêtes très lentes. Cette traîne s’appelle la queue de distribution, comme la queue d’un animal (en anglais tail, d’où l’expression tail latency). Le mot n’a ici rien à voir avec une file d’attente. Les percentiles hauts, p99 et p99,9, sont précisément les indicateurs qui regardent cette queue.

Un point qui surprend

Dans un système distribué, les percentiles s’aggravent avec le parallélisme. Si votre page appelle 20 services en parallèle et attend qu’ils aient tous répondu, elle est aussi lente que le plus lent des 20. Même si chaque service a un p99 excellent, la probabilité qu’au moins un des 20 soit dans son 1 % le plus lent est de $1 - 0{,}99^{20} \approx 18\,\%$ . Autrement dit : une page sur cinq est lente, alors que chaque service pris seul n’est lent qu’une fois sur cent. C’est ce qu’on appelle l’amplification de la queue de distribution.

5.4 Répliquer : plusieurs copies de la même donnée

Trois raisons de répliquer

Disponibilité : si une machine tombe, une autre a la donnée.
Performance en lecture : dix copies peuvent servir dix fois plus de lectures.
Proximité : une copie en Europe, une aux États-Unis, chacun lit près de chez lui.

Le modèle leader / suiveurs

C’est de loin le plus répandu (PostgreSQL, MySQL, MongoDB, Kafka…).

Une machine est le leader. Toutes les écritures passent par elle.
Les autres sont des suiveurs. Elles reçoivent le journal des modifications du leader et l’appliquent.
Les lectures peuvent être servies par n’importe qui.

Le retard de réplication

Les suiveurs sont toujours un peu en retard sur le leader. Ce délai s’appelle le replication lag. Il vaut typiquement quelques millisecondes, mais peut atteindre plusieurs secondes en cas de charge.

Ce retard produit des situations déroutantes :

Vous publiez un commentaire (écriture $\rightarrow$ leader). Vous rechargez la page (lecture $\rightarrow$ un suiveur qui n’a pas encore reçu la mise à jour). Votre commentaire a disparu.

Attention

Ce n’est pas un bug : c’est la conséquence directe d’un choix d’architecture. C’est exactement ce phénomène qui explique qu’un virement bancaire « apparaisse dans 24 h », ou qu’une vidéo mise en ligne ne soit pas immédiatement visible partout. La parade classique s’appelle read-your-writes : après une écriture, on force l’utilisateur à lire sur le leader pendant quelques secondes.

Que se passe-t-il si le leader tombe ?

Il faut élire un nouveau leader parmi les suiveurs : c’est le basculement (failover). Trois questions se posent, et aucune n’a de réponse facile :

Comment être sûr que le leader est mort, et pas seulement lent ? (voir plus bas)
Quel suiveur choisir ? Le plus à jour, mais comment le savoir ?
Que faire des écritures que l’ancien leader avait acceptées mais pas encore transmises ? En général : on les perd.

Le split brain : deux leaders à la fois

Le pire scénario du basculement porte un nom imagé : le split brain, le « cerveau coupé en deux ». Le système se retrouve avec deux leaders, qui acceptent chacun des écritures sans savoir que l’autre existe.

Comment cela arrive. Prenons trois machines, A (le leader), B et C, et déroulons le scénario pas à pas :

Le câble réseau qui relie A aux deux autres est coupé. A fonctionne toujours parfaitement : il est seulement isolé.
B et C ne reçoivent plus les signes de vie de A. Ils ne peuvent pas savoir si A est mort ou simplement injoignable (nous verrons plus bas pourquoi c’est impossible). Ils concluent qu’il est mort et élisent B comme nouveau leader.
A, de son côté, n’a rien remarqué d’anormal. Des clients qui sont du même côté de la coupure que lui continuent à lui envoyer des écritures, et il les accepte.
Pendant ce temps, B accepte d’autres écritures, venues des clients de l’autre côté.

Résultat : il reste un seul exemplaire en stock, A vend cet exemplaire à Alice, B le vend à Bob. Quand le réseau revient, les deux historiques sont contradictoires, et aucun des deux n’est « le bon ».

Pourquoi c’est si grave

Ce n’est pas une donnée perdue, c’est une donnée incohérente : le système a promis deux choses incompatibles à deux clients différents. Les dégâts sont visibles (deux clients pour un objet, un compte bancaire débité deux fois) et ne se réparent qu’à la main, après coup.

Comment l’empêcher. On ne peut pas empêcher les coupures réseau. On peut en revanche faire en sorte que deux leaders ne puissent jamais agir en même temps. Voici les quatre solutions les plus répandues, de la plus fondamentale à la plus brutale. Les systèmes réels en combinent souvent plusieurs.

1. La majorité (le quorum). Un leader n’a le droit d’agir que s’il est soutenu par une majorité stricte des machines, c’est-à-dire plus de la moitié. Avec 3 machines, il en faut 2 ; avec 5, il en faut 3.

L’astuce est arithmétique : quand le réseau se coupe en deux groupes, au plus un des deux groupes contient une majorité. Dans notre scénario, B et C forment 2 machines sur 3 : ils ont la majorité, ils peuvent élire B. A est seul, 1 sur 3 : il sait qu’il n’a plus la majorité, donc il arrête d’accepter des écritures. Il ne peut plus y avoir deux leaders actifs.

Machines	Majorité	Pannes tolérées
3	2	1
4	3	1
5	3	2
7	4	3

C’est pourquoi les grappes de coordination ont presque toujours un nombre impair de machines. Passer de 3 à 4 coûte une machine de plus sans tolérer une panne de plus. Pire : si 4 machines se coupent en deux groupes de 2, aucun groupe n’a la majorité, et plus personne ne peut écrire.

2. Le numéro de mandat. Chaque élection attribue au nouveau leader un numéro strictement plus grand que le précédent : on parle de numéro de mandat, d’époque ou de term. Chaque écriture porte le numéro du leader qui l’envoie. Les machines retiennent le plus grand numéro qu’elles ont vu, et refusent toute écriture portant un numéro plus petit.

Dans notre scénario, A était leader du mandat 7 ; B est élu avec le mandat 8. Quand A revient et tente d’écrire avec le mandat 7, tout le monde le rejette : il comprend qu’il a été remplacé et redevient suiveur. Quand ce numéro est aussi vérifié par le stockage lui-même, on parle de fencing token (jeton de clôture).

3. Le bail (lease). Le leader ne l’est que pour une durée limitée, par exemple 10 secondes, et doit renouveler son bail auprès de la majorité avant qu’il expire. S’il est isolé, il ne peut plus le renouveler, et il s’arrête de lui-même à l’expiration. Les autres attendent au moins cette durée avant d’élire un remplaçant. Attention : cette méthode suppose que les horloges des machines avancent à peu près à la même vitesse, ce qui n’est pas garanti (voir Section 5.7). On la combine donc avec le numéro de mandat.

4. Abattre l’ancien leader. Avant de promouvoir un nouveau leader, on s’assure que l’ancien est réellement hors service, en lui coupant l’alimentation électrique à distance (par la carte d’administration du serveur ou la prise électrique pilotée). La méthode porte un nom volontairement brutal : STONITH, Shoot The Other Node In The Head. Elle est courante dans les grappes classiques à deux serveurs, où aucune majorité n’est possible.

Ce que font les systèmes réels

Personne ne réécrit tout cela à la main : on le délègue à un service de coordination, comme ZooKeeper ou etcd. Ils fonctionnent sur 3 ou 5 machines et mettent en œuvre un algorithme de consensus (Raft, Paxos, Zab) qui combine la majorité et le numéro de mandat. Les autres logiciels leur demandent simplement « qui est le leader ? ». HDFS s’en sert pour choisir son NameNode actif, et Kafka pour désigner son contrôleur, qui choisit à son tour le leader de chaque partition (aujourd’hui avec sa propre mise en œuvre de Raft, nommée KRaft). Voir Kleppmann & Riccomini, Designing Data-Intensive Applications, chapitres 9 et 10.

À retenir

Le split brain se produit quand deux machines se croient leader en même temps, typiquement après une coupure réseau. On l’empêche en exigeant une majorité pour agir (d’où les grappes de 3 ou 5 machines) et en numérotant les mandats pour que l’ancien leader soit rejeté. Le prix à payer : le côté minoritaire de la coupure cesse de répondre.

5.5 Partitionner : découper la donnée

La réplication met la même donnée sur plusieurs machines. Le partitionnement (ou sharding) met des données différentes sur des machines différentes. Les deux se combinent presque toujours.

Pourquoi c’est important

La réplication seule ne règle pas le problème du volume : si votre jeu de données fait 10 To et que vos machines ont 1 To de disque, aucune copie ne tient. Il faut découper. C’est aussi la seule façon de paralléliser les écritures.

Partitionner par plage de clés

On attribue à chaque machine un intervalle : les clés de A à F sur la machine 0, de G à M sur la machine 1, etc.

Avantage. Les requêtes par intervalle sont efficaces (« tous les clients dont le nom commence par D » va sur une seule machine).

Inconvénient. La répartition est presque toujours déséquilibrée. Si vous partitionnez des mesures de capteurs par horodatage, toutes les écritures du jour vont sur la même machine : c’est un point chaud (hot spot).

Partitionner par hachage

On calcule une empreinte de la clé, et on prend le reste de la division par le nombre de machines :

$\texttt{machine} = \texttt{hash(cle)} \bmod N$

Partitionnement par hachage

import hashlib

def partition(cle, nb_partitions):
    """Renvoie un numero de partition stable entre 0 et nb-1."""
    empreinte = hashlib.md5(cle.encode("utf-8")).hexdigest()
    return int(empreinte, 16) % nb_partitions

for mot in ["chat", "chien", "oiseau", "poisson"]:
    print(f"{mot:10s} -> partition {partition(mot, 4)}")

Attention

N’utilisez pas la fonction hash() native de Python pour cela. Depuis Python 3.3, elle est aléatoirement salée au démarrage : hash("chat") donne une valeur différente à chaque exécution du programme. Deux machines de votre système calculeraient donc des partitions différentes pour la même clé, et vos données seraient introuvables. Utilisez hashlib.md5 ou zlib.crc32, qui sont déterministes. C’est un piège classique — et une question d’examen.

Avantage. Répartition très équilibrée.

Inconvénients. On perd les requêtes par intervalle. Et surtout : si $N$ change (on ajoute une machine), presque toutes les clés changent de partition. Il faut alors tout déplacer.

Comment les systèmes réels évitent de tout déplacer

La méthode la plus simple consiste à créer dès le départ beaucoup plus de partitions que de machines, par exemple 110 partitions pour 10 machines, et à tenir une table « partition → machine ». La partition d’une clé ne change jamais (hash(cle) % 110). Quand on ajoute une machine, on lui confie quelques partitions entières, prises aux autres : seules les clés de ces partitions-là déménagent. Kafka et Elasticsearch fonctionnent ainsi, et fixent le nombre de partitions à la création.

Une variante plus élaborée, le hachage cohérent (consistent hashing), place machines et clés sur un même cercle de valeurs de hachage : chaque clé appartient à la première machine rencontrée en tournant. Ajouter une machine ne lui fait prendre que les clés d’un arc du cercle. L’exercice 5.3 vous fait mesurer la différence.

Le point chaud reste possible

Même avec un hachage parfait, une clé unique très demandée sature une machine : c’est le compte d’une célébrité sur un réseau social. Le hachage répartit les clés, pas la charge par clé.

À retenir

Réplication : la même donnée à plusieurs endroits. Sert la disponibilité et les lectures.

Partitionnement : des données différentes à des endroits différents. Sert le volume et les écritures.

Les vrais systèmes font les deux : chaque partition est répliquée.

5.6 Détecter une panne : un problème difficile

Le heartbeat

La technique universelle : chaque machine envoie périodiquement un signe de vie (heartbeat, battement de cœur). Si l’on n’en reçoit plus pendant un certain temps, on la déclare morte.

Détecteur de pannes simplifié

import time

INTERVALLE = 2.0        # un signal attendu toutes les 2 s
TOLERANCE  = 6.0        # mort si silence pendant 6 s

derniers_signaux = {}   # machine -> instant du dernier signal

def signal_recu(machine):
    derniers_signaux[machine] = time.perf_counter()

def machines_mortes():
    maintenant = time.perf_counter()
    return [m for m, t in derniers_signaux.items()
            if maintenant - t > TOLERANCE]

On mesure ici une durée (depuis combien de temps n’a-t-on rien reçu ?), sur une seule machine : c’est le rôle de time.perf_counter(), comme nous le verrons en Section 5.7.

Le dilemme du délai

Toute la difficulté est de choisir TOLERANCE.

	Délai court (1 s)	Délai long (60 s)
Détection	Rapide	Lente : le système reste bloqué longtemps
Risque	Faux positifs : une machine seulement lente est déclarée morte	On garde longtemps une machine réellement morte
Conséquence	Travail refait inutilement, risque de split brain	Requêtes perdues, utilisateurs qui attendent

Attention

Il n’existe aucun moyen de distinguer, à distance, une machine en panne d’une machine simplement très lente ou d’un réseau coupé. Ce n’est pas un manque d’ingéniosité : c’est un résultat théorique démontré (l’impossibilité du consensus en système asynchrone). Tout ce qu’on peut faire, c’est choisir un compromis et accepter de se tromper parfois.

Les huit idées fausses du calcul distribué

Au milieu des années 1990, des ingénieurs de Sun Microsystems ont dressé la liste des hypothèses que font presque tous les débutants, souvent sans s’en rendre compte, quand ils écrivent un programme qui communique par le réseau (Peter Deutsch en a formulé sept, James Gosling, le créateur de Java, a ajouté la huitième). Elles sont toutes fausses, et chacune provoque une famille de bugs.

Le piège est qu’elles sont presque vraies sur votre propre machine : en 127.0.0.1, le réseau ne perd rien, répond en un dixième de milliseconde et ne change jamais. Un programme testé uniquement ainsi peut donc reposer sur les huit sans que rien ne le révèle, jusqu’au jour où il tourne réellement sur plusieurs machines.

1. « Le réseau est fiable. » En réalité, des messages se perdent, des connexions se coupent au milieu d’un échange, une machine devient injoignable sans prévenir. Conséquence : un programme qui attend une réponse sans limite de temps reste bloqué pour toujours. Parade : un timeout sur chaque opération réseau, et un plan pour l’échec (réessayer, ignorer cette machine, signaler l’erreur). Dans le cours : le settimeout de la séance 2, les machines éteintes de la salle.

2. « La latence est nulle. » Chaque aller-retour coûte du temps : une fraction de milliseconde dans une salle, une centaine de millisecondes entre deux continents. Conséquence : 1000 petites requêtes envoyées l’une après l’autre vers l’autre bout du monde prennent 100 secondes, même si chacune est instantanée côté serveur. Parade : regrouper les requêtes, ou les envoyer en parallèle. Dans le cours : le ThreadPoolExecutor de la séance 2.

3. « La bande passante est infinie. » Un lien à 1 Gbit/s transporte au mieux 125 Mo par seconde. Conséquence : déplacer 1 To de données prend plus de deux heures, quelle que soit la puissance des machines. Parade : déplacer le calcul vers les données plutôt que les données vers le calcul. Dans le cours : c’est l’idée fondatrice de MapReduce (séances 6 et 7), où chaque machine traite les fichiers qu’elle a déjà.

4. « Le réseau est sûr. » N’importe qui sur le chemin peut lire, modifier ou forger des messages. Conséquence : un serveur qui fait confiance à tout ce qu’il reçoit peut être bloqué ou détourné. Parade : chiffrer et authentifier (TLS, SSH), et valider chaque donnée reçue. Dans le cours : les clients malveillants de l’exercice 3.5, le garde-fou TAILLE_MAX, l’interdiction de pickle sur le réseau.

5. « La topologie ne change pas. » Des machines sont ajoutées, retirées, redémarrées, changent d’adresse. Conséquence : une liste de machines écrite en dur dans le code est fausse au bout de quelques semaines. Parade : obtenir la liste au démarrage auprès d’un annuaire, et supporter qu’elle change en cours de route. Dans le cours : lister_machines.py en séance 2.

6. « Il y a un seul administrateur. » Le réseau traverse plusieurs équipes, plusieurs pare-feux, plusieurs politiques de sécurité, que vous ne maîtrisez pas. Conséquence : un programme qui marche chez vous échoue ailleurs pour une raison qui n’a rien à voir avec votre code. Dans le cours : l’inventaire des machines, accessible depuis campus-telecom mais pas depuis eduroam.

7. « Le transport ne coûte rien. » Transformer un objet en octets (sérialiser) consomme du processeur, et faire sortir des données d’un hébergeur cloud se paie souvent au gigaoctet. Conséquence : un format bavard ou des envois inutiles coûtent du temps et de l’argent. Parade : mesurer, et choisir un format adapté au volume. Dans le cours : le choix entre texte et binaire en séance 3.

8. « Le réseau est homogène. » Les machines n’ont pas toutes le même système, la même version de Python, le même ordre des octets, le même encodage de texte. Conséquence : un message bien compris par l’une est mal lu par l’autre. Parade : tout fixer explicitement dans le protocole. Dans le cours : le ! de struct (ordre réseau), l’encodage UTF-8 imposé, et os.getloadavg() absent sous Windows en séance 1.

À retenir

Ces huit phrases sont les hypothèses que votre code fait sans le dire. Avant de considérer un programme réseau comme terminé, relisez-les une par une et demandez-vous : que se passe-t-il si celle-ci est fausse ?

5.7 Les horloges mentent

Une dernière hypothèse implicite mérite son propre traitement, parce qu’elle produit des bugs particulièrement déroutants : croire que les machines sont d’accord sur l’heure.

Deux horloges, deux usages

Toute machine expose en réalité deux horloges différentes.

	Horloge murale	Horloge monotone
En Python	`time.time()`	`time.perf_counter()`
Donne	une date absolue	un temps écoulé depuis un instant arbitraire
Peut reculer	oui (resynchronisation NTP)	jamais
Comparable entre machines	mal	pas du tout
Sert à	dater un événement	chronométrer une durée

Attention

Une durée mesurée avec time.time() peut être négative. Il suffit qu’une resynchronisation NTP recule l’horloge entre les deux appels. C’est rare, donc le bug survit des mois avant de se manifester — souvent en production, sur la machine la plus sollicitée. Chronométrez toujours avec perf_counter(), comme nous le faisons depuis le début du cours.

Pourquoi on ne peut pas trier par horodatage

Deux machines synchronisées par NTP conservent un écart résiduel de quelques millisecondes en conditions favorables, et bien davantage si l’une est chargée, virtualisée, ou vient de démarrer.

Concrètement : le maître envoie une requête à 12 h 00 min 00,010 s à sa montre ; le travailleur, dont la montre retarde de 30 ms, journalise sa réponse à 12 h 00 min 00,005 s. En fusionnant les deux journaux et en triant par horodatage, la réponse précède la question.

À retenir

Des horodatages issus de machines différentes ne permettent pas de reconstituer l’ordre des événements. Pour ordonner de façon fiable, il faut une source unique : un compteur attribué par le maître, un numéro de séquence par connexion, ou un identifiant de tâche. C’est ce que nous ferons en séance 7 avec id_map.

Le cas célèbre

Certaines bases de données distribuées résolvent les conflits d’écriture par la règle « la dernière écriture gagne », en comparant des horodatages produits par des machines différentes. Si l’horloge d’un nœud avance de quelques secondes, ses écritures écrasent systématiquement celles des autres — et des données disparaissent silencieusement. Voir Kleppmann & Riccomini, Designing Data-Intensive Applications, chapitre 9 (Unreliable Clocks).

5.8 Le théorème CAP : ce qu’il dit vraiment

Les trois lettres

Le théorème CAP (Eric Brewer, 2000, démontré en 2002) concerne une donnée répliquée sur plusieurs machines. Il porte sur trois propriétés, une par lettre.

C comme Consistency, la cohérence. Toute lecture renvoie la dernière valeur écrite, quelle que soit la machine interrogée. Vu de l’extérieur, tout se passe comme s’il n’existait qu’une seule copie de la donnée. Si Alice vient d’acheter le dernier exemplaire, Bob, qui interroge une autre machine juste après, voit le stock à zéro.
A comme Availability, la disponibilité. Toute machine en fonctionnement répond à toute requête, par une vraie réponse et non par une erreur ou une attente sans fin.
P comme Partition tolerance, la tolérance aux partitions. Le système continue de fonctionner quand le réseau se coupe et que les machines se retrouvent réparties en groupes qui ne peuvent plus se parler.

Deux sens du mot « partition »

Ici, une partition réseau est une panne : une coupure qui sépare les machines en groupes isolés. Cela n’a rien à voir avec le partitionnement des données vu plus haut, qui est un choix de découpage. Les deux mots se ressemblent, les deux notions sont sans rapport.

Le dilemme, sur un exemple

Une boutique en ligne garde son stock sur deux machines, une à Paris et une à Lyon, qui se recopient l’une l’autre. Il reste un exemplaire d’un livre. Soudain, le lien entre Paris et Lyon est coupé. Les deux machines fonctionnent toujours, mais ne peuvent plus se parler.

Un client lyonnais veut acheter le livre. La machine de Lyon ne peut pas savoir si quelqu’un vient de l’acheter à Paris. Elle n’a que deux possibilités, et aucune n’est parfaite :

	Refuser (choix de la cohérence)	Accepter (choix de la disponibilité)
Ce que voit le client	« Service momentanément indisponible »	« Commande confirmée »
Ce qu’on perd	La disponibilité : un client est refoulé	La cohérence : le livre a peut-être aussi été vendu à Paris
Ce qu’il faut faire ensuite	Rien : les données sont restées justes	Réconcilier à la reprise : annuler une des deux ventes, s’excuser
On l’appelle	un système CP	un système AP

Il n’existe pas de troisième voie : répondre « oui » tout en garantissant que le livre n’a pas été vendu ailleurs exigerait de pouvoir demander à Paris, ce qui est justement impossible. C’est tout le contenu du théorème.

La formule « deux sur trois » est trompeuse

On résume souvent CAP par « choisissez deux propriétés parmi les trois ». Ce résumé laisse croire qu’on pourrait choisir « C et A » en renonçant à P. Or on ne choisit pas d’avoir des coupures réseau : elles arrivent, que l’on le veuille ou non. Renoncer à P n’a donc pas de sens pour un système réparti sur plusieurs machines.

La bonne lecture est la suivante :

en temps normal, quand le réseau fonctionne, rien n’empêche d’avoir à la fois la cohérence et la disponibilité ;
pendant une coupure, et seulement pendant celle-ci, il faut choisir ce que l’on sacrifie : la cohérence ou la disponibilité.

Qui choisit quoi ?

Le bon choix dépend du coût d’une erreur.

On choisit la cohérence (CP) quand une réponse fausse coûte plus cher qu’une absence de réponse : un solde bancaire, le dernier exemplaire d’un stock, la réservation d’un siège, l’élection d’un leader. C’est le choix de ZooKeeper et d’etcd, et c’est exactement le mécanisme de la majorité vu plus haut : le groupe minoritaire cesse de répondre.
On choisit la disponibilité (AP) quand une réponse légèrement périmée est acceptable : un panier d’achat, un compteur de « j’aime », un fil d’actualité, l’annuaire DNS. Des bases comme Cassandra ou DynamoDB sont conçues pour ce choix, et laissent régler le niveau de cohérence requête par requête.

À retenir

CAP ne dit pas « choisissez deux lettres sur trois ». Il dit : quand le réseau se coupe, chaque machine isolée doit choisir entre refuser de répondre (garder la cohérence) et répondre avec des données peut-être périmées (garder la disponibilité). Le reste du temps, on peut avoir les deux.

5.9 Où retrouve-t-on tout cela ?

Les mécanismes de cette séance ne sont pas théoriques : ils sont au cœur des deux outils que vous rencontrerez le plus souvent, HDFS et Kafka. Voici de quoi les comprendre, sans entrer dans leur configuration.

HDFS, le système de fichiers de Hadoop

HDFS (Hadoop Distributed File System) stocke de très gros fichiers sur des dizaines ou des centaines de machines ordinaires, en les présentant comme un seul système de fichiers. Son fonctionnement tient en quatre idées.

Les fichiers sont découpés en blocs de 128 Mo. Un fichier de 1 Go devient 8 blocs, répartis sur des machines différentes : c’est du partitionnement.
Chaque bloc est copié trois fois, sur trois machines différentes, de préférence dans des armoires (racks) différentes : c’est de la réplication. La perte d’une machine, et même d’une armoire entière, ne perd aucune donnée.
Deux rôles bien séparés.
- Les DataNodes (nœuds de données) sont les nombreuses machines qui stockent les blocs sur leurs disques. Elles ne savent rien des fichiers : elles conservent des blocs numérotés.
- Le NameNode (nœud des noms) est une machine unique qui connaît l’annuaire : quels fichiers existent, de quels blocs ils sont faits, et sur quels DataNodes se trouve chaque copie. Il ne stocke aucune donnée lui-même.
Pour lire un fichier, un programme demande au NameNode où sont les blocs, puis va les lire directement sur les DataNodes. Le NameNode ne voit donc passer que des questions courtes, jamais les données.
Les DataNodes envoient un signe de vie au NameNode toutes les 3 secondes. Sans nouvelle pendant environ 10 minutes, le NameNode déclare la machine morte et fait recopier ailleurs les blocs qu’elle hébergeait, pour revenir à trois copies.

Ce délai de 10 minutes illustre exactement le dilemme vu plus haut : recopier les blocs d’un disque plein coûte très cher, donc HDFS préfère attendre longtemps plutôt que de déclencher ce travail pour une machine qui redémarre simplement. Et le NameNode, unique, serait un point de panne : en production, on en déploie deux, un actif et un en attente, et c’est ZooKeeper qui désigne l’actif pour éviter le split brain.

Kafka, le journal distribué

Kafka transporte des flux de messages entre programmes : des clics sur un site, des commandes, des mesures de capteurs. Des programmes producteurs y écrivent des messages, des programmes consommateurs les lisent, à leur rythme. Les messages ne sont pas effacés après lecture : ils restent stockés, dans l’ordre, pendant une durée configurée.

Un topic (sujet) est un flux nommé de messages d’une même nature : le topic commandes, le topic clics. C’est l’équivalent d’une table, ou d’un dossier.
Chaque topic est découpé en partitions, réparties sur plusieurs serveurs Kafka (appelés brokers). C’est le partitionnement : il permet d’écrire et de lire en parallèle sur plusieurs machines.
Un message peut porter une clé. Kafka choisit sa partition en calculant un hachage de cette clé modulo le nombre de partitions : exactement hash(cle) % N. Tous les messages de même clé vont donc dans la même partition, et Kafka garantit l’ordre à l’intérieur d’une partition, pas entre partitions.
Chaque partition est répliquée, en général en 3 exemplaires, selon le modèle leader / suiveurs : un broker est le leader de la partition et reçoit toutes les écritures, les autres le suivent.
L’ISR (In-Sync Replicas, répliques synchronisées) est la liste des copies qui sont à jour avec le leader, à un petit retard près. Un suiveur trop en retard en est retiré, puis y revient quand il a rattrapé. L’ISR sert à deux choses. Un producteur peut exiger qu’un message ne soit confirmé qu’une fois reçu par toutes les copies de l’ISR. Et si le leader tombe, le nouveau leader est choisi dans l’ISR, donc parmi les copies qui ont tous les messages confirmés. C’est la réponse de Kafka à la question « quel suiveur choisir ? » posée plus haut.

Comme nous l’avons vu, changer le nombre de partitions changerait la partition de presque toutes les clés. C’est pourquoi on fixe ce nombre à la création du topic, en prévoyant large, et qu’on le change rarement.

Bien choisir une clé de partitionnement

Que ce soit dans Kafka, dans une base distribuée (Cassandra, MongoDB, HBase) ou dans Spark, il faut un jour choisir la clé de partitionnement : le champ dont la valeur décide sur quelle machine va chaque donnée. C’est l’une des décisions les plus lourdes de conséquences, parce qu’elle est très difficile à changer après coup.

Une bonne clé vérifie trois critères :

Beaucoup de valeurs différentes, bien plus que de partitions. Un champ « pays » ou « oui/non » ne peut pas répartir sur 100 machines.
Une charge répartie : aucune valeur ne doit concentrer à elle seule une grande part des requêtes.
Ce qui doit rester ensemble reste ensemble : les données qu’on lit d’un coup, ou dont l’ordre compte, partagent la même clé et donc la même partition.

Clé	Verdict	Pourquoi
Identifiant de client	Bonne en général	Des millions de valeurs ; tous les événements d’un client restent dans l’ordre
Identifiant de commande	Bonne	Très nombreuses, charge uniforme
Date du jour	Mauvaise	Toutes les écritures du jour vont sur une seule partition : point chaud
Pays	Mauvaise	Trop peu de valeurs, et un pays domine souvent les autres
Identifiant de vendeur	À surveiller	Un très gros vendeur peut saturer sa partition

Quand une valeur est trop demandée, la parade classique consiste à lui ajouter un suffixe : vendeur42_0, vendeur42_1, …, vendeur42_9. Ses écritures se répartissent alors sur dix partitions. Le prix à payer : pour lire toutes ses données, il faut interroger les dix.

Récapitulatif

Concept de la séance	Dans les outils réels
Partitionnement	HDFS découpe les fichiers en blocs de 128 Mo ; Kafka découpe un topic en partitions
Réplication	HDFS garde 3 copies de chaque bloc ; Kafka réplique chaque partition
Leader / suiveurs	Chaque partition Kafka a un leader et des suiveurs, dont les à-jour forment l’ISR
Signe de vie	Les DataNodes signalent leur vie au NameNode toutes les 3 s
Split brain, majorité	ZooKeeper désigne le NameNode actif ; KRaft désigne le contrôleur Kafka
Point chaud	Une clé de partitionnement mal choisie, dans Kafka comme dans Spark
Percentiles	Les tableaux de bord de tout service en production
Stockage en colonnes	Les formats Parquet et ORC, dans Hadoop et Spark
Horloge monotone	Toute mesure de durée sérieuse, y compris la vôtre en séance 7

Pourquoi c’est important

Le cours d’Hadoop que vous suivez en parallèle vous fait manipuler ces mécanismes. Cette séance vous donne les mots pour comprendre pourquoi ils existent. Quand vous configurerez un facteur de réplication de 3 ou un nombre de partitions, vous saurez exactement à quel compromis vous touchez.

5.10 Exercices

Les corrigés se trouvent en fin de séance, section Section 5.11.

Exercice 5.1 — Percentiles

L’objectif est de voir de vos yeux que la moyenne cache les requêtes lentes, et que le parallélisme les rend plus fréquentes.

Sur une machine récente, un serveur local répond si vite et si régulièrement que l’on n’observe presque rien, même en chargeant le processeur. On simule donc un serveur qui, comme un vrai service, « hoquette » de temps en temps : il répond immédiatement dans 98 % des cas, et met 200 ms dans les 2 % restants.

Étape 1 — Le serveur. Dans le dossier qui contient déjà votre protocole.py de la séance 3, créez ce fichier, puis lancez-le dans un premier terminal avec python3 serveur_lent.py. Si le serveur de la séance 3 tourne encore, arrêtez-le d’abord (Ctrl+C) : il occupe le même port.

serveur_lent.py

import random
import socket
import threading
import time
from protocole import envoyer_message, recevoir_message

PORT = 50007

def traiter(requete):
    if requete.get("type") == "PING":
        if random.random() < 0.02:      # 2 % des requetes
            time.sleep(0.2)             # un hoquet de 200 ms
        return {"type": "PONG"}
    return {"type": "ERREUR", "message": "type inconnu"}

def servir_client(connexion, adresse):
    try:
        while True:
            requete = recevoir_message(connexion)
            if requete.get("type") == "QUIT":
                break
            envoyer_message(connexion, traiter(requete))
    except (ConnectionError, ValueError, OSError):
        pass
    finally:
        connexion.close()

serveur = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serveur.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serveur.bind(("0.0.0.0", PORT))
serveur.listen(64)
print(f"Serveur lent en ecoute sur le port {PORT}")
while True:
    connexion, adresse = serveur.accept()
    threading.Thread(target=servir_client, args=(connexion, adresse),
                     daemon=True).start()

Étape 2 — Des requêtes isolées. Dans un second terminal, écrivez stats_latence.py qui :

ouvre une connexion vers 127.0.0.1:50007 ;
envoie 1000 fois {"type": "PING"} sur cette connexion, et mesure la durée de chaque aller-retour avec time.perf_counter(), en millisecondes ;
envoie {"type": "QUIT"} et ferme la connexion ;
affiche la moyenne, le p50, le p95, le p99 et le maximum, avec la fonction percentile du cours.

Répondez ensuite : (a) une seule des 1000 requêtes a-t-elle réellement pris le temps de la moyenne ? (b) Lequel des cinq indicateurs révèle les hoquets du serveur, et lequel les cache complètement ?

Étape 3 — Une page qui attend 20 réponses. Une page web appelle souvent plusieurs services à la fois et ne s’affiche qu’une fois tous les résultats arrivés. Simulez-le : ajoutez à votre programme une fonction une_page() qui envoie 20 PING en parallèle (un ThreadPoolExecutor de 20 threads, chaque requête sur sa propre connexion) et mesure le temps jusqu’à la dernière réponse. Appelez-la 100 fois et affichez les mêmes indicateurs, puis le nombre de pages qui ont pris plus de 100 ms.

Répondez enfin : (c) quelle proportion de pages est lente ? Comparez avec le calcul de la section « Un point qui surprend », en remplaçant 0,99 par 0,98 puisque 2 % des requêtes sont lentes.

Exercice 5.2 — Équilibrage du partitionnement

L’objectif est de comparer deux façons de répartir des mots sur $N$ partitions, et de comprendre pourquoi l’une équilibre bien et l’autre non.

Le texte. Il vous faut un fichier texte d’au moins quelques milliers de mots. Le plus simple est de télécharger dès maintenant le corpus de la séance 6 avec telecharger_corpus.py (voir Section 6.2) et de prendre corpus_source/miserables.txt. Tout autre long texte convient.

Le programme. Écrivez repartition.py, qui reçoit le nom du fichier en argument (python3 repartition.py corpus_source/miserables.txt) et :

lit le fichier, le met en minuscules et le découpe en mots avec split() ; il ne garde que les mots distincts, avec un set ;
pour N=4N = 4 puis N=10N = 10, compte combien de mots tombent dans chaque partition, selon deux méthodes :
- par hachage : int(hashlib.md5(mot.encode()).hexdigest(), 16) % N ;
- par première lettre : ord(mot[0]) % N ;
affiche pour chaque méthode et chaque $N$ le plus petit et le plus grand nombre de mots par partition, et l’écart relatif, défini par

$\text{écart relatif} = \frac{\text{max} - \text{min}}{\text{moyenne}}$

Par exemple, avec des partitions de 90, 100 et 110 mots, la moyenne est 100 et l’écart relatif $(110 - 90) / 100 = 20\,\%$ . Un écart de 0 % serait un équilibre parfait.

Pour compter, collections.Counter est pratique : Counter(fonction(m, N) for m in mots) donne directement le nombre de mots par numéro de partition.

Questions. Quelle méthode équilibre le mieux ? Pourquoi la seconde échoue-t-elle ? Indice : combien de valeurs différentes peut prendre mot[0], et ces valeurs sont-elles toutes aussi fréquentes en français ?

Exercice 5.3 — Le coût d’un changement de $N$

L’objectif est de mesurer combien de données il faut déplacer quand on ajoute une seule machine, avec deux méthodes de partitionnement.

Partie A — Avec hash % N. Écrivez changement_n.py qui :

fabrique 10 000 clés : cles = [f"cle{i}" for i in range(10000)] ;
calcule, pour chaque clé, sa partition avec $N = 10$ machines, puis avec $N = 11$ , avec la fonction partition du cours (basée sur hashlib) ;
compte les clés dont la partition a changé, et affiche ce nombre en pourcentage.

Avant de lancer le programme, écrivez votre pronostic : avec une machine de plus sur dix, quel pourcentage de clés vous semblerait raisonnable de déplacer ?

Partie B — Avec des partitions fixes. On applique maintenant la méthode de l’encadré « Comment les systèmes réels évitent de tout déplacer » :

chaque clé a une partition fixe parmi 110 : partition(cle, 110) ;
un dictionnaire proprietaire indique quelle machine possède chaque partition. Au départ, avec 10 machines numérotées de 0 à 9, la partition p appartient à la machine p % 10 ;
on ajoute la machine 10 : chacune des dix machines existantes lui cède une de ses partitions (la première qu’elle possède, par exemple) ;
comptez les clés dont la machine a changé, et affichez le pourcentage, ainsi que le nombre de clés par machine après l’ajout.

Questions. Comparez les deux pourcentages. Pourquoi ajouter une machine est-il si coûteux avec hash % N ? Quel pourcentage minimal est inévitable, puisque la nouvelle machine doit bien recevoir sa part des données ?

Exercice 5.4 — Simuler un détecteur de pannes

L’objectif est de mesurer le nombre de fausses alertes d’un détecteur de pannes selon le délai choisi, sur une simulation entièrement locale (aucun réseau, uniquement des threads).

Ce qu’il faut simuler.

5 machines, chacune représentée par un thread. Chaque thread, en boucle, attend un moment puis enregistre l’instant de son signe de vie dans un dictionnaire partagé derniers[nom] = time.perf_counter().
Les machines 0 à 3 sont normales : elles attendent 2 s plus un petit aléa, 2 + random.uniform(0, 0.2).
La machine 4 est lente : elle attend 2 + random.uniform(0, 10), donc entre 2 et 12 secondes. Elle n’est jamais en panne, seulement irrégulière.
Un superviseur (la boucle principale) se réveille toutes les secondes et déclare morte toute machine dont le dernier signe de vie date de plus de TOLERANCE secondes.

Le dictionnaire derniers est lu et écrit par plusieurs threads : il faut un threading.Lock, comme en séance 2.

Ce qu’il faut compter. Aucune machine n’est réellement en panne : toute déclaration de décès est donc une fausse alerte (un faux positif). Comptez une fausse alerte chaque fois qu’une machine passe de l’état « vivante » à l’état « morte ». Attention : une machine qui reste muette 5 secondes sera vue morte par 5 tours successifs du superviseur, mais ne doit compter que pour une alerte. Il faut donc retenir, pour chaque machine, si elle était déjà considérée comme morte au tour précédent.

Ce qu’il faut faire varier. Écrivez une fonction simuler(tolerance) qui fait tourner la simulation pendant 60 secondes et renvoie le nombre de fausses alertes. Appelez-la pour TOLERANCE = 3, 6 et 15 secondes (soit trois minutes en tout). Pour arrêter proprement les threads à la fin de chaque simulation, utilisez un threading.Event.

Questions. Combien de fausses alertes pour chaque valeur ? Quelle valeur choisiriez-vous, et que perdriez-vous en contrepartie si une machine tombait réellement en panne ?

Exercice 5.5 — Réflexion : diagnostiquer une architecture

Un service de vente en ligne subit les quatre symptômes ci-dessous. Pour chacun, répondez en trois temps :

le mécanisme de la séance qui est en cause (un mot ou une expression du cours) ;
l’explication : ce qui se passe, étape par étape, pour produire ce symptôme ;
une correction possible, et ce qu’elle coûte.

Après avoir ajouté un article au panier, l’utilisateur voit parfois un panier vide, puis l’article réapparaît quelques secondes plus tard.
La latence moyenne est de 30 ms, mais le service client reçoit des plaintes de lenteur tous les jours.
Une seule machine sur les vingt est à 100 % de CPU, les autres à 5 %. Les données sont partitionnées par identifiant de vendeur.
Lors d’une micro-coupure réseau de 4 secondes, le système a accepté deux commandes contradictoires sur le dernier exemplaire d’un produit.

Exercice 5.6 — Majorité, mandats et choix CAP

Exercice sur papier, sans programme.

Une grappe de 5 machines est coupée par une panne réseau en deux groupes : 3 machines d’un côté, 2 de l’autre. Quel groupe peut élire un leader ? Que doit faire l’autre groupe ? Même question si la panne coupe la grappe en trois groupes de 2, 2 et 1 machines.
Pourquoi une grappe de 4 machines ne tolère-t-elle pas plus de pannes qu’une grappe de 3 ? Que se passe-t-il si une coupure sépare 4 machines en deux groupes de 2 ?
L’ancien leader, élu au mandat 12, revient après une coupure et envoie une écriture. Entre-temps, un nouveau leader a été élu au mandat 13. Que font les autres machines de cette écriture, et pourquoi ?
Pour chacun de ces services, préféreriez-vous qu’il reste cohérent (CP) ou disponible (AP) pendant une coupure réseau ? Justifiez en une phrase, en disant ce que coûterait une erreur.
1. la réservation des places dans un train ;
2. le compteur de vues d’une vidéo ;
3. le solde d’un compte bancaire ;
4. le panier d’un site de vente ;
5. le choix du leader d’une grappe de bases de données.

5.11 Corrigés

Exercice 5.1 — Percentiles

stats_latence.py

import socket
import time
from concurrent.futures import ThreadPoolExecutor
from protocole import envoyer_message, recevoir_message

ADRESSE = ("127.0.0.1", 50007)

def percentile(valeurs, p):
    triees = sorted(valeurs)
    return triees[int(round((p / 100.0) * (len(triees) - 1)))]

def afficher(titre, mesures):
    print(f"--- {titre} ({len(mesures)} mesures) ---")
    print(f"moyenne : {sum(mesures) / len(mesures):8.2f} ms")
    for p in (50, 95, 99):
        print(f"p{p:<6d}: {percentile(mesures, p):8.2f} ms")
    print(f"max     : {max(mesures):8.2f} ms")

def un_ping(s):
    """Un aller-retour PING/PONG sur le socket s, en millisecondes."""
    debut = time.perf_counter()
    envoyer_message(s, {"type": "PING"})
    recevoir_message(s)
    return (time.perf_counter() - debut) * 1000

# Etape 2 : 1000 requetes l'une apres l'autre, sur une seule connexion
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(5.0)
s.connect(ADRESSE)
mesures = [un_ping(s) for _ in range(1000)]
envoyer_message(s, {"type": "QUIT"})
s.close()
afficher("requetes isolees", mesures)

# Etape 3 : une page envoie 20 requetes en parallele et attend les 20
def ping_isole(_):
    with socket.create_connection(ADRESSE, timeout=5.0) as c:
        duree = un_ping(c)
        envoyer_message(c, {"type": "QUIT"})
        return duree

def une_page(pool):
    debut = time.perf_counter()
    list(pool.map(ping_isole, range(20)))   # attend les 20 reponses
    return (time.perf_counter() - debut) * 1000

with ThreadPoolExecutor(max_workers=20) as pool:
    pages = [une_page(pool) for _ in range(100)]
afficher("pages de 20 requetes paralleles", pages)
lentes = sum(1 for t in pages if t > 100)
print(f"pages de plus de 100 ms : {lentes} sur {len(pages)}")

Comment le programme mesure. un_ping lit l’horloge juste avant l’envoi et juste après la réception : la différence est la durée de cet aller-retour. On multiplie par 1000 pour passer des secondes aux millisecondes. une_page mesure de la même façon, mais autour de pool.map, qui ne rend la main qu’une fois les 20 requêtes terminées : on mesure donc bien le temps jusqu’à la dernière réponse.

Un résultat typique (les valeurs exactes changent à chaque exécution) :

--- requetes isolees (1000 mesures) ---
moyenne :     5.16 ms
p50    :     0.08 ms
p95    :     0.15 ms
p99    :   203.56 ms
max     :   205.40 ms
--- pages de 20 requetes paralleles (100 mesures) ---
moyenne :    78.79 ms
p50    :     5.02 ms
p95    :   207.93 ms
p99    :   210.38 ms
max     :   220.29 ms
pages de plus de 100 ms : 37 sur 100

(a) Aucune requête n’a pris le temps de la moyenne. Les requêtes prennent soit environ 0,1 ms (les 98 % normales), soit environ 200 ms (les 2 % de hoquets). La moyenne, entre 4 et 5 ms selon les exécutions, est un mélange des deux qui ne décrit aucune requête réelle. Le calcul le confirme : $0{,}98 \times 0{,}1 + 0{,}02 \times 200 \approx 4$ ms.

(b) Le p99 révèle les hoquets, le p50 et le p95 les cachent. Puisque 2 % des requêtes sont lentes, les 98 % les plus rapides sont toutes normales : p50 et p95 tombent dans cette zone et affichent 0,1 ms, comme si tout allait bien. Le p99, lui, regarde la requête classée au rang 99 % : elle fait partie des 2 % lentes, d’où ses 200 ms. Si les hoquets ne concernaient que 0,5 % des requêtes, il faudrait regarder le p99,9 pour les voir.

(c) Environ un tiers des pages est lent. Une page est rapide seulement si ses 20 requêtes le sont toutes. Chacune l’est avec une probabilité de 0,98, donc les 20 à la fois avec une probabilité de $0{,}98^{20} \approx 0{,}67$ . Une page est donc lente dans $1 - 0{,}67 = 33\,\%$ des cas. On observe entre 25 et 40 pages lentes sur 100 environ selon les exécutions : avec seulement 100 pages, le hasard fait varier ce nombre de quelques unités autour de 33. Un défaut rare pour une requête (2 %) devient fréquent pour l’utilisateur (une page sur trois) : c’est l’amplification de la queue de distribution.

Conclusion. La moyenne masque exactement les cas qui dégradent l’expérience utilisateur. On surveille donc les percentiles hauts, et d’autant plus que les requêtes sont nombreuses à être attendues ensemble.

Exercice 5.2 — Équilibrage

repartition.py

import hashlib
import sys
from collections import Counter

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

def par_lettre(cle, N):
    return ord(cle[0]) % N

texte = open(sys.argv[1], encoding="utf-8", errors="ignore").read()
mots = set(texte.lower().split())        # mots distincts

for N in (4, 10):
    for nom, fonction in [("hachage", par_hachage),
                          ("lettre ", par_lettre)]:
        c = Counter(fonction(m, N) for m in mots if m)
        tailles = [c.get(i, 0) for i in range(N)]   # 0 si partition vide
        ecart = (max(tailles) - min(tailles)) / (sum(tailles) / N)
        print(f"N={N:2d} {nom} : min={min(tailles):6d} "
              f"max={max(tailles):6d} ecart={ecart:5.1%}")

Deux détails du code. c.get(i, 0) compte 0 pour une partition qui n’aurait reçu aucun mot : sans cela, une partition vide n’apparaîtrait pas dans la liste, et le minimum serait faux. Et sum(tailles) / N est la moyenne de mots par partition, qui sert de référence à l’écart relatif.

Résultat mesuré sur miserables.txt (21 000 mots distincts environ) :

N= 4 hachage : min=  5200 max=  5371 ecart= 3.2%
N= 4 lettre  : min=  4790 max=  6480 ecart=32.0%
N=10 hachage : min=  2008 max=  2195 ecart= 8.8%
N=10 lettre  : min=  1145 max=  3369 ecart=105.2%

Lecture. Avec le hachage, les partitions ont presque la même taille : quelques pour cent d’écart. Avec la première lettre, avec $N = 10$ , la partition la plus chargée a trois fois plus de mots que la moins chargée : une machine travaillerait trois fois plus que sa voisine, et tout le monde l’attendrait.

Le partitionnement par première lettre échoue pour deux raisons :

Les initiales ne sont pas équifréquentes dans une langue : beaucoup de mots en c, p, s, très peu en k, w, x.
Il n’y a que 26 valeurs distinctes en entrée. C’est bien trop peu pour lisser les écarts de fréquence : un seau reçoit deux ou trois lettres, et il suffit qu’une seule d’entre elles soit fréquente pour le faire déborder.

Ce n’est pas le modulo qui est en cause

On croit souvent que ord() étant concentré sur des valeurs consécutives, le modulo disperserait mal. C’est faux : des entiers consécutifs modulo $N$ se répartissent au contraire très régulièrement. Vérification sur les 26 lettres :

$N$	lettres par seau
4	6, 6, 7, 7
10	2, 2, 2, 2, 3, 3, 3, 3, 3, 3

Tous les seaux sont utilisés, et équitablement. Le problème n’est pas la fonction de répartition : c’est la donnée d’entrée, trop pauvre et mal distribuée. Un hachage, lui, transforme chaque mot entier en une valeur parmi des milliards, ce qui efface les régularités de la langue.

Exercice 5.3 — Coût d’un changement de N

changement_n.py

import hashlib

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

cles = [f"cle{i}" for i in range(10000)]

# Partie A : hash % N, en passant de 10 a 11 machines
change = sum(1 for c in cles if partition(c, 10) != partition(c, 11))
print(f"A. hash % N : {change} cles sur {len(cles)} changent "
      f"({change / len(cles):.1%})")

# Partie B : 110 partitions fixes, et une table partition -> machine
proprietaire = {p: p % 10 for p in range(110)}
avant = {c: proprietaire[partition(c, 110)] for c in cles}

for m in range(10):           # chaque machine cede une partition a la 10
    p = next(p for p in range(110) if proprietaire[p] == m)
    proprietaire[p] = 10

apres = {c: proprietaire[partition(c, 110)] for c in cles}
deplacees = sum(1 for c in cles if avant[c] != apres[c])
print(f"B. partitions fixes : {deplacees} cles sur {len(cles)} changent "
      f"({deplacees / len(cles):.1%})")
print("   cles par machine :",
      [sum(1 for c in cles if apres[c] == m) for m in range(11)])

Résultat mesuré :

A. hash % N : 9017 cles sur 10000 changent (90.2%)
B. partitions fixes : 983 cles sur 10000 changent (9.8%)
   cles par machine : [908, 879, 894, 885, 922, 922, 912, 930, 920, 845, 983]

Partie A : 90 % des clés déménagent. Une clé ne reste à sa place que si le reste de sa division par 10 et le reste de sa division par 11 sont égaux. Or ces deux restes n’ont aucun lien entre eux : ils ne coïncident que dans environ 1 cas sur 11. Plus précisément, les deux restes ne dépendent que du reste de la division par 110 ; sur ces 110 cas possibles, exactement 10 donnent deux restes égaux. Une clé reste donc en place avec une probabilité de $10/110 = 1/11 \approx 9\,\%$ , et déménage dans $91\,\%$ des cas. On mesure 90,2 %.

Partie B : 10 % seulement. La partition d’une clé ne change jamais : seul le propriétaire de 10 partitions sur 110 change. Ce sont donc environ $10/110 \approx 9\,\%$ des clés qui déménagent, et elles vont toutes vers la nouvelle machine. Les dix anciennes machines ne s’échangent rien entre elles.

Le minimum inévitable. Après l’ajout, chacune des 11 machines doit porter environ 1/11 des données. La nouvelle machine, partie de zéro, doit donc recevoir 1/11 des clés, soit environ 9 % : on ne peut pas faire moins. La partie B atteint presque ce minimum ; la partie A déplace dix fois plus que nécessaire.

Conséquence pratique. Avec hash % N, ajouter une machine à un système qui stocke 10 To impose de recopier 9 To sur le réseau. C’est pourquoi les systèmes réels fixent un grand nombre de partitions dès le départ (Kafka, Elasticsearch) ou utilisent le hachage cohérent.

Exercice 5.4 — Détecteur de pannes

La tolérance devient un paramètre, et l’on rejoue la même simulation pour chacune des trois valeurs demandées.

detecteur.py

import threading, time, random

INTERVALLE = 2.0
DUREE = 60.0                  # duree d'une simulation, en secondes

def simuler(tolerance):
    derniers = {}             # machine -> instant du dernier signe de vie
    deja_mortes = {}          # machine -> deja declaree morte au tour precedent ?
    verrou = threading.Lock()
    fin = threading.Event()
    faux_positifs = 0

    def machine(nom, retard_max):
        while not fin.is_set():
            time.sleep(INTERVALLE + random.uniform(0, retard_max))
            with verrou:
                derniers[nom] = time.perf_counter()

    for i in range(5):
        nom = f"machine{i}"
        derniers[nom] = time.perf_counter()
        deja_mortes[nom] = False
        retard = 10.0 if i == 4 else 0.2    # la machine 4 est lente
        threading.Thread(target=machine, args=(nom, retard),
                         daemon=True).start()

    debut = time.perf_counter()
    while time.perf_counter() - debut < DUREE:      # le superviseur
        time.sleep(1.0)
        maintenant = time.perf_counter()
        with verrou:
            silences = {m: maintenant - t for m, t in derniers.items()}
        for nom, silence in silences.items():
            morte = silence > tolerance
            if morte and not deja_mortes[nom]:
                faux_positifs += 1           # transition vivante -> morte
            deja_mortes[nom] = morte
    fin.set()                                # arrete les threads machines
    return faux_positifs

for tolerance in (3.0, 6.0, 15.0):
    print(f"tolerance {tolerance:5.1f} s : "
          f"{simuler(tolerance)} faux positifs")

Comment le programme est organisé.

Chaque thread machine dort, puis inscrit l’instant présent dans derniers. C’est son signe de vie.
Le superviseur se réveille chaque seconde et calcule, pour chaque machine, depuis combien de temps elle n’a rien envoyé : c’est son silence. Il copie les valeurs sous le verrou, puis travaille sur la copie, pour ne pas bloquer les machines pendant ses calculs.
On mesure des durées sur une seule machine : c’est le rôle de time.perf_counter(), qui ne recule jamais (voir Section 5.7).
fin.set() demande aux threads machines de s’arrêter à la fin de chaque simulation, pour que la suivante reparte de zéro.

Compter des transitions, pas des tours. Le superviseur s’exécute une fois par seconde : une machine silencieuse pendant 9 secondes apparaîtrait dans plusieurs listes successives. Incrémenter le compteur à chaque tour mesurerait donc la durée des erreurs, pas leur nombre. On ne compte une fausse alerte qu’au passage de vivante à morte, d’où le dictionnaire deja_mortes.

Ici aucune machine n’est réellement en panne : toute déclaration de décès est une fausse alerte.

`tolerance`	Fausses alertes sur 60 s (deux exécutions)	Lecture
3 s	8, puis 7	La machine lente est déclarée morte à presque chaque cycle.
6 s	5, puis 6	Encore plus d’un cycle sur deux.
15 s	0, puis 0	Plus aucune erreur, mais une vraie panne met 15 s à se voir.

Pourquoi ces nombres. La machine lente attend entre 2 et 12 secondes, 7 en moyenne : elle envoie donc environ $60 / 7 \approx 8{,}5$ signes de vie par minute. Chaque attente supérieure à la tolérance déclenche une fausse alerte. L’attente vaut $2 + u$ , où $u$ est tiré uniformément entre 0 et 10 :

avec 3 s, il faut $u > 1$ , ce qui arrive 9 fois sur 10 : $8{,}5 \times 0{,}9 \approx 7{,}7$ alertes ;
avec 6 s, il faut $u > 4$ , soit 6 fois sur 10 : $8{,}5 \times 0{,}6 \approx 5$ alertes ;
avec 15 s, il faudrait $u > 13$ , ce qui n’arrive jamais : 0 alerte.

Les machines normales, qui attendent au plus 2,2 s, ne déclenchent jamais rien, même avec 3 s. Les chiffres varient d’une exécution à l’autre, puisque les attentes sont tirées au hasard, mais les ordres de grandeur restent ceux du calcul.

Quelle valeur choisir ? Cela dépend du coût relatif des deux erreurs. Ici, le silence peut atteindre 12 s : toute tolérance inférieure garantit des fausses alertes. Si une fausse alerte coûte cher (on relance un long calcul, ou l’on risque un split brain), on prend 15 s, et l’on accepte de mettre 15 s à détecter une vraie panne. Si l’on veut réagir vite et que le travail refait est bon marché, on prend une valeur plus courte et l’on accepte quelques alertes. La bonne démarche consiste à mesurer le silence réel des machines et à placer la tolérance au-dessus de ses plus grandes valeurs, pas au-dessus de sa moyenne : c’est le même raisonnement qu’avec les percentiles de l’exercice 5.1.

Exercice 5.5 — Diagnostiquer

1. Le panier qui se vide puis se remplit.

Mécanisme : le retard de réplication.
Explication : l’ajout au panier est une écriture, envoyée au leader. La page suivante lit le panier, et cette lecture est servie par un suiveur. Si ce suiveur n’a pas encore reçu la modification, il renvoie l’ancien panier, vide. Quelques secondes plus tard, la copie arrive, et l’article « réapparaît ».
Correction : après une écriture, lire sur le leader pendant quelques secondes, pour cet utilisateur (read-your-writes). Coût : un peu plus de charge sur le leader.

2. Une moyenne correcte, des clients mécontents.

Mécanisme : la moyenne masque la queue de distribution.
Explication : si 99 % des requêtes prennent 10 ms et 1 % prend 2 secondes, la moyenne vaut environ 30 ms et semble excellente. Mais un client qui charge chaque jour des dizaines de pages tombe régulièrement sur les requêtes lentes. Et si une page attend plusieurs services à la fois, la proportion de pages lentes est encore plus grande (exercice 5.1).
Correction : mesurer et surveiller le p99 et le p99,9, chercher la cause des requêtes lentes, et fixer un objectif sur ces percentiles plutôt que sur la moyenne. Coût : il faut conserver toutes les mesures (ou un histogramme), pas seulement une somme.

3. Une machine à 100 %, les autres à 5 %.

Mécanisme : un point chaud, dû à la clé de partitionnement.
Explication : le hachage répartit bien les vendeurs entre les vingt machines, mais pas la charge. Toutes les commandes d’un même vendeur vont sur la même partition. Si un vendeur géant réalise une grande part des ventes, sa machine fait une grande part du travail.
Correction : ajouter un suffixe à la clé de ce vendeur (vendeur42_0 à vendeur42_9), pour étaler ses données sur dix partitions. Coût : pour lire toutes les données de ce vendeur, il faut interroger dix partitions et fusionner les résultats. Autre piste : choisir une autre clé, comme l’identifiant de commande, si l’on n’a pas besoin de lire ensemble toutes les commandes d’un vendeur.

4. Le dernier exemplaire vendu deux fois.

Mécanisme : un split brain pendant une partition réseau, ce qui revient à un choix AP au sens du théorème CAP.
Explication : la coupure a séparé les machines en deux groupes. Chaque groupe, ne voyant plus l’autre, a continué à accepter des commandes en pensant être le seul. Chacun a vu le stock à 1, et chacun l’a vendu.
Correction : pour le stock, il faut choisir la cohérence. On exige une majorité des machines pour accepter une commande : pendant la coupure, seul le groupe majoritaire continue, l’autre refuse les commandes. On ajoute un numéro de mandat pour que l’ancien leader, s’il revient, voie ses écritures rejetées. Coût : pendant la coupure, les clients du côté minoritaire reçoivent une erreur. Pour un dernier exemplaire, c’est préférable à une vente impossible à honorer.

Exercice 5.6 — Majorité, mandats et choix CAP

1. Grappe de 5 machines. La majorité est de 3 (plus de la moitié de 5).

Coupure en 3 + 2 : le groupe de 3 a la majorité et peut élire un leader. Le groupe de 2 ne l’a pas : il doit cesser d’accepter des écritures et attendre le retour du réseau. Il peut éventuellement continuer à servir des lectures, en prévenant qu’elles peuvent être périmées.
Coupure en 2 + 2 + 1 : aucun groupe n’atteint 3 machines. Personne ne peut élire de leader, et le système cesse d’accepter des écritures partout. C’est le prix de la cohérence : mieux vaut s’arrêter que risquer deux leaders.

2. Pourquoi 4 machines ne font pas mieux que 3. Avec 4 machines, la majorité est de 3 (plus de la moitié de 4, qui est 2). On ne tolère donc qu’une panne, comme avec 3 machines. Et si une coupure sépare les 4 machines en deux groupes de 2, aucun groupe n’a la majorité : plus personne ne peut écrire, alors qu’avec 3 machines une coupure laisse toujours un groupe majoritaire. La quatrième machine coûte sans rien rapporter, d’où le choix de grappes impaires : 3, 5 ou 7.

3. L’ancien leader du mandat 12. Les autres machines ont vu le mandat 13 : elles refusent l’écriture, qui porte un numéro plus petit. Le refus apprend à l’ancien leader qu’il a été remplacé : il redevient suiveur et se met à jour auprès du leader du mandat 13. C’est exactement ce qui empêche le split brain de produire des écritures contradictoires.

4. Cohérence ou disponibilité ?

Service	Choix	Pourquoi
a) Réservation de places de train	CP	Vendre deux fois le même siège oblige à refouler un voyageur muni d’un billet : bien pire qu’un « réessayez dans une minute ».
b) Compteur de vues	AP	Un compteur légèrement faux pendant quelques secondes ne gêne personne ; une page qui ne s’affiche pas, si.
c) Solde bancaire	CP	Un solde périmé permet de dépenser deux fois le même argent.
d) Panier d’achat	AP	On préfère laisser le client remplir son panier ; on réconcilie au moment du paiement, qui lui est cohérent.
e) Choix du leader	CP	Deux leaders, c’est précisément le split brain : c’est pour cette décision que ZooKeeper et etcd existent.

Le point d) montre qu’un même site combine les deux choix : le panier est AP, le paiement et le stock sont CP. On choisit service par service, pas pour tout le système d’un coup.

5.12 Ce qu’il faut retenir

L’essentiel de la séance

Deux mondes : opérationnel (peu de lignes par clé, faible latence) et analytique (balayage massif, fort débit, stockage en colonnes). MapReduce appartient au second.
On mesure en percentiles (p50, p95, p99), jamais en moyenne.
Attendre $N$ services en parallèle amplifie fortement la queue de distribution.
Répliquer = copies de la même donnée (disponibilité, lectures). Coût : le retard de réplication.
Split brain = deux leaders à la fois, après une coupure. On l’empêche avec une majorité (grappes de 3 ou 5 machines) et des numéros de mandat ; en pratique, on délègue à ZooKeeper ou etcd.
Partitionner = données différentes sur machines différentes (volume, écritures). Par plage ou par hachage.
hash() de Python n’est pas déterministe entre exécutions : utiliser hashlib.
Changer $N$ dans hash % N redistribue presque tout ; un grand nombre de partitions fixes ne déplace que le strict nécessaire.
Une bonne clé de partitionnement a beaucoup de valeurs, une charge répartie, et regroupe ce qui se lit ensemble.
On ne peut pas distinguer « mort » de « lent ». Le choix du timeout est un compromis, jamais une solution.
Les huit idées fausses : le réseau n’est ni fiable, ni instantané, ni illimité, ni sûr, ni stable, ni administré par une seule personne, ni gratuit, ni homogène.
Les horloges mentent : time.time() pour dater, perf_counter() pour chronométrer. On n’ordonne jamais des événements de machines différentes par horodatage.
CAP : quand le réseau se coupe, il faut sacrifier la cohérence ou la disponibilité. On choisit service par service, selon le coût d’une erreur.
HDFS : fichiers en blocs de 128 Mo copiés 3 fois sur les DataNodes, annuaire sur le NameNode. Kafka : topics découpés en partitions répliquées, nouveau leader choisi dans l’ISR.