🥇NewSQL = NoSQL+ACID

Bis vor kurzem wurden in Odnoklassniki etwa 50 TB Daten in Echtzeit in SQL Server verarbeitet. Bei diesem Datenvolumen ist es nahezu unmöglich, einen schnellen und zuverlässigen, zudem ausfallsicheren Rechenzentrumszugang mit SQL-Datenbanken zu gewährleisten. Üblicherweise werden in solchen Fällen eine der NoSQL-Datenbanken eingesetzt, doch nicht alles lässt sich nach NoSQL übertragen: Einige Entitäten benötigen ACID-Transaktionsgarantien.

Das hat uns zur Nutzung einer NewSQL-Datenbank geführt, d.h. einer Datenbank, die die Ausfallsicherheit, Skalierbarkeit und Geschwindigkeit von NoSQL-Systemen bietet, jedoch die gewohnten ACID-Garantien klassischer Systeme beibehält. Es gibt nur wenige funktionierende Industriesysteme dieser neuen Klasse, weshalb wir ein solches System selbst entwickelt und in den produktiven Betrieb genommen haben.

Wie es funktioniert und was dabei herausgekommen ist – lesen Sie weiter unten.

Heute hat die monatliche Nutzerzahl von „Odnoklassniki“ mehr als 70 Millionen einzigartige Besucher. Wir gehören zu den Top fünf eine der größten sozialen Netzwerke der Welt und gehört zu den zehn Websites, auf denen die Nutzer die meiste Zeit verbringen. Die Infrastruktur von "OK" bewältigt extrem hohe Lasten: mehr als eine Million HTTP-Anfragen pro Sekunde für die Frontends. Teile des Serverparks mit über 8000 Servern sind nahe beieinander in vier Rechenzentren in Moskau untergebracht, was eine Netzwerklatenz von unter 1 ms zwischen ihnen ermöglicht.

Wir verwenden Cassandra seit 2010, beginnend mit Version 0.6. Heute sind mehrere Dutzend Cluster im Einsatz. Der schnellste Cluster verarbeitet mehr als 4 Millionen Operationen pro Sekunde, während der größte 260 TB speichert.

Allerdings sind das alles gewöhnliche NoSQL-Cluster, die für die Speicherung verwendet werden. von schwach konsistenten Daten. Wir wollten jedoch das primäre konsistente Speichersystem, Microsoft SQL Server, ersetzen, das seit der Gründung von "Odnoklassniki" verwendet wurde. Das Speichersystem bestand aus mehr als 300 SQL Server Standard Edition Maschinen, auf denen 50 TB an Geschäftseinheiten gespeichert waren. Diese Daten werden im Rahmen von ACID-Transaktionen modifiziert und erfordern eine hohe Konsistenz..

Für die Verteilung von Daten über SQL Server-Knoten haben wir sowohl vertikale als auch horizontale Partitionierung (Sharding) verwendet. Historisch haben wir ein einfaches Sharding-Schema genutzt: Jeder Entität wurde ein Token zugewiesen – eine Funktion, die vom Entitäts-ID abgeleitet ist. Entitäten mit demselben Token wurden auf einen SQL-Server verschoben. Das Verhältnis zwischen Master- und Detaildatensatz wurde so umgesetzt, dass die Tokens des Haupt- und des erzeugten Datensatzes immer übereinstimmten und sich auf demselben Server befanden. In sozialen Netzwerken werden fast alle Datensätze im Namen des Benutzers erzeugt – das heißt, alle Benutzerdaten innerhalb eines funktionalen Subsystems werden auf einem Server gespeichert. Das bedeutet, dass bei Geschäftstransaktionen fast immer Tabellen eines SQL-Servers beteiligt waren, was die Konsistenz der Daten durch lokale ACID-Transaktionen gewährleistete, ohne die Notwendigkeit langsamer und unzuverlässiger verteilte ACID-Transaktionen zu verwenden.

Dank Sharding und zur Beschleunigung der SQL-Leistung:

Verwenden wir keine Foreign Key Constraints, da sich beim Sharding die ID der Entität auf einem anderen Server befinden kann.
Wir verwenden keine gespeicherten Prozeduren und Trigger, da sie die CPU der Datenbank belasten.
JOINs vermeiden wir aus denselben Gründen sowie aufgrund der vielen zufälligen Lesevorgänge von der Festplatte.
Um Deadlocks zu reduzieren, nutzen wir außerhalb von Transaktionen die Isolationsebene Read Uncommitted.
Wir führen nur kurze Transaktionen durch (im Durchschnitt unter 100 ms).
Multireihen-UPDATEs und DELETEs vermeiden wir aufgrund der hohen Anzahl von Deadlocks – wir aktualisieren immer nur einen Datensatz.
Abfragen führen wir stets nur über Indizes aus – eine Abfrage mit einem vollständigen Tabellenansichtsplan bedeutet für uns eine Überlastung der Datenbank und deren Ausfall.

Diese Maßnahmen haben es uns ermöglicht, nahezu die maximale Leistung aus den SQL-Servern herauszuholen. Allerdings wurden die Probleme immer zahlreicher. Lassen Sie uns diese betrachten.

Probleme mit SQL

Da wir eine eigene Sharding-Lösung verwendet haben, wurden zusätzliche Shards manuell von den Administratoren hinzugefügt. Während dieser Zeit haben die skalierbaren Datenreplikate keine Anfragen bedient.
Wenn die Anzahl der Einträge in einer Tabelle wächst, sinkt die Geschwindigkeit beim Einfügen und Modifizieren. Bei der Hinzufügung von Indizes zu einer bestehenden Tabelle sinkt die Geschwindigkeit exponentiell, und die Erstellung sowie das Neuanlegen von Indizes erfolgt mit Ausfallzeiten.
Die Präsenz einer kleinen Anzahl von Windows-Instanzen für SQL Server in der Produktion erschwert das Management der Infrastruktur.

Doch das Hauptproblem ist —

Ausfallsicherheit

Ein klassischer SQL-Server hat eine mangelhafte Ausfallsicherheit. Angenommen, Sie haben nur einen Datenbankserver, der alle drei Jahre ausfällt. Während dieser Zeit ist die Website für 20 Minuten nicht verfügbar, was akzeptabel ist. Bei 64 Servern ist die Website bereits alle drei Wochen nicht verfügbar. Und bei 200 Servern ist die Website jede Woche nicht verfügbar. Das ist ein Problem.

Was kann getan werden, um die Ausfallsicherheit des SQL-Servers zu erhöhen? Wikipedia schlägt vor, einen hochverfügbaren Clustereinzurichten, bei dem im Falle eines Ausfalls einer der Komponenten eine redundante vorhanden ist.

Das erfordert eine teure Hardware-Infrastruktur: umfangreiche Redundanzen, Glasfaser, gemeinsamen Speicher und auch das Einschalten der Reserve funktioniert unzuverlässig: etwa 10 % der Aktivierungen enden mit einem Ausfall der Reserveknoten hinter dem Hauptknoten.

Der Hauptnachteil eines solchen hochverfügbaren Clusters ist jedoch die vollständige Unzugänglichkeit, wenn das Rechenzentrum, in dem es betrieben wird, ausfällt. „OdnoClassniki“ betreibt vier Rechenzentren, und wir müssen sicherstellen, dass der Betrieb bei einem vollständigen Ausfall eines dieser Zentren aufrechterhalten wird.

Zu diesem Zweck könnte man Multi-Master Replikation verwenden, die in SQL Server integriert ist. Diese Lösung ist aufgrund der Softwarekosten deutlich teurer und leidet unter den bekannten Problemen der Replikation – unvorhersehbaren Verzögerungen bei Transaktionen bei synchroner Replikation und Verzögerungen bei der Anwendung von Replikationen (und infolgedessen verloren gegangenen Änderungen) bei asynchroner Replikation. Die implizierte manuelle Konfliktlösung macht diese Option für uns völlig unbrauchbar.

Alle diese Probleme erforderten eine radikale Lösung, und wir begannen mit deren detaillierter Analyse. Hier müssen wir uns mit dem vertraut machen, was SQL Server im Wesentlichen macht – Transaktionen.

Eine einfache Transaktion

Betrachten wir eine einfache Transaktion aus der Perspektive eines praktischen SQL-Programmierers: das Hinzufügen eines Fotos zu einem Album. Alben und Fotos werden in verschiedenen Tabellen gespeichert. Jedes Album hat einen Zähler für öffentliche Fotos. Somit wird diese Transaktion in die folgenden Schritte unterteilt:

Sperren Sie das Album mit dem Schlüssel.
Erstellen Sie einen Eintrag in der Fototabelle.
Wenn das Foto den öffentlichen Status hat, erhöhen wir den Zähler für öffentliche Fotos im Album, aktualisieren den Eintrag und beenden die Transaktion.

Oder in Form von Pseudocode:

TX.start("Alben", id);
Album album = albums.lock(id);
Photo photo = photos.create(…);

if (photo.status == PUBLIC) {
    album.incPublicPhotosCount();
}
album.update();

TX.commit();

Wir sehen, dass das häufigste Szenario einer Geschäftstransaktion darin besteht, Daten aus der DB in den Arbeitsspeicher des Anwendungsservers zu lesen, etwas zu ändern und die neuen Werte zurück in die DB zu speichern. Normalerweise aktualisieren wir in einer solchen Transaktion mehrere Entitäten, mehrere Tabellen.

Bei der Durchführung einer Transaktion kann es zu einer gleichzeitigen Modifikation derselben Daten aus einem anderen System kommen. Beispielsweise könnte ein Antispam-System entscheiden, dass ein Nutzer verdächtig ist, und damit alle Fotos des Nutzers nicht mehr öffentlich sein sollten, was bedeutet, dass der photo.status auf einen anderen Wert geändert und die entsprechenden Zähler zurückgesetzt werden müssen. Offensichtlich wird das Ergebnis nicht den Anforderungen entsprechen, wenn diese Operation ohne Garantien für die atomare Anwendung und Isolation konkurrierender Modifikationen erfolgt, wie in ACID, denn das Ergebnis könnte entweder einen falschen Zählerstand für die Fotos anzeigen oder es werden nicht alle Fotos zur Moderation geschickt.

Solchen Code, der mit verschiedenen Geschäftseinheiten innerhalb einer Transaktion manipuliert, wurde während der gesamten Existenz von Odnoklassniki in großer Zahl geschrieben. Aus Erfahrung mit Migrationen zu NoSQL gibt es Eventual Consistency Wir wissen, dass die größten Herausforderungen (und der damit verbundene Zeitaufwand) darin bestehen, Code zu entwickeln, der die Konsistenz der Daten sicherstellt. Daher war das Hauptaugenmerk bei der Auswahl eines neuen Speichersystems die Bereitstellung echter ACID-Transaktionen für die Anwendungsebene.

Weitere ebenso wichtige Anforderungen waren:

Im Falle eines Ausfalls des Rechenzentrums müssen sowohl Lese- als auch Schreibzugriffe auf das neue Speichersystem möglich sein.
Die aktuelle Entwicklungsgeschwindigkeit muss beibehalten werden. Das bedeutet, dass bei der Arbeit mit dem neuen Speichersystem die Menge des Codes etwa gleich bleiben sollte und es keine Notwendigkeit entstehen darf, zusätzliche Funktionen im Speicher zu implementieren, Algorithmen zur Konfliktlösung zu entwickeln, sekundäre Indizes aufrechtzuerhalten usw.
Die Leistung des neuen Speichersystems muss sowohl bei Lesevorgängen als auch bei der Verarbeitung von Transaktionen ausreichend hoch sein, was effektiv die Unanwendbarkeit akademisch strenger, universaler, aber langsamer Lösungen impliziert, wie z. B. zwei-phasige Commit-Prozesse..
Automatisches On-the-Fly-Scaling.
Der Einsatz normaler kostengünstiger Server, ohne dass exotische Hardware angeschafft werden muss.
Die Möglichkeit, die Speicherlösungen durch die Entwickler des Unternehmens weiterzuentwickeln. Mit anderen Worten, der Fokus lag auf eigenen oder Open-Source-Lösungen, vorzugsweise in Java.

Lösungen, Lösungen

Bei der Analyse möglicher Lösungen kamen wir zu zwei Architekturentscheidungen:

Die erste ist, einen beliebigen SQL-Server zu verwenden und die erforderliche Fehlertoleranz, Skalierungsmechanismen, ein fehlertolerantes Cluster, Konfliktlösungen sowie verteilte, zuverlässige und schnelle ACID-Transaktionen umzusetzen. Wir haben diese Option als recht komplex und arbeitsintensiv eingeschätzt.

Die zweite Möglichkeit besteht darin, ein fertiges NoSQL-Speicher-System zu wählen, das über implementierte Skalierung, ein fehlertolerantes Cluster und Konfliktlösungen verfügt, während wir Transaktionen und SQL selbst umsetzen. Auf den ersten Blick scheint die Umsetzung von SQL, ganz zu schweigen von ACID-Transaktionen, eine jahrelange Aufgabe zu sein. Doch dann erkannten wir, dass der Satz von SQL-Funktionen, den wir in der Praxis nutzen, so weit vom ANSI SQL entfernt ist wie Cassandra CQL weit vom ANSI SQL entfernt ist. Bei näherer Betrachtung von CQL wurde uns klar, dass es ziemlich nah an dem ist, was wir benötigen.

Cassandra und CQL

Was macht Cassandra so interessant und welche Möglichkeiten bietet sie?

Zunächst können hier Tabellen mit Unterstützung für verschiedene Datentypen erstellt werden, und es sind SELECT- oder UPDATE-Abfragen über den Primärschlüssel möglich.

CREATE TABLE photos (id bigint KEY, owner bigint,…);
SELECT * FROM photos WHERE id=?;
UPDATE photos SET … WHERE id=?;

Um die Konsistenz der Replikate zu gewährleisten, verwendet Cassandra einen Quorum-Ansatz. Im einfachsten Fall bedeutet dies, dass bei der Bereitstellung von drei Replikaten derselben Zeile auf verschiedenen Knoten des Clusters der Datensatz als erfolgreich betrachtet wird, wenn die Mehrheit der Knoten (d.h. zwei von drei) den Erfolg dieser Schreiboperation bestätigt hat. Die Daten der Zeile gelten als konsistent, wenn bei der Abfrage die Mehrheit der Knoten befragt wurde und dies bestätigt haben. So wird bei drei Replikaten vollständige und sofortige Datenkonsistenz im Falle eines Ausfalls eines Knotens garantiert. Dieser Ansatz ermöglicht es uns, ein noch zuverlässigeres Schema umzusetzen: immer Anfragen an alle drei Replikate zu senden und auf die Antworten der zwei schnellsten zu warten. Die verspätete Antwort des dritten Replikats wird in diesem Fall verworfen. Der verspätet antwortende Knoten könnte ernsthafte Probleme haben — Verzögerungen, Garbage Collection in der JVM, Direct Memory Reclaim im Linux-Kernel, Hardwarefehler, Netzwerkunterbrechungen. Allerdings hat dies keine Auswirkungen auf die Client-Operationen und die Daten.

Der Ansatz, bei dem wir drei Knoten ansprechen und Antworten von zwei erhalten, wird genannt Spekulation: Anfragen an überflüssige Replikate werden gesendet, bevor der Knoten «ausfällt».

Ein weiterer Vorteil von Cassandra ist der Batchlog – ein Mechanismus, der entweder die vollständige Anwendung oder vollständige Ablehnung eines von Ihnen vorgenommenen Änderungssets garantiert. Dies ermöglicht es uns, A in ACID – die Atomarität – ab Werk zu gewährleisten.

Das, was am nächsten an Transaktionen in Cassandra herankommt, sind die sogenannten „lightweight transactions“. Aber sie sind weit von „echten“ ACID-Transaktionen entfernt: Tatsächlich handelt es sich um die Möglichkeit, CAS auf Daten von nur einem Datensatz zuzugreifen, unter Verwendung eines Konsens über das schwergewichtige Protokoll Paxos. Daher ist die Geschwindigkeit solcher Transaktionen gering.

Was uns in Cassandra gefehlt hat

Also standen wir vor der Herausforderung, echte ACID-Transaktionen in Cassandra zu implementieren. Mit deren Hilfe könnten wir zwei weitere praktische Funktionen klassischer DBMS einfach umsetzen: konsistente schnelle Indizes, die es uns ermöglichen würden, Daten nicht nur nach dem Primärschlüssel abzurufen, sowie einen ordentlichen Generator für monoton steigende Autoinkrement-IDs.

C*One

So entstand ein neues DBMS C*One, das aus drei Typen von Serverknoten besteht:

Speicher — (fast) standardmäßige Cassandra-Server, die für die Speicherung von Daten auf lokalen Festplatten verantwortlich sind. Mit zunehmender Last und Datenmenge lässt sich ihre Anzahl problemlos auf Dutzende oder Hunderte skalieren.
Transaktionskoordinatoren — sorgen für die Ausführung von Transaktionen.
Clients — Anwendungsserver, die Geschäftsoperationen durchführen und Transaktionen initiieren. Solcher Clients kann es Tausende geben.

Server aller Typen sind in einem gemeinsamen Cluster zusammengeschlossen, nutzen das interne Nachrichtenprotokoll von Cassandra zur Kommunikation und entfernt wurde. für den Austausch von Clusterinformationen. Mit Hilfe von Heartbeat erfahren die Server von gegenseitigen Ausfällen, halten ein gemeinsames Datenschema — Tabellen, deren Struktur und Replikation; das Partitionierungsschema, die Cluster-Topologie usw.

Windows XP, Vista, 7, 8, 10, Server 2003, Server 2008, Server 2012

Anstelle von Standardtreibern wird der Modus Fat Client verwendet. Eine solche Node speichert keine Daten, kann jedoch als Koordinator für die Ausführung von Anfragen fungieren, d. h. der Client übernimmt selbst die Funktion des Koordinators seiner Anfragen: er befragt die Replikate des Speichers und löst Konflikte. Dies ist nicht nur zuverlässiger und schneller als der Standardtreiber, der eine Kommunikation mit einem entfernten Koordinator erfordert, sondern ermöglicht auch eine bessere Steuerung der Anfrageübertragung. Außerhalb einer offenen Transaktion des Clients werden die Anfragen an die Speicher gesendet. Wenn der Client jedoch eine Transaktion eröffnet hat, werden alle Anfragen innerhalb der Transaktion an den Transaktionskoordinator gesendet.

Transaktionskoordinator C*One

Der Koordinator ist das, was wir für C*One von Grund auf neu implementiert haben. Er ist verantwortlich für die Verwaltung von Transaktionen, Sperren und Reihenfolge der Anwendungstransaktionen.

Für jede bearbeitete Transaktion erzeugt der Koordinator einen Zeitstempel: jeder nachfolgende ist größer als der vorhergehende. Da in Cassandra das Konfliktlösungsverfahren auf Zeitstempeln basiert (von zwei konkurrierenden Einträgen wird der mit dem späteren Zeitstempel als aktuell betrachtet), wird ein Konflikt immer zugunsten der nachfolgenden Transaktion gelöst. So haben wir realisiert Lamport-Uhren — eine kostengünstige Lösung zur Konfliktbewältigung in verteilten Systemen.

Sperren

Um die Isolation zu gewährleisten, haben wir uns für die einfachste Methode entschieden — pessimistische Sperren basierend auf dem Primärschlüssel des Datensatzes. Das bedeutet, dass in der Transaktion der Datensatz zuerst gesperrt werden muss, bevor er gelesen, modifiziert und gespeichert werden kann. Nur nach einem erfolgreichen Commit kann der Datensatz entsperrt werden, sodass konkurrierende Transaktionen darauf zugreifen können.

Die Umsetzung einer solchen Sperre ist in einer nicht verteilten Umgebung einfach. In einem verteilten System gibt es zwei Hauptansätze: entweder die Umsetzung einer verteilten Sperre im Cluster oder die Verteilung der Transaktionen so, dass Transaktionen, die einen bestimmten Datensatz betreffen, immer von demselben Koordinator bearbeitet werden.

Da in unserem Fall die Daten bereits auf Gruppen lokaler Transaktionen in SQL verteilt sind, wurde beschlossen, den Koordinatoren der Gruppen lokaler Transaktionen bestimmte Bereiche zuzuweisen: Ein Koordinator bearbeitet alle Transaktionen mit einem Token von 0 bis 9, der zweite von 10 bis 19 und so weiter. Dadurch wird jeder der Koordinatoren zum Master der Transaktionsgruppe.

Damit können Sperren in Form einer einfachen HashMap im Speicher des Koordinators implementiert werden.

Ausfälle der Koordinatoren

Da ein Koordinator ausschließlich eine Gruppe von Transaktionen bedient, ist es sehr wichtig, schnell zu erkennen, wenn er ausfällt, damit der erneute Versuch der Transaktionsausführung innerhalb des Timeouts liegt. Um dies schnell und zuverlässig zu erreichen, haben wir ein vollverknüpftes Quorum-Hearbeat-Protokoll angewendet:

In jedem Rechenzentrum werden mindestens zwei Koordinator-Knoten untergebracht. Regelmäßig sendet jeder Koordinator ein Heartbeat-Signal an die anderen Koordinatoren und informiert sie über seinen Status sowie darüber, von welchen Koordinatoren im Cluster er zuletzt Heartbeat-Signale empfangen hat.

Indem er ähnliche Informationen aus den Heartbeat-Signalen der anderen erhält, entscheidet jeder Koordinator für sich, welche Cluster-Knoten funktionieren und welche nicht, wobei er dem Quorum-Prinzip folgt: Wenn Knoten X von der Mehrheit der Knoten im Cluster die Bestätigung für den ordnungsgemäßen Empfang von Nachrichten von Knoten Y erhält, bedeutet dies, dass Y funktioniert. Umgekehrt gilt, sobald die Mehrheit meldet, dass keine Nachrichten mehr von Knoten Y empfangen wurden, gilt Y als ausgefallen. Interessanterweise wird Knoten X, wenn das Quorum meldet, dass es keine Nachrichten mehr von ihm erhält, selbst für ausgefallen halten.

Heartbeat-Nachrichten werden in hoher Frequenz gesendet, etwa 20 Mal pro Sekunde, mit einem Intervall von 50 ms. In Java ist es schwierig, eine Anwendungsantwort innerhalb von 50 ms zu garantieren, da die Pausendauer, die durch die Garbage Collection verursacht wird, vergleichbar ist. Es ist uns gelungen, eine derartige Reaktionszeit mit dem G1-Garbage-Collector zu erreichen, der es ermöglicht, ein Ziel für die Dauer der GC-Pausen anzugeben. Gelegentlich, recht selten, können jedoch die Pausen des Collectors 50 ms überschreiten, was zu einer Fehlinterpretation eines Ausfalls führen kann. Um dies zu vermeiden, meldet der Koordinator den Ausfall eines Remote-Knotens nicht, sobald das erste Heartbeat-Nachricht von ihm fehlt, sondern erst wenn mehrere aufeinanderfolgende ausbleiben. So haben wir es geschafft, den Ausfall eines Koordinator-Knotens innerhalb von 200 ms zu erkennen.

Es reicht jedoch nicht aus, schnell zu erkennen, welcher Knoten ausgefallen ist. Da muss etwas unternommen werden.

Redundanz

Das klassische Schema sieht im Falle eines Master-Ausfalls die Wahl eines neuen Masters durch einen der modernen universellen Algorithmen. Allerdings gibt es bei solchen Algorithmen gut bekannte Probleme mit der Konvergenzzeit und der Dauer des Wahlprozesses selbst. Solche zusätzlichen Verzögerungen konnten wir durch das Koordinator-Ersatzschema in einem vollständig vernetzten Netzwerk vermeiden:

Angenommen, wir möchten eine Transaktion in der Gruppe 50 durchführen. Vorab definieren wir das Ersatzschema, also welche Knoten die Transaktionen der Gruppe 50 im Falle eines Ausfalls des Hauptkoordinators ausführen werden. Unser Ziel ist es, die Funktionsfähigkeit des Systems im Falle eines Ausfalls des Rechenzentrums aufrechtzuerhalten. Wir legen fest, dass der erste Ersatz aus einem anderen Rechenzentrum und der zweite Ersatz aus einem dritten Rechenzentrum besteht. Dieses Schema wird einmal festgelegt und bleibt unverändert, solange sich die Topologie des Clusters nicht ändert, also bis neue Knoten hinzukommen (was sehr selten passiert). Die Reihenfolge der Auswahl eines neuen aktiven Masters im Falle eines Ausfalls des alten wird immer so aussehen: der erste Ersatz wird zum aktiven Master, und wenn dieser nicht mehr funktioniert, wird der zweite Ersatz aktiv.

Ein solches Schema ist zuverlässiger als ein universeller Algorithmus, da zur Aktivierung eines neuen Masters lediglich der Ausfall des alten festgestellt werden muss.

Wie können die Kunden wissen, welcher der Master gerade arbeitet? In 50 ms ist es unmöglich, Informationen an Tausende von Kunden zu senden. Es kann vorkommen, dass ein Kunde eine Anfrage zur Eröffnung einer Transaktion sendet, ohne zu wissen, dass dieser Master bereits nicht mehr funktioniert, und die Anfrage somit im Timeout stecken bleibt. Um dies zu vermeiden, senden die Kunden spekulativ eine Anfrage zur Eröffnung einer Transaktion gleichzeitig an den Gruppenmaster und beide seine Reserven, aber nur der Master, der gerade aktiv ist, wird auf diese Anfrage antworten. Alle weiteren Kommunikationen im Rahmen der Transaktion erfolgen nur mit dem aktiven Master.

Die Reserve-Master platzieren erhaltene Anfragen für nicht eigene Transaktionen in eine Warteschlange der nicht erfüllten Transaktionen, wo sie eine gewisse Zeit gespeichert werden. Wenn der aktive Master ausfällt, bearbeitet ein neuer Master die Anfragen zur Eröffnung von Transaktionen aus seiner Warteschlange und antwortet dem Kunden. Wenn der Kunde bereits eine Transaktion mit dem alten Master eröffnet hat, wird die zweite Antwort ignoriert (und diese Transaktion wird offensichtlich nicht abgeschlossen und muss vom Kunden erneut durchgeführt werden).

Wie funktioniert eine Transaktion

Angenommen, ein Kunde sendet dem Koordinator eine Anfrage zur Eröffnung einer Transaktion für eine bestimmte Entität mit einem bestimmten Primärschlüssel. Der Koordinator sperrt diese Entität und platziert sie in einer Speichertabelle für Sperren. Wenn nötig, liest der Koordinator diese Entität aus dem Speicher und speichert die erhaltenen Daten im Transaktionszustand im Speicher des Koordinators.

Wenn der Kunde die Daten in der Transaktion ändern möchte, sendet er dem Koordinator eine Anfrage zur Modifikation der Entität, und dieser platziert die neuen Daten in der Tabelle für Transaktionszustände im Speicher. Damit ist die Speicherung abgeschlossen – es erfolgt keine Aufnahme in den Speicher.

Wenn der Kunde während einer aktiven Transaktion seine eigenen geänderten Daten anfragt, handelt der Koordinator wie folgt:

Wenn die ID bereits in der Transaktion vorhanden ist, werden die Daten aus dem Speicher entnommen;
Falls die ID im Speicher nicht vorhanden ist, werden die fehlenden Daten aus den Node-Speichern gelesen, mit den bereits im Speicher vorhandenen kombiniert und das Ergebnis dem Kunden übergeben.

Auf diese Weise kann der Kunde seine eigenen Änderungen lesen, während andere Kunden diese Änderungen nicht sehen, da sie nur im Speicher des Koordinators gehalten werden und in den Cassandra-Nodes noch nicht existieren.

Wenn ein Kunde einen Commit sendet, wird der Zustand, der im Dienst gespeichert war, vom Koordinator in einem logged batch gesichert und als logged batch an die Cassandra-Speicher gesendet. Die Speicher führen alles Notwendige aus, damit dieses Paket atomar (vollständig) angewendet wird und geben dem Koordinator eine Rückmeldung, der daraufhin die Sperren aufhebt und den Erfolg der Transaktion dem Kunden bestätigt.

Für die Rücksetzung muss der Koordinator lediglich den Speicher freigeben, der für den Zustand der Transaktion verwendet wurde.

Durch die oben beschriebenen Anpassungen haben wir die ACID-Prinzipien implementiert:

Atomarität. Dies ist eine Garantie dafür, dass keine Transaktion teilweise im System erfasst wird; es werden entweder alle ihre Unteroperationen ausgeführt oder keine. In unserem Fall wird dieses Prinzip durch den logged batch in Cassandra gewährleistet.
Konsistenz. Jede erfolgreiche Transaktion erfasst definitionsgemäß nur zulässige Ergebnisse. Wenn nach dem Öffnen einer Transaktion und der Ausführung einiger Operationen festgestellt wird, dass das Ergebnis nicht zulässig ist, wird ein Rollback durchgeführt.
Isolierung. Bei der Ausführung einer Transaktion sollten parallele Transaktionen das Ergebnis nicht beeinflussen. Wettbewerbsfähige Transaktionen sind durch pessimistische Sperren auf dem Koordinator isoliert. Für Lesevorgänge außerhalb der Transaktion gilt der Isolationsgrundsatz auf der Ebene Read Committed.
Widerstandsfähigkeit. Unabhängig von Problemen auf unteren Ebenen — Stromausfall, Hardwarefehler — müssen die Änderungen, die von einer erfolgreich abgeschlossenen Transaktion vorgenommen wurden, nach dem Wiederanlauf des Systems erhalten bleiben.

Lesen über Indizes

Nehmen wir eine einfache Tabelle:

CREATE TABLE photos (
id bigint primary key,
owner bigint,
modified timestamp,
…)

Diese Tabelle hat eine ID (Primärschlüssel), einen Besitzer und ein Änderungsdatum. Wir müssen eine sehr einfache Abfrage durchführen — die Daten nach dem Besitzer mit dem Änderungsdatum „in den letzten 24 Stunden“ abfragen.

SELECT *
WHERE owner=?
AND modified>?

Damit eine solche Abfrage schnell ausgeführt wird, muss in einer klassischen SQL-Datenbank ein Index über die Spalten (owner, modified) erstellt werden. Das können wir ziemlich einfach umsetzen, da wir nun ACID-Garantien haben!

Indizes in C*One

Es gibt eine Ursprungstabelle mit Fotos, in der die ID des Datensatzes der Primärschlüssel ist.

Für den Index C*One wird eine neue Tabelle erstellt, die eine Kopie der Ausgangstabelle darstellt. Der Schlüssel entspricht dem Indexausdruck und umfasst zudem den Primärschlüssel des Datensatzes aus der Ausgangstabelle:

Nun kann die Abfrage nach «Besitzer in den letzten 24 Stunden» wie folgt umformuliert werden: select aus einer anderen Tabelle:

SELECT * FROM i1_test
WHERE owner=?
AND modified>?

Die Konsistenz der Ausgangstabelle photos und des Index i1 wird automatisch vom Koordinator aufrechterhalten. Basierend auf dem Schema der Daten generiert der Koordinator bei Änderungen nicht nur das Haupttabelle-Update, sondern merkt sich auch die Änderungen bei Kopien. Es sind keine zusätzlichen Aktionen mit der Indextabelle erforderlich, Protokolle werden nicht gelesen, Sperren werden nicht verwendet. Das Hinzufügen von Indizes verbraucht also fast keine Ressourcen und hat praktisch keinen Einfluss auf die Geschwindigkeit der Anwendung von Modifikationen.

Mit ACID haben wir Indizes wie in SQL implementiert. Sie bieten Konsistenz, Skalierbarkeit, hohe Geschwindigkeiten, können zusammengesetzt werden und sind in die CQL-Abfragesprache integriert. Für die Unterstützung der Indizes sind keine Änderungen am Anwendungscode erforderlich. Alles ist so einfach wie in SQL. Und das Wichtigste ist, dass Indizes die Ausführungsgeschwindigkeit der Modifikationen der ursprünglichen Transaktionstabelle nicht beeinflussen.

Was entstanden ist

Wir haben C*One vor drei Jahren entwickelt und in den produktiven Einsatz gebracht.

Was haben wir letztendlich erreicht? Lassen Sie uns das am Beispiel des Systems zur Verarbeitung und Speicherung von Fotos betrachten, einem der wichtigsten Datentypen in sozialen Netzwerken. Dabei geht es nicht um die Fotos selbst, sondern um die vielfältigen Metainformationen. Momentan hat "Odnoklassniki" etwa 20 Milliarden solcher Einträge, das System verarbeitet 80.000 Leseanforderungen pro Sekunde und bis zu 8.000 ACID-Transaktionen pro Sekunde, die mit der Datenmodifikation verbunden sind.

Als wir SQL mit einem Replikationsfaktor von 1 verwendeten (aber in RAID 10), wurde die Metainformation der Fotos in einem hochverfügbaren Cluster aus 32 Maschinen mit Microsoft SQL Server (plus 11 Sicherungsservern) gespeichert. Zusätzlich wurden 10 Server für die Sicherung von Backups bereitgestellt. Insgesamt handelt es sich um 50 kostspielige Maschinen. Dabei arbeitete das System mit einer nominalen Last, ohne Spielraum.

Nach der Migration auf das neue System haben wir einen Replikationsfaktor von 3 erreicht — eine Kopie in jedem Rechenzentrum. Das System besteht aus 63 Knoten des Cassandra-Speichers und 6 Steuerungsmaschinen, insgesamt 69 Server. Diese Maschinen sind jedoch deutlich günstiger, ihre Gesamtkosten betragen etwa 30 % der Kosten des SQL-Systems. Die Last liegt konstant bei 30 %.

Mit der Einführung von C*One haben sich auch die Latenzen verringert: Bei SQL benötigte ein Schreibvorgang etwa 4,5 ms. Bei C*One liegt die Zeit bei ungefähr 1,6 ms. Die Dauer einer Transaktion liegt im Durchschnitt unter 40 ms, der Commit erfolgt in 2 ms, die Lese- und Schreibdauer beträgt im Durchschnitt 2 ms. Der 99. Perzentil liegt bei gerade einmal 3-3,1 ms, die Anzahl der Timeouts ist um das 100-fache gesenkt worden — alles dank der breiten Anwendung von Spekulationen.

Bis jetzt wurde der Großteil der SQL Server-Nodes stillgelegt, neue Produkte werden ausschließlich mit C*One entwickelt. Wir haben C*One für den Betrieb in unserer Cloud optimiert. one-cloud, was die Bereitstellung neuer Cluster beschleunigt, die Konfiguration vereinfacht und den Betrieb automatisiert. Ohne den Quellcode wäre dies deutlich komplizierter und fehleranfälliger gewesen.

Aktuell arbeiten wir daran, andere Speicher in die Cloud zu migrieren — aber das ist eine ganz andere Geschichte.

Quelle: habr.com