Replikation, Partitionierung und Sharding

Zwei verschiedene Gründe, Daten zu verteilen

Ein einzelner Datenbankserver stößt irgendwann an eine Grenze: Verfügbarkeit (er ist eine Maschine, die ausfallen kann), Lesekapazität, Schreibkapazität, Speicherplatz oder Nähe zu Nutzern auf anderen Kontinenten. Es gibt zwei einander ergänzende Wege, Daten über mehrere Maschinen zu verteilen, und es ist wichtig, sie nicht zu verwechseln. Replikation hält eine Kopie derselben Daten auf mehreren Knoten, für Verfügbarkeit, Leseskalierung und geringere Latenz in der Nähe der Nutzer. Partitionierung, in vielen Produkten Sharding genannt, teilt die Daten auf, sodass jeder Knoten eine Teilmenge hält, für Schreibdurchsatz und Speicherplatz über eine Maschine hinaus. Reale Systeme tun meist beides: Jede Partition wird repliziert.

Beide haben ernste Folgen für die Korrektheit. Replikation wirft die Frage auf, was ein Leser sieht, solange die Kopien nicht übereinstimmen; das ist das Thema von Konsistenzmodellen, CAP und PACELC. Partitionierung wirft die Frage auf, was mit Abfragen, Indizes und Transaktionen geschieht, die mehrere Partitionen umfassen. Dieser Abschnitt erklärt die Mechanismen so gründlich, dass Sie die Aussagen einer Datenbank beurteilen und einen Shard-Schlüssel entwerfen können.

Leader-Follower-Replikation

Das häufigste Schema ist die Leader-Follower-Replikation (auch Primary-Replica genannt; ältere Dokumentation sagt Master-Slave). Ein Knoten, der Leader, nimmt alle Schreibvorgänge an. Er zeichnet jede Änderung in seinem Log auf (dem Write-Ahead-Log von PostgreSQL, dem Binary Log von MySQL) und streamt dieses Log an die Follower, die die Änderungen in derselben Reihenfolge anwenden. Lesevorgänge können an den Leader oder an Follower gehen. PostgreSQL Streaming Replication, MySQL-Replikation, SQL Server Availability Groups, MongoDB Replica Sets und die meisten verwalteten Cloud-Datenbanken arbeiten so.

Die zentrale Entwurfsentscheidung ist, wann der Leader einen Commit als erfolgreich meldet. Bei synchroner Replikation wartet er, bis ein Follower bestätigt, dass er die Änderung hat, sodass ein festgeschriebener Schreibvorgang den Verlust des Leaders übersteht; der Preis ist, dass jeder Commit einen Netzwerk-Roundtrip enthält und Schreibvorgänge stocken, wenn der synchrone Follower ausgefallen ist. Bei asynchroner Replikation bestätigt der Leader sofort und versendet die Änderung danach; Commits sind schnell, und Follower können weit entfernt sein, aber wenn der Leader ausfällt, gehen die letzten Änderungen verloren, die noch keinen Follower erreicht hatten, obwohl dem Client gemeldet wurde, sie seien festgeschrieben. Ein verbreiteter Kompromiss, in MySQL semisynchron genannt, wartet auf einen Follower und lässt die übrigen asynchron laufen. In der Bank ist der Verlust einer bestätigten Zahlung inakzeptabel, daher ist mindestens eine synchrone Kopie (oder eine konsensbasierte Datenbank) üblich; für die Speisekarten-Datenbank der Essenslieferplattform kann ein möglicher Verlust von wenigen Sekunden ein akzeptabler Preis für Geschwindigkeit sein. Das ist die RPO-Abwägung aus Verfügbarkeit und Zuverlässigkeit in konkreter Form.

Replikationsverzögerung und die Anomalien, die sie verursacht

Asynchrone Follower liegen meist Millisekunden zurück, aber unter Last, während eines großen Batch-Updates oder über ein langsames Netzwerk kann die Replikationsverzögerung auf Sekunden oder Minuten anwachsen. Lesen von Followern erzeugt dann Anomalien, die Nutzer bemerken. Douglas Terry und Kollegen definierten die ersten beiden Garantien unten, Read-your-Writes und Monotonic Reads, 1994 als Session Guarantees; die dritte ist vor allem wichtig, wenn die Daten zusätzlich partitioniert sind.

Keine dieser Anomalien tritt auf, wenn jeder Lesevorgang an den Leader geht; die einfachste Lösung ist daher oft, Lesevorgänge, die frisch sein müssen, an den Leader zu senden und Follower nur für Lesevorgänge zu nutzen, die veraltete Daten vertragen: Produktlisten, Berichte, Suchfeeds. Entscheiden Sie das bewusst pro Abfrage, statt standardmäßig alle Lesevorgänge an Replikate zu leiten.

Failover und Split Brain

Failover ersetzt einen ausgefallenen Leader durch einen Follower. Es hat drei Schritte: erkennen, dass der Leader ausgefallen ist (meist über ein Timeout), den neuen Leader wählen (idealerweise den aktuellsten Follower, oft über ein Konsensprotokoll wie Raft zwischen den Knoten oder einen externen Koordinator wie etcd oder ZooKeeper) und die Schreibvorgänge auf ihn umleiten. Jeder Schritt kann schiefgehen. Ein zu kurzes Timeout löst bei einem kurzen Netzwerkaussetzer oder einer Lastspitze ein Failover aus, was die Lage verschlimmert; ein zu langes verlängert den Ausfall. Bei asynchroner Replikation gehen Schreibvorgänge verloren, die der alte Leader noch nicht versendet hatte, und kommt der alte Leader zurück und hält sie noch, geraten sie mit neuen Schreibvorgängen in Konflikt.

Der gefährlichste Ausfall ist Split Brain: Zwei Knoten halten sich beide für den Leader und nehmen beide Schreibvorgänge an, sodass die Daten auf eine Weise auseinanderlaufen, die sich womöglich nicht automatisch zusammenführen lässt. Es folgt typischerweise auf eine Netzwerkpartition, bei der der alte Leader abgeschnitten, aber noch aktiv ist. Schutz erfordert ein Quorum (ein Knoten darf nur führen, solange eine Mehrheit zustimmt), Fencing (die Autorität des neuen Leaders wird durch ein ständig steigendes Token durchgesetzt, das der Speicher prüft, oder der alte Leader wird zwangsweise abgeschaltet, eine Technik mit dem Spitznamen STONITH: shoot the other node in the head) und Clients, die zuverlässig dem aktuellen Leader folgen.

GitHubs Vorfall vom 21. Oktober 2018 zeigt, wie schwierig das in der Praxis ist. Ein Netzwerkfehler trennte 43 Sekunden lang die Verbindung zwischen seinem Netzknoten an der US-Ostküste und seinem Hauptrechenzentrum an der Ostküste. Seine Failover-Werkzeuge beförderten, wie vorgesehen, Datenbank-Primaries an der Westküste. Als die Verbindung zurückkehrte, hielten beide Küsten Schreibvorgänge, die der anderen fehlten, sodass GitHub nicht einfach zurückschalten konnte; zum Schutz der Datenintegrität entschied es sich für Wiederherstellen und Abgleichen, und der Dienst war 24 Stunden und 11 Minuten lang eingeschränkt. Automatisches Failover über Regionen hinweg hatte aus einem Aussetzer von 43 Sekunden eine tagelange Wiederherstellung gemacht.

Multi-Leader- und Leaderless Replication

Bei der Multi-Leader-Replikation nehmen mehrere Knoten Schreibvorgänge an und replizieren untereinander. Sie wird für mehrere Rechenzentren genutzt (jede Region schreibt lokal, sodass Nutzer keine Latenz über Ozeane hinweg haben), für Clients, die offline arbeiten (eine Kalender- oder Notiz-App ist faktisch ein Leader auf jedem Gerät), und für kollaboratives Bearbeiten. Das unvermeidliche Problem sind Schreibkonflikte: Zwei Leader ändern denselben Datensatz gleichzeitig. Lösungen sind Last Write Wins (einfach, verwirft aber stillschweigend Daten und hängt von Uhren ab), das Zusammenführen von Werten, das Behalten beider Versionen mit Rückfrage bei einem Menschen oder Datentypen, die sich automatisch zusammenführen lassen (CRDTs, Conflict-free Replicated Data Types). Am zuverlässigsten ist es, Konflikte zu vermeiden: Geben Sie jedem Datensatz eine Heimatregion, die allein seine Schreibvorgänge annimmt, etwa das Heimatland eines Kunden.

Leaderless Replication, bekannt geworden durch Amazons Dynamo-Paper (DeCandia und Kollegen, 2007) und verwendet in Apache Cassandra, ScyllaDB und Riak, hat überhaupt keinen Leader. Der Client oder ein Koordinatorknoten in seinem Auftrag sendet jeden Schreibvorgang an alle N Replikate eines Schlüssels und wertet ihn als erfolgreich, wenn W von ihnen bestätigen; jeden Lesevorgang sendet er an Replikate, wartet auf R Antworten und nimmt die neueste Version. Gilt R + W > N, überschneidet sich jede Lesemenge mit jeder erfolgreichen Schreibmenge in mindestens einem Replikat, sodass ein Lesevorgang normalerweise den neuesten bestätigten Schreibvorgang sieht. Replikate, die Schreibvorgänge verpasst haben, werden durch Read Repair (der Koordinator schreibt die neueste Version zurück, wenn er eine veraltete sieht) und eine im Hintergrund laufende Anti-Entropy repariert, die Replikate vergleicht. Cassandra bietet diese Wahl pro Anfrage als Konsistenzstufen wie ONE, QUORUM und ALL an.

Quoren wird oft zu viel zugetraut. R + W > N macht das System nicht schon allein linearisierbar: Nebenläufige Schreibvorgänge, Schreibvorgänge, die auf manchen Replikaten scheiterten und auf anderen gelangen, Sloppy Quorums (Schreiben auf Ersatzknoten während Ausfällen) und Last Write Wins mit abweichenden Uhren lassen alle veraltete oder verlorene Daten zu. Kapitel 6 geht die Quorum-Arithmetik und diese Grenzfälle in CAP und PACELC durch. Leaderless-Speicher eignen sich hervorragend für hohe Schreibverfügbarkeit über Ausfälle und Rechenzentren hinweg; für Invarianten wie denselben Platz nie zweimal verkaufen passen sie schlecht, sofern sie für solche Operationen keine Lightweight Transactions oder Konsensfunktion anbieten.

Partitionierung: nach Schlüsselbereich oder nach Hash

Partitionierung ordnet jeden Datensatz über seinen Partitionsschlüssel (oder Shard-Schlüssel) einer Partition zu, und das Ziel ist, Daten und Last gleichmäßig zu verteilen. Eine Partition, die unverhältnismäßig viel Last erhält, ist ein Hotspot, und eine ungleichmäßige Verteilung nennt man Schieflage (Skew). Es gibt zwei Grundschemata.

Schlüsselbereichspartitionierung gibt jeder Partition einen zusammenhängenden Bereich von Schlüsseln, wie die Bände eines Lexikons. Bereichsscans sind effizient, weil benachbarte Schlüssel zusammen gespeichert sind (alle Bestellungen vom Oktober in einer oder zwei Partitionen), und Partitionen lassen sich teilen, wenn sie wachsen, wie in Bigtable, HBase, Spanner und beim Range-Sharding von MongoDB. Die Gefahr ist ein Schlüssel, der mit der Zeit wächst: Ist der Schlüssel ein Zeitstempel oder eine fortlaufende Bestellnummer, landet jeder neue Schreibvorgang in der letzten Partition, die zum Hotspot wird, während die anderen untätig sind. Hash-Partitionierung wendet eine Hashfunktion auf den Schlüssel an und weist Partitionen Bereiche von Hashwerten zu, sodass sich selbst fortlaufende Schlüssel gleichmäßig verteilen. Der Preis ist, dass Bereichsabfragen auf den Schlüssel nun jede Partition fragen müssen. Viele Speicher kombinieren beides mit einem zusammengesetzten Schlüssel: Cassandra hasht den Partitionsschlüssel, um den Knoten zu wählen, und sortiert die Zeilen innerhalb der Partition nach Clustering-Spalten, sodass (customer_id, placed_at) die Bestellungen eines Kunden zusammen und in zeitlicher Reihenfolge auf einem Knoten ablegt.

Konsistentes Hashing und Rebalancing

Der naheliegende Weg, Hashes Knoten zuzuordnen, hash(key) mod N, hat einen fatalen Fehler: Ändert sich N, wandert fast jeder Schlüssel. Der Schritt von 4 auf 5 Knoten verlagert etwa 80 % der Schlüssel (jeden Schlüssel, dessen Rest für 4 und 5 verschieden ist), der von 10 auf 11 etwa 91 %, was bedeutet, fast den ganzen Datenbestand zu kopieren, nur um eine Maschine hinzuzufügen. Konsistentes Hashing, 1997 von David Karger und Kollegen für Web-Caches eingeführt, platziert Knoten und Schlüssel gleichermaßen auf einem kreisförmigen Hash-Raum; jeder Schlüssel gehört zum nächsten Knoten im Uhrzeigersinn. Ein neuer Knoten übernimmt nur die Schlüssel zwischen sich und seinem Vorgänger, etwa 1/N des Ganzen, und ein entfernter Knoten übergibt seine Schlüssel an seinen Nachbarn. Bei wenigen Knoten sind die Bögen ungleich, daher erhält jeder physische Knoten viele virtuelle Knoten (Tokens) an verschiedenen Stellen des Rings, was die Last ausgleicht und einen neuen Knoten von jedem bestehenden ein wenig übernehmen lässt. Dynamo und Cassandra verwenden dieses Schema.

Zwei weitere Rebalancing-Strategien sind verbreitet. Bei einer festen Anzahl von Partitionen legen Sie zu Beginn viel mehr Partitionen an, als es Knoten gibt (etwa 1024 für 10 Knoten), und weisen den Knoten ganze Partitionen zu; ein neuer Knoten verschiebt ein paar ganze Partitionen, Schlüssel werden nie neu gehasht. Elasticsearch-Shards, Couchbase-vBuckets und viele selbst gebaute Sharding-Schemata nutzen das, und die wichtigste Entscheidung ist eine Partitionsanzahl, die groß genug für künftiges Wachstum ist. Bei dynamischer Partitionierung, wie in HBase und MongoDB, teilt sich eine Partition, wenn sie eine Größenschwelle überschreitet, und die Hälften können auf verschiedene Knoten wandern. Bei jedem Schema verschiebt Rebalancing große Datenmengen über das Netzwerk, während das System Verkehr bedient, daher ziehen es viele Teams vor, dass ein Mensch jede Verschiebung genehmigt, und drosseln sie, um die Latenz zu schützen.

Anfragen müssen außerdem zur richtigen Partition geleitet werden: Entweder nimmt jeder Knoten eine Anfrage an und leitet sie weiter (Cassandra-Knoten lernen den Ring per Gossip), oder eine Routing-Schicht kennt die Partitionszuordnung (mongos bei MongoDB), oder die Client-Bibliothek hält die Zuordnung. Die Zuordnung selbst liegt meist in einem Koordinationsdienst wie etcd oder ZooKeeper oder in der eigenen Konsensgruppe der Datenbank.

Hot Keys und Schieflage

Hashing verteilt Schlüssel gleichmäßig, nicht die Last. Ist ein Schlüssel weit aktiver als die übrigen, ist seine Partition unabhängig vom Schema heiß: das Konto einer prominenten Person in einem sozialen Netzwerk, das Flaggschiff-Restaurant einer landesweiten Kette während einer Aktion oder das eigene Abwicklungskonto einer Bank, das jede Zahlung berührt. Verwaltete Speicher machen die Grenzen explizit; Amazon DynamoDB etwa dokumentiert eine Obergrenze pro Partition von 3000 Read Units und 1000 Write Units pro Sekunde. Gegenmaßnahmen sind das Cachen heißer Lesevorgänge, das Aufteilen eines Hot Key durch Anhängen eines kleinen zufälligen Suffixes (Schreibvorgänge auf restaurant:88#0 bis restaurant:88#9, um den Preis, alle zehn zu lesen und zusammenzuführen), das Verlagern eines sehr großen Mandanten auf eigene Partitionen und das Aggregieren von Zählern im Arbeitsspeicher, bevor sie geschrieben werden. Alle erhöhen die Komplexität, also messen Sie zuerst, welche Schlüssel tatsächlich heiß sind.

Sekundärindizes über Partitionen hinweg

Für Abfragen über den Partitionsschlüssel funktioniert Partitionierung reibungslos. Abfragen über irgendetwas anderes brauchen einen Sekundärindex, und es gibt zwei Arten, ihn zu partitionieren. Ein lokaler Index (dokumentpartitioniert) liegt bei jeder Partition und deckt nur deren Zeilen ab. Schreibvorgänge ändern eine einzige Partition, aber eine Abfrage über das indexierte Attribut muss an jede Partition gesendet und die Ergebnisse müssen zusammengeführt werden (Scatter-Gather), sodass ihre Latenz von der langsamsten Partition bestimmt wird und ihre Kosten mit der Zahl der Partitionen wachsen. Ein globaler Index (termpartitioniert) ist selbst nach dem indexierten Wert partitioniert, sodass eine Abfrage eine Indexpartition liest; dafür muss ein einzelner Schreibvorgang nun womöglich Indexpartitionen auf anderen Knoten ändern, was meist asynchron geschieht, sodass der Index den Daten kurz hinterherhinken kann. DynamoDB zeigt beides: Ein lokaler Sekundärindex teilt den Partitionsschlüssel der Tabelle, ein globaler Sekundärindex hat einen eigenen und ist eventually consistent.

Häufige Fehler

Replikate als Backups behandeln. Ein Replikat kopiert ein irrtümliches DELETE getreulich innerhalb von Millisekunden; Backups und Point-in-Time-Recovery sind weiterhin nötig (Kapitel 0). Standardmäßig von Replikaten lesen, ohne zu entscheiden, welche Abfragen Verzögerung vertragen. Einem automatischen Failover vertrauen, das nie geprobt wurde oder das einen Knoten befördern kann, während der alte Leader noch Schreibvorgänge annimmt. Zu früh sharden: Eine geshardete Datenbank erschwert Joins, Transaktionen, Schemaänderungen und den Betrieb, und moderne Einzelknoten sind sehr groß; schöpfen Sie zuerst Indizierung, Caching, Read Replicas, Archivierung und vertikale Skalierung aus. Nach einem monoton steigenden Schlüssel sharden, was jedes Insert auf eine Partition konzentriert. Sekundäre Zugriffswege vergessen und erst nach der Aufteilung feststellen, dass eine wichtige Abfrage nun jeden Shard fragen muss. Wie der Rest dieses Kapitels zeigt, beruhen die guten Alternativen oft auf abgeleiteten Daten: Lesemodelle, Caches und analytische Speicher, jeweils für eine Art von Abfrage gebaut.