Réplication, partitionnement et sharding

Deux raisons différentes de répartir les données

Un serveur de base de données unique finit toujours par manquer de quelque chose : de disponibilité (c’est une seule machine, qui peut tomber en panne), de capacité de lecture, de capacité d’écriture, de stockage, ou de proximité avec des utilisateurs situés sur d’autres continents. Il existe deux façons complémentaires de répartir les données sur plusieurs machines, et il est important de ne pas les confondre. La réplication conserve une copie des mêmes données sur plusieurs nœuds, pour la disponibilité, le passage à l’échelle des lectures et une latence plus faible près des utilisateurs. Le partitionnement, appelé sharding dans de nombreux produits, découpe les données pour que chaque nœud en détienne un sous-ensemble, pour un débit d’écriture et un stockage qui dépassent les capacités d’une seule machine. Les systèmes réels font généralement les deux : chaque partition est répliquée.

Les deux ont de lourdes conséquences sur l’exactitude. La réplication pose la question de ce que voit un lecteur pendant que les copies divergent, ce qui est le sujet des modèles de cohérence, de CAP et de PACELC. Le partitionnement pose la question de ce qui arrive aux requêtes, aux index et aux transactions qui s’étendent sur plusieurs partitions. Cette section explique les mécanismes de façon assez approfondie pour juger les promesses d’une base de données et concevoir une clé de sharding.

Réplication leader-suiveur

Le schéma le plus courant est la réplication leader-suiveur (aussi appelée primaire-réplique ; la documentation plus ancienne dit maître-esclave). Un nœud, le leader, accepte toutes les écritures. Il enregistre chaque modification dans son journal (le journal d’écriture anticipée de PostgreSQL, le journal binaire de MySQL) et transmet ce journal en continu aux suiveurs, qui appliquent les modifications dans le même ordre. Les lectures peuvent aller au leader ou aux suiveurs. La réplication en continu (streaming replication) de PostgreSQL, la réplication MySQL, les groupes de disponibilité de SQL Server, les replica sets de MongoDB et la plupart des bases de données cloud managées fonctionnent ainsi.

Le choix de conception clé est le moment où le leader annonce qu’un commit a réussi. Avec la réplication synchrone, il attend qu’un suiveur confirme avoir reçu la modification, si bien qu’une écriture validée survit à la perte du leader ; le prix est que chaque commit inclut un aller-retour réseau, et que si le suiveur synchrone est hors service, les écritures se bloquent. Avec la réplication asynchrone, le leader confirme immédiatement et transmet la modification ensuite ; les commits sont rapides et les suiveurs peuvent être très éloignés, mais si le leader meurt, les dernières modifications qui n’avaient atteint aucun suiveur sont perdues, alors même que le client a été informé de leur validation. Un compromis courant, appelé semi-synchrone dans MySQL, attend un suiveur et laisse les autres être asynchrones. À la banque, perdre un paiement confirmé est inacceptable, si bien qu’au moins une copie synchrone (ou une base de données fondée sur le consensus) est la norme ; pour la base de données des cartes de la plateforme de livraison de repas, quelques secondes de perte potentielle peuvent être un prix acceptable pour la vitesse. C’est, sous une forme concrète, le compromis sur le RPO vu dans disponibilité et fiabilité.

Le retard de réplication et les anomalies qu’il provoque

Les suiveurs asynchrones n’ont généralement que quelques millisecondes de retard, mais sous charge, pendant une grosse mise à jour par lots ou sur un réseau lent, le retard de réplication peut atteindre des secondes, voire des minutes. Lire sur les suiveurs produit alors des anomalies que les utilisateurs remarquent. Douglas Terry et ses collègues ont défini en 1994 les deux premières garanties ci-dessous, la lecture de ses propres écritures et les lectures monotones, comme des garanties de session ; la troisième compte surtout lorsque les données sont aussi partitionnées.

Aucune de ces anomalies n’existe si toutes les lectures vont au leader ; le remède le plus simple consiste donc souvent à envoyer au leader les lectures qui doivent être fraîches, et à n’utiliser les suiveurs que pour les lectures qui tolèrent des données périmées : listes de produits, rapports, flux de recherche. Concevez cela délibérément, requête par requête, au lieu d’envoyer par défaut toutes les lectures vers les répliques.

Failover et split brain

Le failover (basculement) remplace un leader défaillant par un suiveur. Il comporte trois étapes : détecter que le leader est tombé en panne (généralement par un délai d’expiration), choisir le nouveau leader (idéalement le suiveur le plus à jour, souvent par un protocole de consensus comme Raft entre les nœuds ou par un coordinateur externe comme etcd ou ZooKeeper), et rediriger les écritures vers lui. Chaque étape peut mal tourner. Un délai d’expiration trop court déclenche un failover lors d’une brève coupure réseau ou d’un pic de charge, ce qui aggrave la situation ; un délai trop long prolonge la panne. Avec la réplication asynchrone, les écritures que l’ancien leader n’avait pas encore transmises sont perdues, et si l’ancien leader revient en les détenant toujours, elles entrent en conflit avec les nouvelles écritures.

La défaillance la plus dangereuse est le split brain (cerveau divisé) : deux nœuds se croient tous deux leader et acceptent tous deux des écritures, si bien que les données divergent d’une manière qu’il peut être impossible de fusionner automatiquement. Il survient généralement après une partition réseau dans laquelle l’ancien leader est isolé mais toujours en vie. La protection exige un quorum (un nœud ne peut être leader que tant qu’une majorité est d’accord), le fencing (l’autorité du nouveau leader est imposée par un jeton toujours croissant que le stockage vérifie, ou l’ancien leader est mis hors tension de force, une technique surnommée STONITH : shoot the other node in the head, « abattre l’autre nœud »), et des clients qui suivent de façon fiable le leader actuel.

L’incident de GitHub du 21 octobre 2018 montre à quel point c’est difficile en pratique. Une panne réseau a coupé pendant 43 secondes la liaison entre son point d’interconnexion de la côte Est des États-Unis et son principal centre de données de la côte Est. Son outillage de failover, fonctionnant comme prévu, a promu des bases de données primaires de la côte Ouest. Quand la connectivité est revenue, chaque côte détenait des écritures qui manquaient à l’autre, si bien que GitHub ne pouvait pas simplement rebasculer vers la côte Est ; pour protéger l’intégrité des données, l’entreprise a choisi de restaurer puis de réconcilier, et le service a été dégradé pendant 24 heures et 11 minutes. Le failover automatique entre régions avait transformé une coupure de 43 secondes en une journée entière de rétablissement.

Réplication multi-leader et sans leader

Dans la réplication multi-leader, plusieurs nœuds acceptent des écritures et se répliquent entre eux. On l’utilise pour plusieurs centres de données (chaque région écrit localement, ce qui évite aux utilisateurs une latence transocéanique), pour les clients qui travaillent hors ligne (une app d’agenda ou de notes est en fait un leader sur chaque appareil) et pour l’édition collaborative. Le problème inévitable, ce sont les conflits d’écriture : deux leaders modifient le même enregistrement en même temps. Les résolutions possibles sont last write wins (la dernière écriture l’emporte ; simple, mais qui écarte des données sans avertissement et dépend des horloges), la fusion des valeurs, la conservation des deux en demandant à une personne de trancher, ou l’utilisation de types de données conçus pour fusionner automatiquement (CRDT, conflict-free replicated data types). L’approche la plus fiable est d’éviter les conflits : donner à chaque enregistrement une région d’origine qui est seule à accepter ses écritures, comme le pays de résidence d’un client.

La réplication sans leader, popularisée par l’article Dynamo d’Amazon (DeCandia et ses collègues, 2007) et utilisée dans Apache Cassandra, ScyllaDB et Riak, n’a aucun leader. Le client, ou un nœud coordinateur agissant pour lui, envoie chaque écriture aux N répliques d’une clé et la considère comme réussie quand W d’entre elles l’ont confirmée ; il envoie chaque lecture aux répliques et attend R réponses, en retenant la version la plus récente. Si R + W > N, chaque ensemble de lecture recoupe chaque ensemble d’écriture réussie sur au moins une réplique, si bien qu’une lecture voit normalement la dernière écriture confirmée. Les répliques qui ont manqué des écritures sont réparées par la réparation à la lecture (read repair : le coordinateur réécrit la version la plus récente quand il en voit une périmée) et par une anti-entropie en arrière-plan qui compare les répliques. Cassandra expose ces choix requête par requête sous forme de niveaux de cohérence comme ONE, QUORUM et ALL.

On accorde souvent trop de confiance aux quorums. R + W > N ne rend pas à lui seul le système linéarisable : les écritures concurrentes, les écritures qui ont échoué sur certaines répliques mais réussi sur d’autres, les sloppy quorums (quorums approximatifs, qui écrivent sur des nœuds de remplacement pendant les pannes) et last write wins avec des horloges décalées permettent tous des données périmées ou perdues. Le chapitre 6 détaille l’arithmétique des quorums et ces cas limites dans CAP et PACELC. Les stockages sans leader sont excellents pour une haute disponibilité en écriture malgré les pannes et entre centres de données ; ils conviennent mal aux invariants comme ne jamais vendre deux fois la même place, sauf s’ils proposent une fonction de transaction légère ou de consensus pour ces opérations.

Partitionnement : par plage de clés ou par hachage

Le partitionnement affecte chaque enregistrement à une partition selon sa clé de partition (ou clé de sharding), et l’objectif est de répartir les données et la charge de manière égale. Une partition qui reçoit une charge disproportionnée est un point chaud, et une répartition inégale s’appelle un déséquilibre (skew). Il existe deux schémas de base.

Le partitionnement par plage de clés donne à chaque partition une plage contiguë de clés, comme les volumes d’une encyclopédie. Les parcours de plages sont efficaces parce que les clés adjacentes sont stockées ensemble (toutes les commandes d’octobre dans une ou deux partitions), et les partitions peuvent être scindées quand elles grossissent, comme dans Bigtable, HBase, Spanner et le sharding par plage de MongoDB. Le danger est une clé qui croît avec le temps : si la clé est un horodatage ou un numéro de commande séquentiel, chaque nouvelle écriture tombe sur la dernière partition, qui devient un point chaud pendant que les autres restent inactives. Le partitionnement par hachage applique une fonction de hachage à la clé et affecte des plages de valeurs de hachage aux partitions, ce qui répartit uniformément même les clés séquentielles. Le prix à payer est que les requêtes par plage sur la clé doivent désormais interroger toutes les partitions. De nombreux stockages combinent les deux avec une clé composée : Cassandra hache la clé de partition pour choisir le nœud et trie les lignes au sein de la partition selon des colonnes de clustering, si bien que (customer_id, placed_at) regroupe les commandes d’un client, dans l’ordre chronologique, sur un seul nœud.

Hachage cohérent et rééquilibrage

La façon évidente d’associer les valeurs de hachage aux nœuds, hash(key) mod N, a un défaut rédhibitoire : quand N change, presque toutes les clés se déplacent. Passer de 4 à 5 nœuds déplace environ 80 % des clés (toute clé dont le reste diffère pour 4 et pour 5), et passer de 10 à 11 environ 91 %, ce qui revient à copier presque tout le jeu de données simplement pour ajouter une machine. Le hachage cohérent (consistent hashing), introduit par David Karger et ses collègues en 1997 pour les caches web, place à la fois les nœuds et les clés sur un espace de hachage circulaire ; chaque clé appartient au nœud suivant dans le sens des aiguilles d’une montre. Un nœud ajouté ne reprend que les clés situées entre lui et son prédécesseur, environ 1/N du total, et un nœud retiré cède ses clés à son voisin. Avec peu de nœuds, les arcs sont inégaux ; on attribue donc à chaque nœud physique de nombreux nœuds virtuels (tokens) en différents points de l’anneau, ce qui équilibre la charge et permet à un nouveau nœud de prendre un peu à chaque nœud existant. Dynamo et Cassandra utilisent ce schéma.

Deux autres stratégies de rééquilibrage sont courantes. Avec un nombre fixe de partitions, vous créez au départ beaucoup plus de partitions que de nœuds (par exemple 1024 pour 10 nœuds) et vous affectez des partitions entières aux nœuds ; ajouter un nœud déplace quelques partitions entières et ne rehache jamais les clés. Les shards d’Elasticsearch, les vBuckets de Couchbase et de nombreux schémas de sharding faits maison procèdent ainsi, et la décision principale est de choisir un nombre de partitions assez grand pour la croissance future. Avec le partitionnement dynamique, comme dans HBase et MongoDB, une partition se scinde quand elle dépasse un seuil de taille, et les deux moitiés peuvent aller sur des nœuds différents. Quel que soit le schéma, le rééquilibrage déplace de grandes quantités de données sur le réseau pendant que le système sert du trafic ; de nombreuses équipes préfèrent donc qu’une personne approuve chaque déplacement, et elles le brident pour protéger la latence.

Les requêtes doivent aussi être acheminées vers la bonne partition : soit n’importe quel nœud accepte une requête et la transmet (les nœuds Cassandra apprennent l’anneau par gossip), soit une couche de routage connaît la carte des partitions (mongos de MongoDB), soit la bibliothèque cliente détient la carte. La carte elle-même est généralement conservée dans un service de coordination comme etcd ou ZooKeeper, ou dans le groupe de consensus propre à la base de données.

Clés chaudes et déséquilibre

Le hachage répartit uniformément les clés, pas la charge. Si une clé est bien plus active que les autres, sa partition est chaude quel que soit le schéma : le compte d’une célébrité sur un réseau social, le restaurant phare d’une chaîne nationale pendant une promotion, ou le compte de règlement propre à une banque, qui intervient dans chaque paiement. Les stockages managés explicitent les limites ; Amazon DynamoDB, par exemple, documente un plafond par partition de 3000 unités de lecture et 1000 unités d’écriture par seconde. Les parades consistent notamment à mettre en cache les lectures chaudes, à scinder une clé chaude en lui ajoutant un petit suffixe aléatoire (écritures de restaurant:88#0 à restaurant:88#9, au prix de la lecture des dix clés et de leur combinaison), à déplacer un très gros locataire vers des partitions dédiées et à agréger les compteurs en mémoire avant de les écrire. Toutes ajoutent de la complexité : mesurez donc d’abord quelles clés sont réellement chaudes.

Index secondaires entre partitions

Le partitionnement fonctionne sans heurt pour les recherches par clé de partition. Les requêtes portant sur autre chose ont besoin d’un index secondaire, et il existe deux façons de le partitionner. Un index local (partitionné par document) vit avec chaque partition et ne couvre que les lignes de celle-ci. Les écritures ne mettent à jour qu’une partition, mais une requête sur l’attribut indexé doit être envoyée à toutes les partitions et les résultats fusionnés (scatter-gather, dispersion puis regroupement), si bien que sa latence est fixée par la partition la plus lente et que son coût augmente avec le nombre de partitions. Un index global (partitionné par terme) est lui-même partitionné selon la valeur indexée, si bien qu’une requête lit une seule partition d’index ; mais une seule écriture peut alors devoir mettre à jour des partitions d’index situées sur d’autres nœuds, ce qui se fait généralement de façon asynchrone, de sorte que l’index peut brièvement être en retard sur les données. DynamoDB illustre les deux : un index secondaire local partage la clé de partition de la table, tandis qu’un index secondaire global a la sienne et offre une cohérence à terme.

Erreurs courantes

Traiter les répliques comme des sauvegardes. Une réplique copie fidèlement un DELETE erroné en quelques millisecondes ; les sauvegardes et la restauration à un instant donné restent nécessaires (chapitre 0). Lire sur les répliques par défaut, sans décider quelles requêtes tolèrent le retard. Se fier à un failover automatique jamais répété, ou capable de promouvoir un nœud alors que l’ancien leader accepte encore des écritures. Partitionner trop tôt : une base de données partitionnée complique les jointures, les transactions, les changements de schéma et l’exploitation, et les nœuds uniques modernes sont très puissants ; épuisez d’abord l’indexation, la mise en cache, les répliques en lecture, l’archivage et le passage à l’échelle vertical. Partitionner selon une clé qui croît de façon monotone, ce qui concentre chaque insertion sur une seule partition. Oublier les chemins d’accès secondaires, et découvrir après le découpage qu’une requête importante doit désormais interroger chaque shard. Comme le montre la suite de ce chapitre, les bonnes alternatives passent souvent par des données dérivées : des modèles de lecture, des caches et des stockages analytiques, chacun construit pour un type de requête.