NoSQL-Umstellung bei RIPE NCC

Wir bei Xebia zeigen großes Interesse an der sich entwickelnden NoSQL-Gemeinschaft und all der großartigen Software und Lösungen, die daraus resultieren. Die Analyse großer Datenmengen und stark frequentierte Websites und Anwendungen werden sich durchsetzen und wir brauchen Lösungen, die damit umgehen können. Der Commodity-Stack einer relationalen Datenbank mit einem Java-App-Server darauf und dem zustandsorientierten Modell der serverseitigen Sitzungen reicht in manchen Fällen einfach nicht aus. Mit wachsendem Datenvolumen und Datenverkehr werden diese Fälle immer häufiger auftreten. In unserem App Incubator Programm sehen wir ein großes Interesse an nicht relationalen Datenbanken und zustandslosen serverseitigen Setups mit mehr Logik auf der Client-Seite (clever ausgedrückt: NoJSP). Auch bei Kunden stellt sich das Problem der ständig wachsenden Datenmengen und der fehlenden Möglichkeiten, mit den vorhandenen Tools und Datenbanken eine angemessene Analyse durchzuführen. Einer dieser Kunden ist das RIPE NCC. Die Geschichte sieht ungefähr so aus: Pro Tag kommen etwa 80 GB an Daten herein und es gibt zehn Jahre historischer Daten derselben Art und desselben Umfangs; wir müssen diese Daten abfragen und Antworten in Sekundenschnelle erhalten. Wir lösen dieses Problem mit Hadoop und HBase.

Die RIPE NCC ist eines von fünf regionalen Internet-Registern (RIRs), die Internet-Ressourcenzuweisungen, Registrierungsdienste und Koordinierungsaktivitäten anbieten, die den Betrieb des Internets weltweit unterstützen. Das RIPE NCC bietet auch Dienstleistungen für die Internet-Gemeinschaft als Ganzes an. Dazu gehört unter anderem der Routing-Informationsdienstder Internet-Routing-Daten von mehreren Standorten sammelt und speichert. mehreren Standorten rund um den Globus.   Bei RIPE NCC sind wir gerade dabei, ein bestehendes MySQL-basiertes System auf HBase als Speicher-Backend und Hadoop MapReduce als Framework für die Verarbeitung von Importaufträgen zu migrieren. In diesem Beitrag erfahren Sie mehr über die Hintergründe unserer Bemühungen, die Implementierung auf der Grundlage von Hadoop und HBase und unsere Erfahrungen mit der Verwendung von Hadoop und HBase in einem realen Szenario.   Das globale Internet ist ein großes und verteiltes System. Einer der Gründe, warum es funktioniert, ist die Tatsache, dass es eine "Shared-Nothing"-Architektur als Kernstück seines Betriebs hat. Die verschiedenen Teile des Netzwerks sind autonome Systeme, die ihr Ziel, den Datenverkehr zu Zielen in anderen Teilen des Netzwerks zu leiten, durch die Kommunikation mit ihren benachbarten Netzwerken erreichen, ohne über ihre Peers hinaus globale Kenntnisse über das Netzwerk zu erhalten. Ein globaler Überblick über den aktuellen Stand der betrieblichen Details eines solchen Systems ist sehr schwer zu erstellen und erfordert eine große Menge an Daten. Die Abteilung Information Services des RIPE NCC ist bestrebt, eine solche Übersicht zu erstellen, indem sie umfangreiche Daten über das Routing in der ganzen Welt sammelt. Ein solcher Überblick kann aus mehreren Gründen nützlich sein. Zum Beispiel, wenn im Internet mal etwas schief läuft. Zum Beispiel, als Pakistan Telecom versehentlich den gesamten Verkehr von Youtube gekapert hat, oder als Google einen beträchtlichen Teil seines Datenverkehrs an einen Anbieter in Rumänien verloren hat.   Die autonomen Systeme des Internets kommunizieren über ein Protokoll namens BGP um Informationen über das Routing auszutauschen. Unser System funktioniert, indem es BGP-Ankündigungen an verschiedenen Stellen im Internet abhört, in der Regel in der Nähe von oder an großen Internetknotenpunkten. BGP kann manchmal ein ziemlich geschwätziges Protokoll sein, vor allem wenn Router häufig ihre Meinung darüber ändern, welche Route sie für ein bestimmtes Ziel verwenden sollen. Wir sammeln derzeit etwa 80 Millionen BGP-Ankündigungen pro Tag (etwa 925 pro Sekunde). Diese Datenerfassung läuft seit fast zehn Jahren, so dass wir über eine umfangreiche Datensammlung verfügen. Die rohen, unkomprimierten Quelldaten hierfür umfassen mehrere Gigabytes pro Tag und die gesamte Historie beläuft sich auf Terabytes.   Wir betreiben derzeit ein MySQL-basiertes System in der Produktion, um diese Daten zu speichern und Abfragen zu ermöglichen. Das System hat eine Reihe von Einschränkungen:

• Er enthält nur Daten aus drei Monaten. Alles, was älter ist, wird derzeit verworfen und ist nur als Rohdaten (.tar.gz-Dateien) verfügbar, aber nicht durchsuchbar. Die Untersuchung von Ereignissen, die weniger als drei Monate zurückliegen, ist also eine sehr banale Aufgabe.
• Die Einfügegeschwindigkeit ist ein Problem. Das Einfügen in relationale Datenbanken kann aufgrund von Indexaktualisierungen und Integritätsprüfungen langsam werden. Derzeit können Daten nur geringfügig schneller eingefügt werden, als sie ankommen. Das bedeutet, dass es Wochen dauern kann, wenn der Einfügeprozess aus irgendeinem Grund hinterherhinkt, um wieder aufzuholen.
• Das Internet wächst immer noch rasant, während dieses System bereits an seiner Kapazitätsgrenze angelangt ist, es ist also nicht zukunftssicher.
• Aus demselben Grund können wir auch keine Datenerfassungspunkte hinzufügen.

  Aufgrund dieser Einschränkungen haben wir uns für eine Neuimplementierung des Speicher-Backends entschieden. Angesichts der Art der Daten entschieden wir uns für einen spaltenbasierten Speicher. Zunächst haben wir sowohl HBase als auch Cassandra untersucht. Die Entscheidung für HBase fiel aus einer Reihe von Gründen, darunter:

• Es verfügt über eine sehr gute sequentielle Lese-/Scanleistung, was für eine umfassende Datenanalyse von Vorteil ist.
• Damals gab es bereits eine anständige Dokumentation und genügend Online-Ressourcen, um mit dem Programm loszulegen.
• Diese Vorgehensweise hat sich in einer Reihe von realen Szenarien bewährt.
• Es ist von Natur aus mit Hadoop und damit MapReduce verbunden, was bei der Arbeit mit großen Datenmengen von Vorteil ist.

    [caption id="attachment_5516" align="alignnone" width="475"] Architekturübersicht[/caption] Oben sehen Sie einen Überblick über unsere Einrichtung. Sobald die Quelldaten eintreffen, verschieben wir sie nach HDFS und führen dort die gesamte Verarbeitung durch. Aus den Quelldaten erstellen wir mehrere abgeleitete Datensätze, die wir in einer Zwischenform im HDFS speichern und in HBase einfügen.   Wir benötigen die abgeleiteten Datensätze, um bestimmte Abfragen schnell beantworten zu können. Dies ist ein großer Unterschied zur MySQL-basierten Lösung. In der relationalen Welt normalisieren Sie normalerweise Ihre Daten, verteilen sie auf eine Reihe von Tabellen und wenden Joins an, wenn Sie eine Antwort auf eine bestimmte Frage benötigen. In HBase funktioniert das nicht. HBase kann nicht viel mehr als eine Schlüsselsuche nach einem bestimmten Wert oder Wertebereich durchführen. Bei diesem Modell müssen Sie wirklich im Voraus wissen, welche Art von Fragen Sie beantworten müssen und die Datenspeicherung entsprechend modellieren. Wir speichern mehrere Datensätze mit denormalisierten Antworten auf eine Reihe von spezifischen Fragen. Sie können sie als Tabellen mit vorberechneten Verknüpfungen für alle möglichen Verknüpfungskombinationen sehen. Außerdem haben alle Ihre Daten in HBase nur einen Index. Das bedeutet, dass wir für jeden zusätzlichen Index, den wir hinzufügen, die (denormalisierten) Daten effektiv duplizieren müssen.   Die Arbeit mit MapReduce hat sich als eine sehr angenehme Art der Datenverarbeitung erwiesen. Wir versuchen, so viel wie möglich in MapReduce-Aufträgen zu erledigen. Das bringt den Vorteil der Fehlertoleranz und Parallelisierung, ohne dass wir darüber nachdenken müssen. Wir weisen den Aufträgen einfach mehr Kapazität zu, wenn der Datensatz größer wird. Einem Auftrag, der einen ersten Import von Monaten oder Jahren historischer Daten durchführt, werden beispielsweise mehr Map- und Reduce-Slots zugewiesen als einem Import, der alle fünf Minuten läuft, um auf dem neuesten Stand zu bleiben. Abgesehen von der Kapazitätsplanung handelt es sich um genau denselben Code, was sehr angenehm ist. Das Einfügen in HBase wird ebenfalls als MapReduce-Auftrag modelliert, um eine hohe Parallelisierung zu erreichen.   Wir sind mit dem neuen System noch nicht in die Produktion gegangen, aber einige Schlussfolgerungen aus unseren bisherigen Erfahrungen sind:

• Die Laufzeit eines Imports von Daten eines ganzen Jahres wurde von Monaten auf Tage verkürzt.
• Die Lösung skaliert gut mit der Menge der Daten. Es macht keinen Unterschied, ob ein paar Monate oder ein paar Jahre an Daten online sind.
• Die Abfrageleistung ist mit den meisten RDBMS konkurrenzfähig (insbesondere im Vergleich zu einem Sharded-Setup). Die Abfrage von HBase für typische Abfragen liefert Ergebnisse im Sub-Sekunden-Bereich. Ergebnisse für große Abfragen werden mit weit über 15 mb/s gestreamt.
• Sie müssen nicht unbedingt Terabytes an Daten verarbeiten, damit MapReduce effektiv ist. Es geht um die Fähigkeit, entsprechend der Größe eines Auftrags zu skalieren. Es funktioniert auch bei kleineren Aufträgen und bietet dabei eine gewisse Fehlertoleranz.

  Die oben genannten Ergebnisse bedeuten im Grunde, dass HBase sein Versprechen gehalten hat. Auf unserem Entwicklungssystem mit vier Knoten führt es problemlos bis zu 300.000 Operationen pro Sekunde aus. Es löst für uns ein wichtiges Problem, das wir mit dem relationalen Backend nicht lösen konnten. Die Verwendung eines nicht-relationalen Speichersystems für Live-Abfragen gegen große Datensätze erfordert eine Menge Vorüberlegungen, aber unterm Strich würde der relationale Speicher einfach nicht in dem Maße skalieren, wie wir es brauchen.   Unser Produktionssystem wird etwas größer sein als die Entwicklungsumgebung. Der Cluster wird aus 8 Arbeitsknoten bestehen, auf denen jeweils ein Datenknoten, ein Task Tracker und ein Region Server laufen. Es gibt zwei Master-Knoten in einem Linux HA-Setup, auf denen der Name Node, der Job Tracker und der Master-Server laufen. Da wir feststellen, dass wir bei unseren Aufträgen auf dem Entwicklungscluster völlig an die IO gebunden sind, haben wir uns für eine Skalierung der IO entschieden; jeder Datenknoten hat 10 Datenfestplatten mit 600 GB (10K RPM). Die Worker haben jeweils 64 GB RAM. Ein großer Teil davon geht an HBase. Das bedeutet im Grunde, dass sich der Hotspot in unserem Datensatz größtenteils im RAM befindet, was zur Entlastung der Festplatten beiträgt.   Die Tatsache, dass wir alle unsere Daten auf HDFS und in HBase gespeichert haben, bietet viele Möglichkeiten, Dinge zu tun, die vorher schwierig oder unmöglich waren. Die Erstellung eines Auftrags, der alle historischen Daten durchgeht und statistische Informationen extrahiert, Anomalien erkennt oder Daten für interessante Diagramme oder Trends erzeugt, ist viel einfacher geworden. Wir sind bereits dabei, Dinge in dieser Richtung zu entwickeln.   Unser Setup verwendet CDH3 beta. In der Entwicklung installieren wir aus tar-Archiven. In der Produktion verwenden wir die mitgelieferten RPMs. Unsere Erfahrungen mit Hadoop und HBase sind im Allgemeinen gut. Es funktioniert größtenteils so, wie es angekündigt wurde. Wir haben eine Reihe von Lektionen gelernt:

• Es ist relativ einfach, ein Basis-Setup einzurichten und zum Laufen zu bringen. Das Tuning und die Fehlersuche sind der schwierige Teil.
• Verteiltes Debugging ist schwierig und vor allem zeitaufwendig. Stellen Sie sicher, dass Sie wissen, wo alle Ihre Protokolle hingehen (bevor Sie sie brauchen). Wenn Sie wissen, welche Informationen Sie benötigen, sollte es kein Hindernis sein, sie zu bekommen.
• Das Abstimmen von Hadoop-Jobs und HBase ist nicht trivial (genau wie das richtige Abstimmen eines RDBMS für hohe Leistung). Bereiten Sie sich darauf vor, Zeit damit zu verbringen, die Interna des Frameworks und von HBase zu erforschen, oder holen Sie sich Hilfe.

  Die oben beschriebene Arbeit ist nur ein erster Versuch. Wahrscheinlich werden wir Hadoop und HBase auch für den Import und die Analyse einiger anderer Datenquellen nutzen, die wir neben den Routing-Informationen sammeln. Außerdem wird sich die Produktions-Cluster-Umgebung mehr und mehr zu einer erstklassigen Bürger der IT-Infrastruktur von RIPE NCC zur Speicherung, Abfrage und Verarbeitung von Daten.