Kai Londenberg's Weblog

Inhaltsverzeichnis Von Google lernen Google File System (GFS) Lokalität ausnutzen Googles Map/Reduce Google Bigtable Open Source Implementationen Apache Hadoop Nachteile von Hadoop Open Terracotta GlusterFS Verteilte Decision/Locking Engine Gruppenkommunikation

---

Von Google lernen

Es gibt Anwendungen, die sind einfach zu gross um ihre Daten zentral abzuspeichern, oder um ihre Berechnungen die auf diesen Daten laufen. Ein gutes Beispiel sind insbesondere Suchmaschinen wie Google.

• Wie schafft man es, wie Google, Herr über hunderte von Terabytes von Daten zu werden, diese zu analysieren und indexieren und in absehbarer Zeit Daten aus ihnen zu gewinnen ?
• Wie kann man eine Anwendung schreiben die am Anfang mit kostengünstiger Hardware auskommt, schnell und einfach zu entwickeln ist, und später das Wachstum verkraftet ?

Um zu verstehen wie man diese Dinge tun kann, lohnt es sich einen Blick auf bekannte und erfolgreiche Ansätze zur Lösung dieser Probleme die Entwicklungen zu werfen. Im Anschluss gehe ich auf kostenlose Open Source Lösungen ein, welche einem Werkzeuge an die Hand geben ebenfalls derartig skalierbare Architekturen zu gestalten.

Was Programmiersprachen betrifft, so hat dieser Artikel eine klare Tendenz, Lösungen aus der Java Welt hevorzuheben. Dennoch ist gerade auch der erste Teil "Von Google lernen" von allgemeinem Interesse.

Google ist von einem kleinen Startup mit anfänglich 20 Mitarbeitern innerhalb weniger Jahre zu einem internationalen Grosskonzern aufgestiegen. In Google arbeiten tausende hochtalentierte Entwickler daran, neue Methoden zu finden die Datenberge die Google auf seinen Servern - es müssten inzwischen einige Hunderttausend güstige Linux Maschinen sein - gespeichert hat auszuwerten und sinnvolle Informationen daraus zu gewinnen.

Von Google wurden so einige Algorithmen und Techniken publiziert die man kennen sollte um zu verstehen wie man diese Probleme angehen kann. Bei Google werden sehr, sehr viele güstige Linux Rechner eingesetzt. Die Rechner selbst bestehen nicht aus besonders hochwertigen Komponenten, und es gilt der Grundsatz dass Hardwareausfälle eher die Regel denn die Ausnahme sind. (D.h. ein gewisser Prozentsatz an Kaputten Servern wird einkalkuliert).

Google File System (GFS)

Bei Google werden sehr, sehr viele güstige Linux Rechner eingesetzt. Diese Rechner formen einen Cluster, auf dem Daten gespeichert und Berechnungen ausgefürt werden können. Dabei ist zu beachten, dass bei einigen Hunderttausen Rechnern davon auszugehen ist, dass Rechner oder auch ganze Netzsegmente ausfallen können und werden.

Um die Daten auf diesem Cluster zu speichern wurde bei Google ein eigenes Dateisystem entwickelt, welches Daten auf multiple Rechner verteilt, und einen globalen Namespace bildet, über den jede Anwendung die auf dem Google Cluster läft, und über die nötigen Rechte verfügt. Wird im GFS eine Datei gespeichert, so wird sie dabei zunächst zerstückelt, und jedes Stück wird auf eine konfigurierbare (normalerweise 3) Anzahl von Rechnern gespeichert. Dies hat den Vorteil, dass wenn Server ausfallen die Daten dennoch nicht verloren gehen. Mehr zum GFS findet sich bei Wikipedia, und im original im englischsprachigen Paper bei Google Research.

Lokalität ausnutzen

Werden diese Daten nun analysiert, z.B. indexiert, so wird diese Aufgabe ebenfalls an Server aus dem gleichen Cluster auf dem die Daten liegen delegiert. Um die Netzwerklast zu minimieren wird in einem solchen Fall versucht die Aufgabe die Daten zu analysieren einem Rechner zu übertragen der diese Daten ohnehin bereits auf seiner Festplatte vorliegen hat. Ist dies nicht möglich, wird versucht die Aufgabe zumindest einem Rechner zu übertragen, der möglichst nahe an den Daten sitzt. (z.B. im gleichen Rack wie ein Rechner der die Daten vorliegen hat.).

Googles Map/Reduce

Die Tatsache, dass Datenanalyse meistens daraus besteht, eine grosse Datenmenge zu nehmen, und eine kleine Menge an Informationen daraus zu distillieren, hat bei Google zur Entwicklung der sogenannten Map/Reduce Laufzeitumgebung geführt. Map/Reduce steht dabei für eine einfache Methode und Laufzeitumgebung für Algorithmen zur Datenanalyse- und Transformation die dann leicht auf multiplen Maschinen des Google Clusters hochgradig parallelisiert ausgeführt werden können.

Kern des ganzen ist die Idee, dass ein solches Map/Reduce Programm immer zwei Funktionen zur Verfügung stellt: eine Map-Funktion, die eine eingabedatei, oder ein Fragment einer Eingabedatei nimmt, und eine beliebige Anzahl an Name->Wert Paaren ausgibt. Dem gegenübergestellt wird eine Reduce Funktion, welche eine anhand der Namen sortierte Liste von Name->Wert Paaren (die gesammelte Ausgabe alle Map Tasks zu einem Namen) erhät, und daraus Informationen extrahiert.

Um eine Analogie zu SQL Datenbanken zu ziehen: die Map-Funktion stellt so etwas wie die "GROUP BY" Klausel in einem SQL Statement dar, während die Reduce Funktion vergleichbar mit einer Aggregate-Funktion (wie z.B. SUM(x), COUNT(x), AVG(x)) ist.

Die Map/Reduce Laufzeitumgebung hingegen übernimmt den Rest: Daten und Aufgaben werden auf den Cluster verteilt, dann werden zunächst auf unterschiedlichen Rechnern Map Tasks ausgeführt. Dann werden die Ausgaben der Map Tasks sortiert und Reduce Tasks gestartet, welche die Daten zusammenführen. Häufig dienen diese Daten dann wiederum als Ausgangsbasis für eine weitere Analyse.

Mit Map/Reduce ist es möglich, Datenanalyse von unstrukturierten (also un-indexierten) Daten hochgradig zu parallelisieren. Gleichzeitig ist das Programmiermodell extrem einfach zu verstehen und leicht anzuwenden.

Bei Google wird die Map/Reduce Laufzeitumgebung z.B. eingesetzt um den invertierten Suchindex zu erstellen. Alles in allem ist die Map/Reduce Library so beliebt, dass alleine intern bei Google hunderte, vielleicht inzwischen schon Tausende von Programmen entwickelt wurden die dieses Verfahren nutzen, um Terabytes an Daten mit Hilfe tausender von einfachen Rechnern schnell zu analysieren.

Mehr zu Map/Reduce findet sich im Original-Paper bei Google Research oder in diesem englischen Wikipedia Artikel.

Eine wichtige Kritik am Map/Reduce Algorithmus, die ihn aktueller Datenbanktechnologie gegenüberstellt, findet sich hier.

Zusammenfassend sei zu sagen, dass Map/Reduce meiner Meinung nach eine sehr sinnvolle Methode zur Analyse eines unstrukturierten, oder zumindest un-indexierten Corpus von Daten (z.B. Archive von Webseiten, Texte, Logfiles) - Es für einen wiederholten Zugriff auf die Daten jedoch Sinn macht, diese in ein sinnvoll indexiertes Format zu überführen. Eben diesen Schritt kann man allerdings gut mittels eines Map/Reduce Frameworks machen.

Google Bigtable

Google BigTable ist ein verteiltes Storage System für strukturierte Daten, das ebenfalls dazu ausgelegt ist bis zu extremen Datenmengen (hunderte bis Tausende von Terabytes) zu skalieren. Viele Google Projekte speichern Daten in Bigtable, u.A. der Google Web-Indexer (800 Terabytes), Google Earth (70 TB) und Google Analytics (200 TB).

Bigtable ist keine relationale Datenbank im herkömmlichen Sinne, ist ihr aber in vielen Dingen sehr ähnlich. Bigtable liefert keine komplexen Transaktionen (Begin .. Commit/Rollback), sondern lediglich atomare Operationen auf Zeilen und auch eine SQL Implementation sucht man vergeblich. Zusätzlich zu dieser Architektur baut Bigtable noch auf einem verteilten globalen Locking-Mechanismus namens Chubby auf.

Bei Bigtable werden Daten, vergleichbar einer normalen relationalen Datenbank, in Tabellen gespeichert. Zu jeder Zeile und jeder Spalte einer Zeile können jedoch unterschiedliche Versionen abgelegt werden, so dass man z.B. auch den Zustand einer Zeile zu einem bestimmten Zeitpunkt in der Vergangenheit abfrage kann (bis zu einem konfigurierbaren Limit).

Bigtable basiert auf GFS, und funktioniert im Grunde genommen so, dass eine Tabelle von Daten auf verschiedenste Tablet-Server verteilt werden kann. Es gibt dabei zwar einen Master-Server der die Koordination übernimmt, allerdings kommuniziert der Client im allgemeinen direkt mit den Table Servern welche die Daten ausliefern und ggf. auch Cachen. Auf diese Art bildet der Master Server keinen Bottleneck.

Das Schema eines Bigtables ist um einiges flexibler als das einer üblichen SQL Datenbank: Anstelle von Zeilen deren Namen und Typen bereits feststehen, werden bei Bigtable nur Zeilen-Familien angelegt, in denen dann zu einer Zeile der Tabelle noch beliebig viele Name->Wert Paare des Typs dieser Zeilen-Familie abgelegt werden können. Dabei ist es leicht, einer bestehenden Zeile noch einen zusätzliches Name->Wert paar einer bereits bestehenden Zeilen-Familie hinzuzufügen, aber es ist sehr aufwendig eine Schemaänderung vorzunehmen.

Mehr zu Bigtable findet sich bei Google Research

Open Source Implementationen

Für die eben vorgestellten Werkzeuge bei Google gibt es inzwischen einige Open-Source Alternativen. Auch wenn diese sicherlich noch bei weitem nicht so effektiv und ausgereift wie die bei Google im Einsatz befindlichen Originale sind, so sind sie meiner Meinung nach dennoch auf jeden Fall einen Blick wert.

Apache Hadoop

Bei Apache Hadoop handelt es sich um ein von Yahoo gesponsertes Open-Source Projekt unter der Apache Lizenz, welches (z.T. in Subprojekten) folgende Features bietet:

• Ein GFS-artiges Verteiles Dateisystem: HDFS
• Ein Map/Reduce Framework
• Eine Bigtable Implementation (HBase) in einem Subprojekt

Es gibt Berichte, wonach Yahoo seinen eigenen Indexer auf Hadoop umstellen möchte, so dass man davon ausgehen kann dass Hadoop die nötige Effizienz, Skalierbarkeit und Robustheit mitbringt um es auch als Basis für sehr grosse Projekte verwenden zu können.

Zudem gibt es bereits Erfahrungen mit mittelgrossen Clustern von über 4000 Prozessoren die mit Apache Hadoop arbeiten: M45 - Hadoop Installation mit 4000 Prozessoren

Nachteile von Hadoop

• Bei Hadoop handelt es sich um eine reine Java Software-Lösung. Das HDFS muss über eine Java Client Library angesprochen werden.
• Das Dateisystem ist nicht POSIX kompatibel
• HDFS Dateien werden einmal erzeugt, und sind danach unveränderlich. Auch anfügen ist nicht möglich
• Die Skalierbarkeit scheint sehr gut, aber die rohe Performance ist anscheinend noch optimierungswürdig

Diese Nachteile lassen sich jedoch zum Teil ausgleichen. Zunächst einmal bietet das Hadoop File System API auch Alternativen an: Man muss sich nicht mit HDFS begnügen, sondern kann seine Dateien auch in einem Kosmos File System (KFS) unterbringen, was in Zukunft vermutlich eine exzellente Option sein könnte, insbesondere da KFS auch via FUSE unter Linux als normales Dateisystem mountbar sein sollte.

Meine eigenen Experimente mit KFS zeigen momentan jedoch noch viele Problemstellen auf. KFS, insbesondere dessen Client-Code, ist zum jetzigen Zeitpunkt (3.2008) auf keinen Fall reif für einen Produktiveinsatz. Das könnte sich allerdings innerhalb weniger Monate ändern.

Zudem ist es mit Hadoop möglich, ein Interface zu Amazons S3 Storage System als Dateisystemimplementation zu verwenden.

Abgerundet wird all dies durch eine Implementation eines PhasedFilesystems, welches so funktioniert, dass Dateien zunächst lokal zwischengespeichert und modifiziert werden, um sie dann bei einem commit dauerhaft in einem der anderen System (HDFS, KFS oder S3) zu speichern.

Für Entwicklungszwecke gibt es auch eine einfache Implementation die einfach ein lokales Dateisystem verwendet.

Hadoop findet man auf den Hadoop Seiten der Apache Foundation

Open Terracotta

Open Terracotta ist ein Java Clustering Framework das es erlaubt normale Java Anwendungen mit minimalem Aufwand, und häfig ohne Änderungen im Qellcode auf einem Cluster von Rechnern zu betreiben, als würde es sich dabei nur um eine einzige Java Virtual Machine handeln.

OpenTerracotta nutzt Bytecode-Manipulation, um effektiv die Objekte und Threads von multiplen JVMs auf multiplen Rechnern miteinander kooperieren zu lassen. Wird ein Objekt als Cluster-weit markiert, erhalten danach alle Schreib- und Lesevorgänge, sowie Locking Vorgänge auf diesem Objekt eine Cluster-weite Bedeutung.

All dies erfolgt transparent für die Anwendung, und wird zudem so effizient wie möglich gehandhabt. So werden Änderungen z.B. immer erst an synchronisationsgrenzen übertragen, und Open Terracotta nutzt gezielt Möglichkeiten zur Optimierung, indem nur und erst dann Daten übertragen werden wenn dies nötig ist. Zudem bietet das System exzellente Möglichkeiten, ein System im Betrieb zu überwachen und für den Cluster-Betrieb zu optimieren.

Open Terracotta folgt einem völlig anderem Ansatz als Apache Hadoop - meiner Meinung nach ergäzen sich die beiden Systeme wunderbar. Terracotta bietet Clustering für Echtzeitanwendungen, mit minimalen Latenzzeiten bei gleichzeitig sehr guter Skalierbarkeit und der Möglichkeit, das System ausfallsicher zu gestalten. Hadoop und KFS können dies mit einem Cluster-weiten Dateisystem und einer verteilten Datenbank untermauern.

OpenTerracotta findet man hier

GlusterFS

Gerade in Kombination mit Clusteringlösungen wie OpenTerracotta benötigt man ein robustes und hochperformantes Cluster-Dateisystem. Hier muss zunächst unterschieden werden zwischen Cluster-Dateisystemen für Shared-Storage Systeme (z.B. Global FS und Oracle Cluster FS), und Netzwerk Cluster Dateisysteme für verteilte Storage.

Wenn man nicht gerade ein Storage Area Network (SAN) einsetzt, gibt es für Linux Rechner primär 2 Cluster-Dateisysteme. Zunächst einmal das ältere Lustre, welches u.A. in vielen der Top 500 Supercomputer eingesetzt wird (z.B. Blue Gene/L), und dessen Technologie vor kurzem von Sun gekauft wurde, sowie GlusterFS, ein freies Cluster Dateisystem unter GNU Lizenz.

Beide Dateisysteme sind sehr performant, skalierbar, Posix kompatibel und ausgereift. Allerdings benötigt man für Lustre einen speziell angepassten Linux Kernel, wohingegen GlusterFS als FUSE Modul komplett im Userspace läuft.

GlusterFS ist relativ unkompliziert und sehr flexibel konfigurierbar, unterstützt Replikation, Striping, Auto-Healing und viele andere Features die es für mich zur ersten Wahl machen. Eine der wenigen für mich erkennbare Nachteile im Gegensatz zu Systemen wie dem Google File System, HDFS oder KosmosFS sind die Tatsache, dass GlusterFS nicht darauf ausgelegt ist, dynamisch zu wachsen oder zu schrumpfen, Replikation-Level nicht auf Datei-Ebene festlegt werden können, und Dateien nicht automatisch neu verteilt werden, sobald neue Storage Nodes (bzw. Bricks wie sie bei GlusterFS heissen) hinzukommen.

Ein weiterer Vorteil von GlusterFS für den Einsatz als Dateisystem für Map/Reduce Operationen ist die Tatsache, dass GlusterFS auf einem beliebigen lokalen Dateisystem aufsetzt. Zugriff auf lokale Dateien ist somit auch über das lokale Dateisystem möglich, womit effiziente Map/Reduce implementationen möglich wären.

Sowohl GlusterFS als auch Lustre sind auf Systeme optimiert die via teuren high-throughput, low-latency Netzwerken (wie Infiniband 10GBit Interconnect ) miteinander verbunden sind. Allerdings sollte es auch auf 1000 oder 100 MBit Netzwerken eine gute Performance erzielen können wenn man bei der Konfiguration einige wichtige Tuning Parameter beachtet (Read-Ahead, Writebehind etc.).

Verteilte Decision/Locking Engine

Manche Dinge die auf einzelnen Systemen ganz einfach sind, werden sehr schwer, wenn man auf einmal einen Cluster aus verschiedenen Rechnern hat, bei denen (bei N-facher Redundanz) bis zu N Server ausfallen dürfen, ohne dass das Gesamtsystem zusammenbricht. Eines dieser Dinge ist die Fähigkeit, verbindliche Entscheidungen für den gesamten Cluster zu treffen. Wieder einmal kommt ein Lösungsansatz dazu von Google - und nennt sich dort "Chubby". Und wieder einmal gibt es auch für diese Technologie ein Pendant aus dem Open Source Bereich, nämlich Apache ZooKeeper, welches lose mit dem Hadoop Projekt assoziiert ist, und u.A. von HBase (der verteilten Hadoop basierten Bigtable Implementation, s.o) verwendet wird.

Gruppenkommunikation

Ein weiterer Aspekt der Programmierung für verteile Systeme ist der Aspekt der Gruppenkommunikation. Häufig hat man bei Anwendungen kein klares Client-Server Modell, sondern es geht darum dass Nachrichten, Daten oder Events von einem Teil einer Anwendung an beliebig viele Listener verteilt werden.

Zu diesem Zweck gibt es Group Communication Frameworks wie JGroups und Spread

Beide Frameworks dazu, Nachrichten mit konfigurierbaren Performance- und Sicherheitseigenschaften an eine Gruppe von Clients zu verteilen. JGroups ist dabei eine reine Java Lösung die i.A. etwas leichter zu konfigurieren ist, und Auto-Discovery bietet. Spread ist eine native Lösung mit APIs für C++, Java und Python.

Oftmals ist die Verwendung von echten Clustering Frameworks reiner Overkill, oder auch schlicht zu langsam. Mit Gruppenkommunikations-Frameworks wie JGroups oder Spread hat man eine gute Lösung wenn es z.B. darum geht, selber Clustering-Middleware wie z.B. einen verteilten Cache, replizierte Dateisysteme, replizierte Datenbanken oder ähnliches zu schreiben und sich dabei weniger Gedanken um die Sicherheit und Performance der Transportschicht machen möchte.