Le partitionnement (sharding)

La réplication, sujet du chapitre précédent, consiste à conserver plusieurs copies des mêmes données sur des nœuds différents. Pour de très grands jeux de données, ou pour un débit de requêtes très élevé, cela ne suffit pas : il faut aussi découper les données en partitions. Le mot partition a beaucoup de synonymes selon les produits — shard chez MongoDB, Elasticsearch et SolrCloud, region chez HBase, tablet chez BigTable, vnode chez Cassandra et Riak, vBucket chez Couchbase — mais le terme partitioning reste le plus établi. À ne pas confondre, au passage, avec les partitions réseau (netsplits), qui sont des pannes et seront traitées plus loin dans le livre.

L'idée fondatrice est simple : chaque enregistrement appartient à exactement une partition, et chaque partition est en réalité une petite base de données à elle seule, même si le système sait parfois exécuter des opérations qui touchent plusieurs partitions à la fois. Ce chapitre déroule, dans un ordre pédagogique, comment choisir la clé de répartition, comment composer le partitionnement avec les index secondaires, comment rééquilibrer un cluster, et enfin comment router une requête vers la bonne partition.

Pourquoi partitionner : la scalabilité

La raison principale de partitionner est la scalabilité. Dans une architecture sans partage (shared-nothing), des partitions différentes peuvent vivre sur des nœuds différents. Un grand jeu de données est ainsi réparti sur de nombreux disques, et la charge de requêtes répartie sur de nombreux processeurs. Pour les requêtes qui ne touchent qu'une seule partition, chaque nœud traite indépendamment les requêtes de sa partition : le débit augmente alors en ajoutant des nœuds. Les requêtes complexes, elles, peuvent en principe être parallélisées sur plusieurs nœuds, mais cela devient nettement plus difficile.

L'objectif est de répartir données et charge équitablement. Si dix nœuds prennent chacun leur juste part, ils devraient en théorie absorber dix fois plus de données et dix fois le débit d'un seul nœud. Quand la répartition est injuste — certaines partitions concentrant plus de données ou de requêtes que les autres — on parle de déséquilibre (skew). Dans le cas extrême, toute la charge atterrit sur une partition : neuf nœuds sur dix sont inactifs et le goulot d'étranglement est l'unique nœud occupé. Une partition à charge disproportionnée est un point chaud (hot spot).

On pourrait éviter les points chauds en assignant les enregistrements aux nœuds au hasard. La répartition serait équilibrée, mais avec un défaut rédhibitoire : pour relire un élément précis, on ignore sur quel nœud il se trouve, et il faut interroger tous les nœuds en parallèle. On peut faire bien mieux en supposant un modèle clé-valeur, où l'on accède toujours à un enregistrement par sa clé primaire — comme on cherche une entrée dans une encyclopédie papier par son titre, les entrées étant triées alphabétiquement.

Partitionner les données clé-valeur

Deux grandes stratégies s'opposent pour décider quelle clé va dans quelle partition : par intervalle de clé, ou par hachage de clé.

Par intervalle de clé

Le partitionnement par intervalle de clé (key range) assigne à chaque partition une plage continue de clés, du minimum au maximum, comme les volumes d'une encyclopédie papier. Connaissant les bornes entre plages, on détermine immédiatement quelle partition contient une clé donnée ; connaissant l'affectation des partitions aux nœuds, on adresse la requête directement au bon nœud.

Les plages ne sont pas forcément de taille égale, car les données ne sont pas réparties uniformément : un volume « A–B » peut être bien plus chargé qu'un volume « T–Z ». Pour équilibrer, les bornes doivent s'adapter aux données ; elles sont choisies manuellement par un administrateur, ou automatiquement par la base. Cette stratégie est utilisée par BigTable, son équivalent libre HBase, RethinkDB, et MongoDB avant la version 2.4.

À l'intérieur d'une partition, les clés restent triées. Avantage majeur : les scans de plage (range scans) sont efficaces, et l'on peut traiter la clé comme un index concaténé pour récupérer plusieurs enregistrements liés en une requête. Pour un réseau de capteurs dont la clé est l'horodatage de la mesure (année-mois-jour-heure-minute-seconde), on récupère ainsi facilement toutes les lectures d'un mois donné.

La parade consiste à ne pas mettre l'horodatage en premier élément de la clé. En préfixant par exemple chaque horodatage par le nom du capteur, on partitionne d'abord par capteur puis par temps : avec beaucoup de capteurs actifs simultanément, la charge d'écriture se répartit. En contrepartie, pour récupérer plusieurs capteurs sur une fenêtre de temps, il faut désormais une requête de plage distincte par capteur.

Par hachage de clé

À cause de ce risque de déséquilibre, beaucoup de bases distribuées utilisent une fonction de hachage (hash function) pour déterminer la partition. Une bonne fonction de hachage transforme des données déséquilibrées en distribution uniforme : même des chaînes très proches en entrée produisent des hachages bien répartis. Inutile qu'elle soit cryptographique — Cassandra et MongoDB utilisent MD5, Voldemort la fonction Fowler–Noll–Vo.

On assigne alors à chaque partition une plage de hachages (et non de clés) : toute clé dont le hachage tombe dans la plage d'une partition y est stockée. Les bornes peuvent être réparties uniformément ou choisies pseudo-aléatoirement.

Le hachage fait toutefois perdre les scans de plage efficaces : des clés autrefois adjacentes sont éparpillées sur toutes les partitions, l'ordre de tri disparaît. En mode haché, MongoDB doit envoyer toute requête de plage à toutes les partitions ; Riak, Couchbase et Voldemort ne supportent pas du tout les requêtes de plage sur la clé primaire.

Cassandra propose un compromis élégant. On déclare une clé primaire composée de plusieurs colonnes : seule la première partie est hachée pour déterminer la partition, les autres servant d'index concaténé pour trier les données dans les SSTables. On ne peut donc pas chercher une plage de valeurs sur la première colonne, mais en fixant sa valeur, on obtient un scan de plage efficace sur les colonnes suivantes. C'est idéal pour les relations un-à-plusieurs : avec une clé (user_id, update_timestamp), on récupère efficacement tous les messages d'un utilisateur sur un intervalle de temps, triés chronologiquement.

Critère	Par intervalle de clé	Par hachage de clé
Répartition de la charge	Inégale, à ajuster aux données	Uniforme par construction
Scans de plage	Efficaces (clés triées)	Inefficaces (ordre perdu)
Risque de point chaud	Élevé sur clés voisines (horodatage)	Réduit, mais non éliminé
Rééquilibrage typique	Dynamique (split de plage)	Nombre fixe de partitions
Exemples	HBase, BigTable, RethinkDB	Cassandra, MongoDB (mode haché), Voldemort
Compromis Cassandra	Clé composée : 1ʳᵉ partie hachée, reste trié

Charges déséquilibrées et points chauds résiduels

Le hachage réduit les points chauds mais ne les élimine pas. Dans le cas extrême où toutes les lectures et écritures visent la même clé, toutes les requêtes finissent sur la même partition : hacher n'y change rien, car deux identifiants identiques ont le même hachage. Ce cas n'est pas qu'une curiosité — sur un réseau social, une célébrité aux millions d'abonnés peut déclencher une tempête d'activité sur une seule clé (son identifiant, ou l'identifiant de l'action commentée).

Aujourd'hui, la plupart des systèmes ne compensent pas automatiquement une charge aussi déséquilibrée : c'est à l'application de la corriger. Technique simple pour une clé connue comme très chaude : lui ajouter un nombre aléatoire en préfixe ou suffixe. Un simple nombre à deux chiffres répartit les écritures sur 100 clés distinctes, donc potentiellement sur des partitions différentes.

Index secondaires et partitionnement

Tout ce qui précède suppose un accès par clé primaire. La situation se complique avec les index secondaires (secondary indexes), qui n'identifient généralement pas un enregistrement de façon unique mais permettent de chercher les occurrences d'une valeur : tous les messages de l'utilisateur 123, tous les articles contenant le mot hogwash, toutes les voitures rouges. Ces index sont le pain quotidien des bases relationnelles, courants dans les bases documentaires, et la raison d'être des moteurs de recherche comme Solr et Elasticsearch. Leur problème : ils ne s'alignent pas proprement sur les partitions. Deux approches existent.

Index local : partitionnement par document

Imaginez un site de vente de voitures d'occasion. Chaque annonce a un identifiant — l'identifiant de document — et l'on partitionne par cet identifiant. Pour permettre la recherche par couleur et par marque, il faut un index secondaire sur color et make. Dans cette approche, chaque partition maintient ses propres index secondaires, couvrant uniquement ses documents, sans rien savoir des autres. À l'écriture (ajout, suppression, modification), on ne touche que la partition contenant le document : on parle d'index local.

Partition 0                          Partition 1
  191 {color: red,    make: Honda}     515 {color: silver, make: Ford}
  214 {color: black,  make: Dodge}     768 {color: red,    make: Volvo}
  306 {color: red,    make: Ford}      893 {color: silver, make: Audi}

  color:red  -> [191, 306]            color:red    -> [768]
  make:Ford  -> [306]                 make:Ford    -> [515]

Requête « voitures rouges » : scatter/gather sur TOUTES les partitions

La lecture, en revanche, demande de la prudence : rien ne garantit que toutes les voitures rouges soient sur la même partition. Pour les trouver, il faut envoyer la requête à toutes les partitions et recombiner les résultats — une approche dite scatter/gather (« disperser/rassembler »). Cela rend les requêtes sur index secondaires coûteuses : même exécutée en parallèle, l'opération souffre de l'amplification de la latence de queue (tail latency amplification). Elle reste néanmoins très répandue : MongoDB, Riak, Cassandra, Elasticsearch, SolrCloud et VoltDB utilisent tous des index secondaires partitionnés par document.

Index global : partitionnement par terme

L'alternative est de construire un index global couvrant toutes les partitions. On ne peut pas le stocker sur un seul nœud — il deviendrait un goulot d'étranglement — il doit donc lui-même être partitionné, mais différemment de l'index par clé primaire. Les voitures rouges de toutes les partitions figurent sous color:red, et cet index est réparti par exemple ainsi : couleurs de a à r sur la partition 0, de s à z sur la partition 1.

Partition 0 (termes a–r)             Partition 1 (termes s–z)
  color:red  -> [191, 306, 768]       color:silver -> [515, 893]
  make:Audi  -> [893]                 make:Volvo   -> [768]
  make:Ford  -> [306, 515]
  make:Honda -> [191]

Requête « voitures rouges » : UNE seule partition interrogée

On parle de partitionnement par terme (term-based), car c'est le terme recherché (ici color:red) qui détermine la partition de l'index. Le nom terme vient des index plein-texte, où les termes sont tous les mots d'un document. On peut partitionner par le terme lui-même — utile pour les scans de plage, sur un prix par exemple — ou par son hachage, pour une répartition plus uniforme.

En pratique, les mises à jour des index secondaires globaux sont souvent asynchrones : si vous lisez l'index juste après une écriture, votre changement peut ne pas encore y figurer. DynamoDB annonce ainsi une propagation « en une fraction de seconde » en temps normal, avec des délais plus longs en cas de panne. Riak (fonction de recherche) et l'entrepôt de données Oracle (au choix entre index local et global) recourent aussi à des index par terme.

Rééquilibrer les partitions

Au fil du temps, le débit augmente (il faut plus de CPU), le volume croît (il faut plus de disque et de RAM), ou une machine tombe (d'autres doivent reprendre sa charge). Tous ces cas imposent de déplacer des données entre nœuds : c'est le rééquilibrage (rebalancing). Quel que soit le schéma, on attend de lui trois propriétés minimales : après coup, la charge est équitablement répartie ; pendant l'opération, la base continue d'accepter lectures et écritures ; et l'on ne déplace pas plus de données que nécessaire pour ne pas saturer le réseau.

Surtout pas : hash mod N

On pourrait croire que hash(clé) mod N (avec N le nombre de nœuds) suffit à affecter chaque clé à un nœud. C'est un piège : si N change, presque toutes les clés doivent être déplacées.

// hash(clé) % N : pourquoi c'est une mauvaise idée.
// Une même clé saute de nœud à chaque changement de N.
const h = 123456;

h % 10; // => 6  (cluster de 10 nœuds)
h % 11; // => 3  (on passe à 11 : la clé doit migrer)
h % 12; // => 0  (on passe à 12 : elle migre encore)

Ce déplacement massif rend le rééquilibrage excessivement coûteux. Il faut une approche qui ne bouge pas les données plus que nécessaire.

Nombre fixe de partitions

La solution la plus simple : créer beaucoup plus de partitions que de nœuds, et en assigner plusieurs à chaque nœud. Un cluster de 10 nœuds peut démarrer avec 1 000 partitions, soit environ 100 par nœud. À l'ajout d'un nœud, le nouveau venu « vole » quelques partitions à chaque nœud existant jusqu'à rééquilibre ; au retrait, l'inverse se produit. Seules des partitions entières se déplacent : ni leur nombre, ni l'affectation des clés aux partitions ne changent — seule change l'affectation des partitions aux nœuds. Comme le transfert prend du temps, l'ancienne affectation sert aux lectures et écritures pendant la migration. On peut même tenir compte d'un matériel hétérogène en donnant plus de partitions aux nœuds les plus puissants.

Cette approche est utilisée par Riak, Cassandra (depuis 1.2), Elasticsearch, Couchbase et Voldemort.

Partitionnement dynamique

Un nombre fixe convient bien au hachage, qui répartit uniformément les clés. Mais pour le partitionnement par intervalle, des bornes figées seraient désastreuses : mal placées, elles concentreraient toutes les données dans une partition et videraient les autres. C'est pourquoi HBase et RethinkDB créent les partitions dynamiquement. Quand une partition dépasse une taille configurée (10 Go par défaut sur HBase), elle est scindée en deux, chaque moitié recevant à peu près la moitié des données ; inversement, une partition qui rétrécit trop peut être fusionnée avec une voisine — comme au sommet d'un arbre B (B-tree). Après scission, l'une des moitiés peut être transférée vers un autre nœud pour équilibrer.

Avantage : le nombre de partitions s'adapte au volume. Peu de données, peu de partitions, donc peu de surcoût ; beaucoup de données, taille de chaque partition plafonnée. Réserve : une base vide démarre avec une seule partition, donc toutes les écritures vont sur un seul nœud tant que la première scission n'a pas eu lieu. Pour atténuer cela, HBase et MongoDB permettent de pré-découper (pre-splitting) un ensemble initial de partitions sur une base vide. Le partitionnement dynamique vaut aussi bien pour les données hachées : MongoDB (depuis 2.4) le pratique dans les deux modes.

Proportionnel au nombre de nœuds

Avant la version 1.2, Cassandra utilisait le hachage cohérent à bornes pseudo-aléatoires de Karger et al., avec une grosse partition par nœud. Cette approche souffrait d'une mauvaise répartition et rendait l'ajout de nœuds difficile (un nœud devait scinder sa plage pour en céder la moitié, opération coûteuse en arrière-plan). Elle a été remplacée par l'approche à nombre fixe de partitions. En pratique, les modèles les plus répandus sont donc soit le hachage avec un nombre fixe de partitions, soit le partitionnement dynamique par intervalle (quand les requêtes de plage sont requises).

Automatique ou manuel ?

Reste une question : le rééquilibrage est-il automatique ou manuel ? Il existe un dégradé entre les deux. Couchbase, Riak et Voldemort, par exemple, génèrent automatiquement une affectation suggérée, mais exigent qu'un administrateur la valide avant application.

Le routage des requêtes

Le jeu de données est désormais réparti sur plusieurs machines. Mais une question reste ouverte : quand un client veut faire une requête, comment sait-il à quel nœud se connecter ? L'affectation des partitions aux nœuds changeant à chaque rééquilibrage, quelqu'un doit suivre ces changements pour répondre à : « si je veux lire ou écrire la clé foo, à quelle adresse IP et quel port me connecter ? » C'est un cas particulier d'un problème plus général, la découverte de service (service discovery). Il existe trois grandes approches.

Approche	Qui décide du routage	Mécanique
Nœud passe-plat	N'importe quel nœud contacté	Si le nœud possède la partition, il répond ; sinon il transfère la requête au bon nœud.
Couche de routage	Un niveau de routage dédié	Un répartiteur de charge conscient des partitions aiguille la requête ; il ne traite rien lui-même.
Client averti	Le client lui-même	Le client connaît l'affectation et se connecte directement au bon nœud, sans intermédiaire.

Dans tous les cas, le problème clé est le même : comment le composant qui décide apprend-il les changements d'affectation des partitions aux nœuds ? C'est ardu, car tous les participants doivent être d'accord, sinon les requêtes partent vers les mauvais nœuds. Il existe des protocoles de consensus en système distribué, mais ils sont difficiles à implémenter correctement.

Cassandra et Riak adoptent une voie différente : un protocole de bavardage (gossip protocol) entre nœuds diffuse et fait converger l'état du cluster. Une requête peut aller à n'importe quel nœud, qui la transfère au bon (approche « nœud passe-plat »). Cela complexifie les nœuds mais supprime la dépendance à un service externe comme ZooKeeper. Quant aux adresses IP, elles changent moins vite que l'affectation des partitions : un simple DNS suffit souvent à les retrouver.

Exécution parallèle de requêtes

Jusqu'ici, on n'a vu que des requêtes simples sur une clé (plus le scatter/gather). C'est le niveau d'accès de la plupart des bases NoSQL distribuées. Les bases relationnelles massivement parallèles (MPP), utilisées pour l'analytique, vont bien plus loin : une requête d'entrepôt de données typique comporte plusieurs jointures, filtres, regroupements et agrégations. L'optimiseur MPP la découpe en étapes d'exécution dont beaucoup s'exécutent en parallèle sur différents nœuds — un sujet spécialisé, particulièrement utile aux requêtes balayant de grandes parties du jeu de données.

À retenir

• Partitionner (une partition = un shard) répartit données et charge sur plusieurs machines pour la scalabilité ; chaque partition est une petite base autonome. L'ennemi est le déséquilibre (skew) et son cas extrême, le point chaud (hot spot).
• Par intervalle de clé : clés triées, scans de plage efficaces, mais risque de point chaud (par exemple une clé horodatée). Par hachage de clé : répartition uniforme, mais perte des scans de plage. Cassandra concilie les deux via une clé composée.
• Hacher réduit les points chauds sans les supprimer : une célébrité reste un point chaud résiduel, à corriger côté application (préfixe aléatoire), au prix de lectures plus lourdes.
• Index secondaires : par document (index local, écriture sur une seule partition, lecture en scatter/gather coûteux) vs par terme (index global, écritures multi-partitions plus complexes, lectures ciblées sur une seule partition).
• Rééquilibrage : jamais hash mod N (déplacement massif) ; préférer un nombre fixe de partitions (Riak, Elasticsearch), le partitionnement dynamique (HBase), ou proportionnel aux nœuds (Cassandra). Le mode automatique est risqué avec la détection de panne : un humain dans la boucle évite les cascades.
• Routage : nœud passe-plat, couche de routage dédiée, ou client averti ; le suivi de l'affectation passe par un service de coordination (ZooKeeper) ou un protocole de bavardage (gossip).