Amazon Dynamo

Amazon Dynamo ist eine verteilte Hashtabelle, die bei der Firma Amazon.com intern genutzt wird. Wie auch das Google File System ist Dynamo für eine konkrete Anwendung optimiert, die auf die Anforderungen einiger Amazon Web Services zugeschnitten ist, die eine hohe Ausfallsicherheit benötigen.

Amazon-Anwendungen erwarten, dass ein Speichersystem hochverfügbar und extrem ausfallsicher ist. Insbesondere muss in jeder Situation gespeichert werden können.

Das System muss jederzeit inkrementell skalierbar sein, um Belastungsspitzen abdecken zu können, beispielsweise im Weihnachtsgeschäft. Komplizierte Datenbankzugriffe werden vermieden, der Zugriff erfolgt direkt über einen Schlüssel. Weiterhin muss an dieser Stelle auch nicht auf Sicherheit geachtet werden, da sich das System in einer „freundlichen“ Umgebung innerhalb der Amazon-Services befindet, die von außen abgeschottet ist.

Dynamo baut auf einem Netz vollständig gleichberechtigter Rechner auf, d. h., es gibt keine zentrale Steuerung oder Verwaltung, jeder Knoten kann jede Aufgabe wahrnehmen. Diese Architektur wurde gewählt, um die Skalierbarkeit des Systems zu gewährleisten.

Dienste wie der Shopping Cart Service (der Dienst, der den Warenkorb verwaltet) erwarten, dass auf das Speichersystem immer schreibend zugegriffen werden kann, das System hoch verfügbar ist und geringe Latenzzeiten aufweist. Da die in den ACID-Kriterien definierten Eigenschaften der Konsistenz und der hohen Verfügbarkeit gegensätzlich sind, wurde – im Gegensatz zu traditionellen Datenbanksystemen – die Konsistenz zu einer eventual consistency ‚irgendwann schließlich konsistent‘, aufgeweicht. Eine weitere Eigenschaft war, dass überwiegend kleine (weniger als 1 MB große) Dateien in Form von Schlüssel-Werte-Paaren gespeichert werden sollen. Komplizierte Datenbankanfragen müssen nicht unterstützt werden.

Um die gewünschten Eigenschaften zu erreichen, wurde eine Reihe bereits in anderem Zusammenhang bekannter Verfahren genutzt:

• Consistent Hashing
• Sloppy Quorum und Hinted Handoff
• Vector Clocks
• Anti-Entropie durch Merkle Trees
• Gossip-basiertes Protokoll

Alle Rechner sind als Ring angeordnet (zumindest logisch, physikalisch ist die Vernetzung anders). Aus jedem Key wird mittels MD5 ein Hashwert berechnet. Jedem Knoten ist nun ein bestimmter Teil des Wertebereichs des Ergebnisses der Hashfunktion zugeordnet, zu dem der jeweilige Knoten die zugehörige Datei speichert. Eine bestimmte Anzahl der im Ring nachfolgenden Knoten speichern zudem eine Kopie, wobei die Gesamtzahl der speichernden Knoten konfigurierbar ist. Um die Ausfallsicherheit zu maximieren, werden Knoten nicht nur auf unterschiedliche Rechner und Racks, sondern auch verschiedene Rechenzentren verteilt.

Ein Beispiel für den Fall von insgesamt sechs Knoten mit redundanter Speicherung auf jeweils drei Knoten (N=3) findet sich in nebenstehender Abbildung.

Da es sich um eine heterogene Systemlandschaft handelt, bei der die Speicherkapazität der eingesetzten Knotenrechner unterschiedlich sein kann (und zudem manche Dateien häufiger nachgefragt werden als andere) nutzt Dynamo sogenannte virtuelle Knoten. Das Konzept virtueller Knoten ermöglicht, dass sich auf einem physikalischen Knoten mehrere virtuelle Knoten desselben Rings befinden können. Dies ermöglicht eine bessere Auslastung bei unterschiedlichen Speicherkapazitäten der physikalischen Knoten, da durch die Gleichverteilung der Hashfunktion die Speicherauslastung für alle Knoten gleich groß ist und so einem physischen Knoten mit höherer Speicherkapazität mehrere virtuelle Knoten zugeordnet werden können.

Um die Ausfallsicherheit des Systems zu gewährleisten, wurden neben dem Parameter N (der Anzahl an Knoten, auf denen repliziert wird) noch die Parameter R (Read ‚Lesen‘) und W (Write ‚Schreiben‘) eingeführt, die ebenfalls konfigurierbar sind. Diese Parameter sind so ähnlich auch schon aus Quorumsystemen bekannt. Allerdings wurden sie hier soweit abgewandelt, dass man von sloppy ‚schlampig‘, sprechen kann. Diese legen fest, wie viele der Knoten eine Lese- oder Schreiboperation als erfolgreich melden müssen, damit die Aktion als erfolgreich gilt. In der Standardkonfiguration ist das Tupel (N, R, W) mit den Werten (3, 2, 2) belegt. Dies bedeutet, dass

• jede Datei dreimal gespeichert wird,
• ein Lesezugriff als erfolgreich gilt, sobald mindestens zwei Knoten die Datei zurückliefern und
• ein Schreibzugriff als erfolgreich gilt, sobald mindestens zwei Knoten den Zugriff als erfolgreich melden.

Die Parameter erlauben es auch einer Anwendung, das System genau für ihren Bedarf anzupassen. Beispielsweise würde eine Konfiguration von (3, 1, 3) dafür sorgen, dass man eine Art Lesepuffer realisiert hat (nur ein Knoten muss für ein Lesezugriff antworten, alle Kopien müssen immer erfolgreich geschrieben werden, da N = W), wohingegen ein System mit W = 1 für schnellstmögliche Schreibzugriffe optimiert ist. Die Konfiguration (1, 1, 1) wiederum realisiert einfach ein ganz normales (allerdings auch nicht hoch verfügbares) Dateisystem (entsprechend mit Replikation auch als (2,2,2), (3,3,3) usw.).

Falls der Koordinatorknoten (der Knoten, in dessen Bereich der Hashwert eigentlich fällt) nicht verfügbar ist, greift das sogenannte Hinted Handoff: Wenn im Beispiel der obigen Abbildung der Hashwert 3 und Knoten A nicht verfügbar wäre, so würde die Kopie stattdessen an Knoten D weitergegeben (Handoff) mit dem Vermerk (Hinted), dass diese Datei eigentlich zu Knoten A gehört. Darum speichert D diese Kopien auch in einer separaten lokalen Datenbank und fragt von Zeit zu Zeit bei A nach, ob der Knoten wieder verfügbar ist. Sobald dies der Fall ist, werden alle hinted-Kopien an A übertragen. Nach erfolgreichem Transfer kann D das hinted-Objekt löschen.

Durch die Sloppy-Quorum-Konfiguration von (3, 2, 2) kann es unter Umständen zu unterschiedlichen Versionen kommen. Da Updates auch im Hintergrund weitergegeben werden dürfen (z. B. an den dritten Knoten), kann es sein, dass nach einem erfolgreichen Schreibzugriff (der aber nur zwei Knoten erreicht hat) direkt ein Lesezugriff folgt, der nun möglicherweise zwei verschiedene Versionen zurückliefert. Um diesen Konflikt zu lösen, gibt es die sogenannten Vektoruhren, auch Vector Clocks genannt, die im Prinzip einfach nur Versionszähler sind. Jede Datei enthält einen Vektor aus Tupeln der Form (Knoten-ID, Versionsnummer), wobei bei einem Update jeder Knoten immer seine in der Datei enthaltene Versionsnummer um eins erhöht. In dem beschriebenen Problemfall würde nun der Koordinator beispielsweise für denselben Knoten einmal die Version 14 und einmal Version 15 zurückbekommen und anhand dieser Versionsnummern erkennen können, welche Version die neueste ist. Dementsprechend würde der anfragende Client auch nur die neueste Version mit der Versionsnummer 15 zurückgeliefert bekommen.

Problematisch wird es eigentlich nur, wenn der eigentliche Koordinator aus irgendeinem Grund ausgefallen ist und es gleichzeitig zu parallelem Zugriff kommt. Beispielsweise könnte sich folgender Ablauf ergeben:

1. Knoten A koordiniert ein Write ⇒ ([A,1]).
2. Knoten A koordiniert ein Write ⇒ ([A,2]).
3. Knoten A fällt aus.
4. Knoten B koordiniert ein Write ⇒ ([A,2],[B,1]). Gleichzeitig koordiniert Knoten C ein Write ⇒ ([A,2],[C,1]).
5. Knoten A ist wieder verfügbar.
6. Knoten A koordiniert ein Read und bekommt die Version ([A,2],[B,1]) und die Version ([A,2],[C,1]) zurückgeliefert.

In diesem Fall ist nicht entscheidbar, ob die Version von B oder C die neuere ist, und die Auflösung wird in die Anwendungsebene verschoben, der Client erhält beide Versionen. Im Beispiel des Shopping Cart Service würden beispielsweise beide Versionen vereinigt werden und von Knoten A die neue Version ([A,3],[B,1],[C,1]) geschrieben werden. Dies ist aber abhängig von der jeweiligen Anwendung. Sofern eine Anwendung es vorzieht, sich nicht um Fehlerauflösung zu kümmern, so gibt es auch einfache last-write-wins-Strategien vorimplementiert.

Durch das Hinted Handoff können weitere Probleme entstehen. Beispielsweise ist folgender Ablauf möglich:

1. Knoten A fällt aus, Knoten B, C und D müssen neue Replikas speichern.
2. Ein Write wird von B koordiniert, D markiert die Datei als Hinted Handoff.
3. Knoten D fällt aus.
4. Knoten A ist wieder verfügbar, bekommt aber, weil D offline ist, die Kopie nicht zurückgespielt.

Problem: A bekommt gar nicht mit, dass es eine alte Version hat und es zu dem Zeitpunkt nur N−1 Kopien gibt. Um dieses Problem zu umgehen, vergleicht A beim Neustart seine Kopien mit denen von B und C. Um allerdings den Traffic und die Rechenbelastung möglichst gering zu halten, werden dafür sogenannte Merkle Trees verwendet. Merkle Trees sind Bäume, die in ihren Blättern Hashwerte der Dateien haben, in der Wurzel einen Hash über alle Hashs und in den Knoten dazwischen entsprechende Hashs für den Teilbaum. Dadurch müssen A und B zunächst nur den Wurzelhash austauschen und können dann feststellen, ob ihre Kopien alle identisch sind oder nicht. Falls nicht, wird der Baum traversiert, bis das schuldige Blatt gefunden ist. Anschließend kann entsprechend über die Vector Clocks geschaut werden, welches die neuere Version ist, und diese entsprechend kopiert werden.

Im Fall, dass (analog zum Beispiel) die Netzwerkverbindung zu A abreißt und A das direkt gar nicht mitbekommt, wird entweder A beim nächsten Read mit Hilfe der Vector Clocks feststellen, dass eine alte Version vorliegt, oder im Rahmen der regelmäßigen Gossipnachrichten, da dort auch die Hashs der Merkle Trees übermittelt werden.

Damit bei einem temporären Ausfall eines Knotens nicht jedes Mal die gesamte Kreisstruktur neu aufgebaut werden muss, gibt es die Hinted Handoffs. Allerdings muss es auch möglich sein, Knoten dauerhaft aus dem Netz zu entfernen oder hinzuzufügen. Um dies einfach zu ermöglichen, wird per Kommandozeilentool oder Browser von einem Administrator nach Login auf einem beliebigen Knoten ein Eintrag in einer entsprechenden Konfigurationsdatei vorgenommen. Diese Änderung wird dann an alle anderen Knoten des Rings über ein Gossip-basiertes Protokoll kommuniziert. Über dieses Protokoll wird sowohl die Aufteilung der virtuellen Knoten auf den Rechnern als auch eine Liste der Rechner ständig aktuell gehalten.

Ein einfaches Beispiel für das explizite Hinzufügen von Knoten X zu Netzwerk ABCD wäre dann wie folgt:

Schritt	Aktion	Tabelle von A	Tabelle von B	Tabelle von C	Tabelle von D	Tabelle von X
1	Ausgangszustand	ABCD	ABCD	ABCD	ABCD	X
2	X wird bei A angemeldet	ABCDX	ABCD	ABCD	ABCD	ABCDX
3	A kommuniziert mit B	ABCDX	ABCDX	ABCD	ABCD	ABCDX
4	C kommuniziert mit D	ABCDX	ABCDX	ABCD	ABCD	ABCDX
5	B kommuniziert mit D	ABCDX	ABCDX	ABCD	ABCDX	ABCDX
6	A kommuniziert mit C	ABCDX	ABCDX	ABCDX	ABCDX	ABCDX
7	Endzustand erreicht	ABCDX	ABCDX	ABCDX	ABCDX	ABCDX

Die Reihenfolge bei der Kommunikation (wer sich mit wem austauscht) ist dabei zufällig und es muss sich nicht bei jeder Kommunikation eine Änderung ergeben (im Beispiel: Schritt 4).

Wird ein Knoten entfernt oder hinzugefügt, muss sich zwangsläufig auch die Aufteilung der virtuellen Knoten auf die physikalischen Rechner verändern, wofür es mehrere Verfahren gibt.^[1] Die einfachste Variante davon ist – im Fall einer nicht heterogenen Systemlandschaft – dass auf jedem physikalischen Rechner die gleiche Anzahl an gleich großen virtuellen Knoten liegen soll. Beim Entfernen eines Knotens werden somit die betreffenden virtuellen Knoten auf zufällig ausgewählte physikalische Knoten kopiert, die weniger virtuelle Knoten als der Rest des Rings besitzen. Umgekehrt übernimmt ein neu hinzukommender Knoten virtuelle Knoten von voll ausgelasteten Knoten – ebenfalls zufällig ausgewählt.

Seit 2012 wird Dynamo von Amazon Web Services als Speicherservice unter dem Namen DynamoDB angeboten. Der IaaS-Dienst unterscheidet sich jedoch in einigen Punkten von der ursprünglichen Dynamoimplementierung. Beispielsweise bietet DynamoDB ein Bigtable-ähnliche Schnittstelle, bei dem mehrdimensionale Schlüssel auf einen Wert abbilden. Damit lässt sich eine Tabellenstruktur ähnlich der einer relationalen Datenbank darstellen.

• Paper zum Thema Amazon Dynamo

1. ↑ ^{a b} Werner Vogels: Amazon's Dynamo. In: allthingsdistributed.com. 2. Oktober 2007, abgerufen am 21. März 2017.