Anwendungen von MapReduce

Anwendungen von MapReduce
Hauptseminar: Multicore-Programmierung
Jakob Wasserthal
17. November 2011
1
Abstract
Nachdem das Framework MapReduce in der letzten Zeit stark an Popularität gewonnen hat, stellt diese Arbeit nun einige Anwendungen hierzu vor.
Nach einer allgemeinen Einführung in das Thema werden fünf Anwendungen
aus dem Bereich des Scientific Computing (Genanalyse und Verhaltenssimulation) genauer betrachtet. Es wird gezeigt, wie sich die sehr vielfältigen Problemstellungen in Map-und Reduce-Phasen umsetzen lassen und welche weiteren Anpassungen für eine effiziente Implementierung gemacht werden. Im
letzten Teil werden die Anwendungen noch anhand mehrerer ausgewählter
Kriterien verglichen, um Vor - und Nachteile der verschiedenen Ansätze
abschätzen zu können.
2
Inhaltsverzeichnis
1 Einleitung
4
2 Vorstellung MapReduce
2.1 Ablauf . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
2.2 Framework Eigenschaften . . . . . . . . . . . . . . . . . . . .
2.3 Praxistauglichkeit . . . . . . . . . . . . . . . . . . . . . . . . .
4
4
4
5
3 Anwendungen
3.1 CloudBurst . . . . . . . . . . . . . . . . .
3.1.1 Überblick . . . . . . . . . . . . . .
3.1.2 Workflow . . . . . . . . . . . . . .
3.1.3 Filterung . . . . . . . . . . . . . .
3.1.4 Speedup . . . . . . . . . . . . . . .
3.2 Contrail . . . . . . . . . . . . . . . . . . .
3.2.1 Ansatz . . . . . . . . . . . . . . . .
3.2.2 Parallelisierung . . . . . . . . . . .
3.3 Crossbow . . . . . . . . . . . . . . . . . .
3.3.1 Allgemeines . . . . . . . . . . . . .
3.3.2 Workflow . . . . . . . . . . . . . .
3.3.3 Performance . . . . . . . . . . . .
3.4 Multiple Sequence Alignment . . . . . . .
3.4.1 Problemstellung . . . . . . . . . .
3.4.2 Workflow . . . . . . . . . . . . . .
3.4.3 Laufzeit . . . . . . . . . . . . . . .
3.5 BRACE . . . . . . . . . . . . . . . . . . .
3.5.1 Problemstellung . . . . . . . . . .
3.5.2 Lösungansatz . . . . . . . . . . . .
3.5.3 State-Effect Pattern in MapReduce
3.5.4 Architektur . . . . . . . . . . . . .
3.5.5 Beispiel Fischschwarm Simulation
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
6
6
6
7
7
7
8
8
8
9
9
9
10
10
10
11
11
12
12
13
13
14
15
4 Vergleich
4.1 Speedup . . . . . . . . . . . . . . . . . . . .
4.2 Scaleup . . . . . . . . . . . . . . . . . . . .
4.3 Rechenintensität Phasen . . . . . . . . . . .
4.4 Anzahl verschiedener Mapper und Reducer
4.5 Speichergröße . . . . . . . . . . . . . . . . .
4.6 Größe des Einsatzgebietes . . . . . . . . . .
4.7 Verwendete MapReduce Implementierung .
4.8 Clustergröße . . . . . . . . . . . . . . . . . .
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
16
16
16
16
17
17
17
17
18
5 Schluss
18
3
1
Einleitung
Im Jahre 2004 stellte Google ein neues Framework namens MapReduce [1]
für die Verarbeitung von großen Datenmengen auf Clustern von StandardComputern vor. Die Besonderheit des Frameworks liegt in der einfachen
Verwendbarkeit, die es auch Nutzern ohne Erfahrung mit paralleler Programmierung ermöglicht, Programme parallel auf einem Cluster ausführen
zu lassen.
Dies führte dazu, dass MapReduce schnell an Popularität gewann und nun
in einer Vielzahl von Gebieten zum Einsatz kommt. Insbesondere im Bereich des Scientific Computing wird das Framework geschätzt, da es den
Wissenschaftlern ermöglicht ihre komplexen Berechnungen mit vergleichsweise geringem Aufwand zu parallelisieren.
Diese Arbeit beschäftigt sich nun mit einigen Anwendungen in diesem Bereich und analysiert, wie die Problemstellung der jeweiligen Anwendung auf
MapReduce umgesetzt wurde.
2
2.1
Vorstellung MapReduce
Ablauf
Um eine Anwendung für MapReduce zu erstellen, muss der Nutzer eine
Map-und eine Reduce-Funktion spezifizieren. Die Map-Funktion verarbeitet
Key/Value-Paare und erzeugt neue intermediate Paare. Diese werden nun
nach dem intermediate Key sortiert und jeder Key mit seinen zugehörigen
Values wird an die Reduce-Funktion weitergeleitet. Diese kann nun weitere
Operationen auf den Daten ausführen und schreibt die finalen Daten schließlich in Ouput Files.
Die bereits verteilt auf dem Cluster vorliegenden Eingabedaten werden in
mehrere Splits aufgeteilt und von einem Master Knoten an die verschiedenen
Worker Knoten verteilt. Diese führen draufhin die Map-Funktion auf ihrem
Teil der Daten aus. Die intermediate Daten werden genauso aufgeteilt und
an die Reduce-Tasks verteilt. Der Master ist für die Verteilung der Map-und
Reduce-Task auf die einzelnen Knoten zuständig. An Abbildung 1 lässt sich
dieser Ablauf graphisch nachvollziehen.
2.2
Framework Eigenschaften
MapReduce bietet Toleranz gegenüber dem Ausfall einzelner Knoten. Kann
ein Knoten vom Master nicht mehr erreicht werden, so werden die von dem
Knoten bearbeiteten Tasks zurück auf den Zustand idle gesetzt, sodass sie
erneut einem Knoten zugewiesen werden können.
Das MapReduce Framework verwendet ein eigenes Distributed File System,
4
Abbildung 1: MapReduce Ablaufübersicht
welches es ermöglicht die großen Mengen an Daten auf den Festplatten der
einzelnen Knoten zu verwalten. Das File System verwendet Replikate, um
Zuverlässigkeit und Verfügbarkeit auf unsicherer Hardware zu garantieren.
Die Input sowie Output Daten werden auf dem Distributed File System gespeichert. Die intermediate Daten hingegen werden lokal auf dem jeweiligen
Knoten gespeichert, der die entsprechende Map-Task ausführt. Der jeweilige
Knoten sendet die Position seiner indermediate Daten an den Master, der
diese Information dann an die Reduce Worker weiterleitet.
Um Netzwerk Bandbreite zu sparen, versucht der Master Map-Tasks auf
den Knoten zu starten, die die nötigen Input Daten bereits auf ihrer lokalen
Festplatte bereit halten. In der Praxis kann meist ein Großteil der Input
Daten lokal gelesen werden, sodass das Netzwerk nicht belastet wird.
2.3
Praxistauglichkeit
Dieses hohe Maß an Abstraktion ermöglicht es Nutzern ohne Kenntnis von
paralleler Programmierung oder verteilten Systemen Anwendungen auf großen
Clustern von Standard Hardware effizient parallel auszuführen. Der Nutzer
muss seine Anwendung nur auf das Prinzip von Map-und Reduce-Funktionen
herunterbrechen und das Framework kümmert sich eigenständig um den
Rest wie Verwaltung von Maschinenausfällen oder Kommunikation zwischen
5
den Knoten. In der Praxis hat sich gezeigt, dass sich eine große Anzahl von
Problemstellungen in MapReduce umsetzen lassen.
Google verwendet eine eigene Implementierung des Frameworks. Mittlerweile gibt es auch eine OpenSource Alternative von Apache, welche sich Hadoop
nennt und große Beliebtheit erlangt hat.
3
Anwendungen
Die ersten vier Anwendungen stammen aus dem Gebiet der Bioinformatik.
Dabei geht es um die Analyse des menschlichen Genoms. Die Maschinen zur
chemischen Analyse eines Genoms liefern Millionen kurzer Sequenzen der
DNA (je 25-250 Basenpaare lang), welche sich “Short Reads”nennen. Diese
befinden sich allerdings in einer zufälligen Reihenfolge und müssen nun wieder zu einer gesamten Sequenz zusammengebaut werden [2].
Hierzu gibt es zwei verschiedene Ansätze. Einmal das sogenannte Alignment,
bei dem es um die Anordnung beziehungsweise Zuordnung von zwei Sequenzen geht, wobei die Ähnlichkeiten maximiert werden sollen [7]. Für das
Alignment von Short Reads mit einem Referenzgenom wird auch der Begriff Mapping verwendet. Die gefundenen Ähnlichkeiten und Unterschiede
zwischen den beiden Genomen können beispielsweise Aufschluss über evolutionäre Verwandschaft geben.
Der zweite Ansatz nennt sich “Assembly”. Hierbei wird versucht die Short
Reads, mit Hilfe von Überlappungen zwischen diesen, wieder in eine sinnvolle Reihenfolge zu bringen. Dieses Verfahren erfordert einen deutlich höheren
Aufwand als das Alignment [4].
Bei der Analyse eines menschlichen Genoms müssen zwischen 3 und 4 Milliarden Short Reads verarbeitet werden. Aufgrund dieser enormen Datenmenge und der aufwendigen Rechenoperationen, die darauf durchgeführt werden,
bietet es sich an, diese Algorithmen parallel auf Clustern auszuführen. Im
Folgenden werden nun einige Anwendungen vorgestellt, die dies auf Basis
von MapReduce umzusetzen.
3.1
3.1.1
CloudBurst
Überblick
Bei CloudBurst [2] handelt es sich um ein Programm, welches einen parallelen Read mapping Algorithmus auf Basis von hadoop implementiert. Dieser
arbeitet nach dem seed-and-extend Verfahren. Hierbei werden zunächst Substrings gesucht, welche exakt zwischen Read und Referenz-Sequenz übereinstimmmen. Diese Substrings werden dann schrittweise zu längeren Anordnungen erweitert, wobei in einem gewissen Toleranzbereich Fehler oder Lücken
6
in der Übereinstimmung erlaubt werden.
3.1.2
Workflow
CloudBurst teilt sich auf in eine Map-, Shuffle- und Reduce-Phase. In der
Map-Phase werden aus den Short Reads, sowie aus der Referenzsequenz
Substrings fester Länge extrahiert. Hieraus werden nun Key/Value-Paare
(seed, MerInfo) erzeugt, wobei seed den Substring darstellt und MerInfo
zusätzliche Informationen zu diesem; beispielweise, ob der Substring aus einem Read oder der Referenz stammt und an welcher Stelle sich der Substring
in der Sequenz befindet.
In der Shuffle Phase gruppiert hadoop alle Key/Value-Paare mit dem selben
Key. Dadurch werden alle Paare aus Reads und Referenz zusammengeführt,
die den gleichen seed haben. Dies stellt die ersten exakten Übereinstimmungen
zwischen Reads und Referenzsequenz dar.
Die Reduce-Phase nimmt nun diese exakten Zuordnungen der seeds und erweitert sie zu längeren Zuordnungen, die dafür aber nicht mehr exakt sein
müssen. Es kann zu Diskrepanzen zwischen den Sequenzen oder zu Lücken
kommen. Bleiben die Fehler unter einer festgelegten Schranke, so wird die
Zuordnung des Reads zum Ouput hinzugefügt.
3.1.3
Filterung
Über den Ergebnissen ist nun eine weitere Filterung möglich, um für jeden
Read nur jeweils die beste Zuordnung - die mit der geringsten Diskrepanz
- zu behalten. Die Filterung ist als ein weiterer MapReduce Algorithmus
implementiert. Die Map-Funktion erzeugt Key/Value-Paare mit dem Read
als Key und der Zuordnung als Value. Während der Shuffle-Phase werden
für jeden Read alle erzeugten Zuordnungen gruppiert. Die Reduce-Funktion
wählt nun aus der Liste von Zuordnungen je Read die besten Zuordnungen
aus.
3.1.4
Speedup
Bei geringem Fehlertoleranzlevel in der Zuordnung fällt der Speedup durch
die Verwendung von mehr Prozessoren gering aus, da der Overhead für die
Shuffle Phase und die Verteilung über das Netzwerk gegenüber einer inmemory Auswertung auf einem einzelnem PC überwiegt. Erlaubt man aber
mehr Fehler, so dominiert die Auswertung der Zuordnungen in der ReducePhase die Laufzeit. Dadurch erhält man einen nahezu linearen Speedup mit
steigender Prozessoranzahl.
7
3.2
3.2.1
Contrail
Ansatz
Um aus den einzelnen Short Reads wieder eine ganze Sequenz zu erzeugen
verwendet Contrail [3] kein Mapping, sondern verfolgt den Ansatz des Assembly. Contrail konstruiert dazu den De Bruijn Graph aus den Short Reads.
Die Knoten in dem Graph stellen Substrings der Reads dar und gerichtete Kanten verbinden aufeinanderfolgende Substrings. Ob ein Substring auf
einen anderen folgt wird durch die Überscheidung der beiden Strings festgestellt. Hierbei kann es jedoch leicht zu Mehrdeutigkeiten kommen. Nun wird
nach Eulerwegen gesucht. Ein Eulerweg stellt in dieser Anwendung einen
Weg durch den Graphen dar, der alle Kanten mindestens ein mal enthält.
Jeder dieser Wege stellt eine mögliche Genom Rekonstruktion dar.
Abbildung 2: De Bruijn Graph Konstruktion
3.2.2
Parallelisierung
Für kleine Genome wurde dieses Verfahren bereits erfolgreich angewandt.
Bei dem deutlich größerem menschlichen Genom hatte man bisher noch
Probleme, weil der zu konstruierende Graph zu groß für den Hauptspeicher
wurde. Contrail löst dieses Problem nun, indem es Hadoop verwendet, um
den Graphen auf Basis von MapReduce in einem Cluster zu konstruieren
und zu manipulieren.
Der Graph wird als Menge von (Knoten-ID / Kanten-Adjazenzliste)-Paaren
modelliert. Die Kommunikation zwischen den Knoten läuft über Tupel der
Form (Ziel: Nachbar-Knoten-ID / Nachricht). Sowohl die Knoten als auch die
Nachrichten-Tupel werden in der Map-Phase erzeugt. In der Shuffle-Phase
werden nun alle Tupel mit gleicher Knoten-ID gruppiert. Somit erhält man
jeden Knoten mit allen Nachrichten, die an diesen gesendet wurden. In der
Reduce-Phase werden diese Nachrichten ausgewertet und ein neuer Graph
erzeugt. Nach diesem Prinzip führt Contrail seine Berechnungen auf dem
Graph durch.[4]
8
Bevor Contrail nach Eulerwegen sucht führt es mehrere Graphtransformationen durch, um die Anzahl der Knoten von über 7 Milliarden auf wenige Millionen zu reduzieren. Hierzu besitzt Contrail 24 verschiedene MapFunktionen und 22 verschiedene Reduce-Funktionen.
3.3
3.3.1
Crossbow
Allgemeines
Crossbow [5] ist ein Tool, welches die beiden Programme Bowtie und SOAPsnp kombiniert und diese parallel in einem Cluster ausführt. Crossbow basiert
auf Hadoop und nutzt dessen “streaming mode”, der es ermöglicht bereits
kompilierte Programme aus einer beliebigen Sprache parallel auf mehreren
Knoten auszuführen. Tupel werden dabei über die Standard-Ein-und Ausgabe als Tabulator-separierte Textzeilen übergeben.
Bowtie berechnet Alignments. Ähnlich wie CloudBurst nimmt Bowtie Short
Reads und versucht diese möglichst gut einem Referenzgenom zuzuordnen.
Bowtie verwendet eine qualitätsgelenkte Best-First-Search auf einem komprimierten Index, dem sogenannten FM index (full-text minute-space index) [6]. Dadurch erzielt Bowtie eine deutlich größere Geschwindigkeit als
der seed-and-extend Algorithmus von CloudBurst. Da der FM index eine
Burrows-Wheeler Kompression verwendet, braucht der Index für das gesamte menschliche Referenzgenom nur in etwa 3GB Speicher. Dieser Index
wird auf alle Knoten des Clusters verteilt.
Das Tool SOAPsnp dient dazu, SNPs (Single Nucleotide Polymorphisms Variationen einzelner Basenpaare) auf den von Bowtie zugeordneten Reads
zu finden. Im Kern basiert es auf einem Bayesschen SNP Model mit konfigurierbaren Wahrscheinlichkeiten.
An den beiden Programmen wurden kleine Änderungen vorgenommen, hauptsächlich an den Ein - und Ausgabe Formaten, um in dem MapReduce Framework zu funktionieren.
3.3.2
Workflow
In der Map-Phase findet das Alignment der Short Reads durch Bowtie statt.
Die Ausgabe der Map-Phase ist ein Stream von Tupeln, wobei jedes Tupel
als primären Schlüssel die Partition zu der es gehört und als sekundären
Schlüssel den Chromosom Offset enhält. Der Wert des Tupel enhält die angeordnete Sequenz und Qualitätswerte. In der Shuffle Phase werden die Tupel
nun nach dem Primärschlüssel gruppiert, um unterschiedliche Partitionen
parallel von verschiedenen Reducern bearbeiten zu können. Außerdem erfolgt eine Sortierung nach dem Sekundärschlüssel, da SOAPsnp sortierte
Sequenzen benötigt.
9
In der Reduce-Phase wird nun SOAPsnp ausgeführt. Die Ausgabe der ReducePhase ist ein Stream von Tupeln der gefundenen SNPs.
3.3.3
Performance
Auf einem normalen Single-Core-Rechner würde eine gleichartige Genanalyse für das menschliche Genom Wochen dauern. Mit Crossbow ist die Analyse
des Datensets in unter 3 Stunden auf einem 320-Core Cluster möglich. Durch
die Verwendung des Standard-Frameworks Hadoop, welches auf StandardHardware läuft, benötigt der Wissenschaftler auch keinen eigenen Cluster,
sondern kann Crossbow in der Cloud (z.B. in Amazon EC2) ausführen.
Crossbow erreicht keinen linearen Speedup (siehe Abblidung 3). Der Grund
dafür liegt hauptsächlich in Ungleichgewichten in der Auslastung der einzelnen CPUs.
Abbildung 3: Crossbow Speedup
3.4
3.4.1
Multiple Sequence Alignment
Problemstellung
Manche Problemstellungen der Genetik erfordern ein Alignment von mehr
als zwei DNA Sequenzen. Bei diesem sogenannten Multiple Sequence Alignment (MSA) sind Geschwindigkeit und Genauigkeit die kritischen Größen.
Es existieren auf Näherungsverfahren basierende Algorithmen, welche sehr
schnell arbeiten, dafür aber an Genauigkeit einbüßen. Mit Algorithmen, die
auf dynamischer Programmierung basieren, wie dem Needleman-Wunsch Algorithmus, können präzise Alignments berechnet werden. Allerdings erfordern diese viel Rechenaufwand und waren daher bisher nur für eine kleine
Menge kurzer Sequenzen geeignet.
Sadasivam und Baktavatchalam [7] verwenden nun das Hadoop Framework,
um den Needleman-Wunsch Algorithmus auf MapReduce umzusetzen. Dies
10
ermöglicht die parallele Ausführung von MSA auf einem Cluster und dadurch die präzise Analyse größerer Datenmengen.
Beim MSA werden die einzelnen Sequenzen jeweils untereinander verglichen
und nicht mit einem Referenzgenom.
3.4.2
Workflow
Das Verfahren läuft folgendermaßen ab: Von den n Eingabe Sequenzen werden alle möglichen n! Permutationen gebildet. Für jede Permutation kann
nun parallel das Alignment berechnet werden und am Schluss das Beste
ausgewählt werden. Je Permutation wird das Alignment auf paarweise Alignments heruntergebrochen (die paarweisen Alignments werden mit dem
Needleman-Wunsch Algorithmus berechnet). Je paarweisem Alignment ist
eine Map und eine Reduce-Task erforderlich. Man beginnt mit zwei Sequenzen und berechnet deren Alignments. Dann wird in jedem Schritt eine
Sequenz hinzugenommen und das paarweise Alignment mit jeder bereits
vorhanden Sequenz berechnet (was wieder in mehreren parallel laufenden
MapReduce-Jobs erfolgen kann). Dies führt zu n-1 Map-und Reduce-Tasks,
die seriell hintereinander ausgeführt werden.
Abbildung 4 zeigt die Berechnung der Alignments für drei Sequenzen innerhalb von einer Permutation. Im ersten Schritt werden für die ersten beiden
Sequenzen S1 und S2 die Alignments A1S1 und A1S2 berechnet. Im nächsten
Schritt werden nun die Alignments zwischen A1S1 und S3 sowie A1S2 und
S3 berechnet. Daraus erhält man die neuen Alignments A2S1 und A2S2
sowie A1S3. (Für S3 erhält man aus beiden durchgeführten paarweisen Alignments ein Ergebnis. Da diese jedoch identisch sind, können sie zu einem
A1S3 zusammengefasst werden.)
3.4.3
Laufzeit
Bei der folgenden Laufzeitberechnung wird von einem unendlich großen
Cluster ausgegangen, d.h. der Grad an Parallelisierung wird nicht durch
mangelnde Knoten begrenzt.
Die Laufzeit des Needleman-Wunsch Algorithmus für ein paarweises Alignment zweier Sequenzen der Länge N beträgt O(N ∗ N ). Die einzelnen
Alignments können wieder in b Blöcke aufgeteilt werden, die dann parallel
ausgeführt werden. Die Komplexität für ein paarweises Alignment reduziert
sich somit auf O((N ∗ N )/b). Dies führt zu einer Gesamtlaufzeit je Permutation von O((n − 1) ∗ (N ∗ N )/b). Dies ist gleichbedeutend mit der kompletten
Laufzeit, da die einzelnen Permutationen parallel ausgeführt werden können.
Die Laufzeit ohne die Parallelisierung durch MapReduce würde O(N n ) betragen.
Um die ideale Laufzeit von O((n − 1) ∗ (N ∗ N )/b) zu erhalten sind min-
11
Abbildung 4: Ablauf innerhalb einer Permutation
destens n! ∗ (n − 1) ∗ b Prozessoren nötig, legt man folgende Berechnung zu
Grunde:
Bei n Sequenzen erhält man n! Permutationen. Je Permutation gibt es n-1
Schritte, die seriell hintereinander ausgeführt werden. Allerdings ist die Anzahl an paarweisen Alignments je Schritt unterschiedlich. Im ersten Schritt
wird nur ein paarweises Alignment berechnet, bis zur Berechnung von n-1
Alignments im letzten Schritt. Um ein Maximum an Parallelität zu erhalten
muss der Schritt mit der maximalen Anzahl an Alignments betrachtet werden. Maximal müssen also je Permutation n-1 paarweise Alignments parallel
berechnet werden. Jedes Alignment kann wiederum auf b Prozessoren parallel berechnet werden. Daraus ergibt sich die ideale Anzahl an Prozessoren
von n! ∗ (n − 1) ∗ b.
3.5
3.5.1
BRACE
Problemstellung
BRACE [8] ist darauf ausgerichtet große Verhaltenssimulationen effizient
auf einem Cluster zu berechnen. Die Simulationen basieren auf einer sehr
großen Anzahl (Millionen bis Milliarden) sogenannter Agenten. Jeder dieser
Agenten entscheidet individuell, basierend auf der Beobachtung seiner Umgebung, wie er sich im nächsten Schritt verhält.
Die Schwierigkeit, diese Anwendung zu parallelisieren, besteht in dem hohen
Maß an Interaktion zwischen den Agenten und den damit verbundenen Nebenläufigkeitsproblemen. Diese liesen sich durch exzessive Synchronisation
oder etliche Rollbacks lösen, was jedoch zu einer schlechten Skalierbarkeit
12
führt.
3.5.2
Lösungansatz
BRACE löst dieses Problem durch den Einsatz von MapReduce unter Berücksichtigung zweier Prinzipien:
State-Effect Pattern: Die Simulation wird in Ticks aufgeteilt, wobei ein
Tick eine abgeschlossene Zeiteinheit darstellt. Pro Tick durchläuft jeder
Agent zwei Phasen. In der Query Phase fragt er den Zustand seiner Umwelt (State) ab und berechnet daraus Auswirkungen (Effects) auf seinen
Zustand, welchen er dann in der Update Phase aktualisiert. Innerhalb von
einem Tick muss nur die Reihenfolge der beiden Phasen eingehalten werden.
Die Aktionen je Phase sind reihenfolgeunabhängig.
Neighbourhood Property: Jeder Agent besitzt Restriktionen bezüglich
seiner Sichtbarkeit und Erreichbarkeit. Der Sichtbarkeitsbereich eines Agenten beschreibt die Menge der Agenten, von denen er lesen kann oder auf die
er Effekte haben kann. Der Agent braucht Zugriff zu dieser Region um seine Query-Phase ausführen zu können. In den meisten Simulationen ist der
Sichtbarkeitsbereich eines Agenten verhältnismässig klein, was den Kommunikationsaufwand gering hält.
Die Erreichbarkeit beschreibt die Region, in die sich der Agent innerhalb
eines Ticks bewegen kann. Sie ist somit ein Maß dafür, wie viel sich die
räumliche Verteilung innerhalb eines Ticks verändern kann.
Die Agenten werden in Partitionen aufgeteilt, sodass jedem Worker eine Partition zugeteilt werden kann. Die Vereinigung der Sichtbarkeitsbereiche aller
Agenten einer Partition stellt den Sichtbarkeitsbereich der Partition dar. Um
die Querys der Agenten einer Partition ohne Inter-Knoten Kommunikation
durchführen zu können, werden alle Agenten des Sichtbarkeitsbereiches der
Partition zusätzlich auf den Knoten kopiert.
3.5.3
State-Effect Pattern in MapReduce
Abbildung 5 zeigt, wie diese Aktionen in MapReduce umgesetzt werden. Ein
Tick t beginnt mit der ersten Map-Task mapt1 . Diese ordnet jedem Agenten eine Partition zu (distribute). Jede Partition ist einem Reduce-Worker
zugeordnet. Dieser besitzt alle Agenten des Sichtbarkeitsbereiches der ihm
zugeordneten Partition, also auch Kopien von Agenten, die Teil einer anderen Partition sind, aber im Sichtbarkeitsbereich dieser Partition liegen.
Die erste Reduce-Task reducet1 führt die Query auf der eigenen Partition
durch und berechnet daraus die Effect-Werte für die einzelnen Agenten.
13
Für Agenten aus einer anderen Partition allerdings zunächst nur auf deren
im Hauptspeicher befindlichen Kopien. In mapt2 werden diese neuen EffectWerte nun zu den Partitionen übertragen, zu denen die Agenten eigentlich
gehören. Von reducet2 werden sie dort zu den finalen Werten aggregiert.
Schließlich führt die erste Map-Task des nächsten Ticks mapt+1
die Update
1
Phase durch.
Abbildung 5: State-Effect Pattern in MapReduce
3.5.4
Architektur
BRACE verwendet keine vorhandene MapReduce Implementierung wie Hadoop, sondern stellt eine eigene zur Verfügung, um diese genau an die
Bedürfnisse von Simulationen anpassen zu können. Die BRACE Architektur
enhält folgende besonderen Eigenschaften:
Hauptspeicher: Da bei Verhaltenssimulationen das Datenvolumen verhältnismässig klein ist (nur der Zustand der Agenten muss gespeichert werden),
können die Map-und Reduce-Tasks im Hauptspeicher ausgeführt werden.
Fehlerbehandlung: Der traditionelle Weg von MapReduce um Fehlertoleranz zu erhalten besteht darin, Zwischenergebnisse im Filesystem zu speichern und fehlerhafte Task neu zu starten. Da Ticks bei BRACE aber sehr
kurz sind und sie im Hauptspeicher verarbeitet werden, würde eine Speicherung im Filesystem das System ausbremsen. Stattdessen werden vom Master
Knoten koordiniert Checkpoints angelegt, die die Zustände der Agenten im
Filesystem speichern. Dabei speichern sie jedoch nicht jeden Tick, sondern
nur jede Epoche, wobei eine Epoche eine bestimmte Anzahl an Ticks bedeutet. Tritt ein Fehler auf, so werden alle Iterationen seit dem letztem
14
Checkpoint wiederholt.
Load Balancing: Zu Beginn teilt der Master Knoten die Partitionen ein
und weist jedem Worker eine zu. Da sich mit der Zeit die Anzahl an Agenten pro Partition ändert, kann es zu Unbalanciertheiten in der Auslastung
kommen. Der Master erhält von den Workern regelmässig Auslastungsstatistiken. Darauf basierend wägt er ab, ob der Vorteil einer Repartitionierung
gegenüber den Kosten der Umverteilung überwiegt. Gegebenenfalls führt
der Master nun ein Broadcast der neuen Partitionen durch.
Lokalität: Die Map-und Reduce-Tasks für eine Partition werden immer auf
dem selben Knoten ausgeführt. Dadurch kann der Großteil der Kommunikation über Shared Memory erfolgen anstatt über das Netzwerk.
High Level Language: Neben BRACE stellen die Entwickler noch eine
High Level Simulation Language namens BRASIL zur Verfügung, die es
ermöglicht relativ einfach Simulationen für BRACE zu schreiben. Hierauf
soll aber im Rahmen dieser Arbeit nicht genauer eingegangen werden.
3.5.5
Beispiel Fischschwarm Simulation
Eine Beispielanwendung von BRACE ist die Simulation eines Fischschwarms.
Hierbei geht es um die Studie des Informationsflusses in Gruppen von Fischen, wobei kein Individuum weiß, welcher Fisch eine Nahrungsquelle kennt
(”Fisch ist informiert”).
Die Simulation verläuft in Zeitschritten, wobei jeder Fisch seine Umgebung
analysiert und daraufhin entscheidet in welche Richtung er im nächsten
Schritt schwimmt. Die Entscheidung basiert dabei auf den Prinzipien Vermeidung und Anziehung. Wenn sich ein Fisch zu nah an einem anderen
befindet, versucht er die Entfernung zu vergrößern. Wenn sich hingegen kein
anderer Fisch in einer bestimmten Entfernung befindet, wird der Fisch von
anderen Fischen angezogen. Informierte Individuen haben eine bevorzugte
Richtung (die Richtung der Nahrungsquelle). Diese wird mit dem sozialem
Einfluss aus Vermeidung und Anziehung kombiniert (unter einer bestimmten Gewichtung) und ergibt die letztendliche Bewegungsrichtung.
In einem Experiment mit zwei Klassen von informierten Individuen, die sich
jeweils in eine andere Richtung bewegen, kommt es mit der Zeit zu einer
deutlichen Änderung der räumlichen Verteilung der Fische. Ist das Load Balancing aktiviert erhält man in etwa eine lineare Skalierung des Durchsatzes
mit steigender Knotenanzahl. Ohne Load Balancing hingegen konzentrieren
sich die beiden Fischschwärme, die sich mit der Zeit bilden, auf wenige Knoten, während die Belastung der restlichen Knoten auf Null fällt. Dies hat
zur Folge, dass der Durchsatz trotz steigender Knotenanzahl konstant bleibt.
15
4
Vergleich
Leider lassen sich nicht zu allen Anwendungen entsprechende Daten finden,
um sie an jedem Kriterium zu messen. Daher tauchen Anwendungen bei
manchen Kriterien nicht auf.
4.1
Speedup
Der Speedup bezeichnet die Laufzeit-Verbesserung bei steigender Anzahl an
Prozessoren und ist somit ein Indikator dafür, wie gut das System skaliert.
• CloudBurst: nahezu linear mit zunehmenden Cores (Abhängig von Toleranzlevel)
• Crossbow: nicht linear (ungleichmässige Load Balance)
4.2
Scaleup
Beim Scaleup betrachtet man wie schnell die Prozessorzahl steigen muss,
um die Laufzeit bei steigender Datenmenge konstant zu halten.
• CloudBurst: linear
• BRACE: annähernd linear (bei aktiviertem Load Balancing)
4.3
Rechenintensität Phasen
Die Rechenintensität der Map-und Reduce-Phasen kann Auskunft darüber
geben, wie sinnvolle eine Parallelisierung ist. Wenn der Rechenaufwand nur
gering ist, besteht die Gefahr, dass der Overhead der Verteilung über das
Netzwerk und der Sortierung in der Shuffle-Phase überwiegt.
• CloudBurst
Map: gering (Extraktion von Substrings aus Input)
Reduce: hoch(seed-and-extend Algorithmus)
• Crossbow
Map: hoch (Bowtie)
Reduce: hoch (SOAPsnp)
• MSA:
Map/Reduce hoch (Needleman-Wunsch Algorithmus)
• BRACE
Map: gering (Aufteilung in Regionen)
Reduce: abhängig von Simulation
16
4.4
Anzahl verschiedener Mapper und Reducer
Die Anzahl an verschiedenen Mappern und Reducern lässt erkennen wie
komplex eine MapReduce Anwendung ist. Hierbei ist jedoch auch zu beachten, ob einzelne Mapper/Reducer iterativ öfter hintereinander ausgeführt
werden.
• CloudBurst: 2/2
• Contrail: 24/22 (Anzahl Iterationen dynamisch)
• Crossbow: 1/1
• MSA: 1/1 (Anzahl Iterationen dynamisch)
• BRACE: 2/2 (Anzahl Iterationen dynamisch)
4.5
Speichergröße
Die Short Reads des menschlichen Genoms, die den Input der Bioinformatik Anwendungen bilden, befinden sich im GigaByte Bereich. Der Input von
BRACE ist lediglich die Definition der Simulation. Interessant ist nun zu
betrachten, ob während der Ausführung viele zusätzliche Daten hinzu kommen.
• CloudBurst: gering (Anwendung erzeugt wenig zusätzliche Daten)
• Contrail: hoch (Konstruierter Graph verbraucht viel Hauptspeicher)
• Crossbow: mittel (3GB Index auf allen Knoten)
• BRACE: gering (Zustände der Agenten verbrauchen wenig Speicher)
4.6
Größe des Einsatzgebietes
Die Bioinformatik Anwendungen sind jeweils speziell auf eine Problemstellung zugeschnitten und können auch nur für diese verwendet werden.
BRACE hingegen bietet eine Plattform, auf der man mit anderen Simulationsanwendungen aufbauen kann.
4.7
Verwendete MapReduce Implementierung
Durch die Verwendung von Standard MapReduce Implementierungen (vor
allem Hadoop), lässt sich die Anwendung leicht auf einer Vielzahl von Clustern ausführen (z.B. Amazon Elastic Compute Cloud, dort kann Hadoop
bereitstellt).
17
• CloudBurst, Contrail, Crossbow, MSA: Hadoop
• BRACE: eigene spezialisierte Implementierung
4.8
Clustergröße
Im Allgemeinen findet MapReduce Anwendung auf Clustern, deren Kernzahl im zwei-bis vierstelligen Bereich liegt. Für noch größere Cluster ist
MapReduce aufgrund seiner Architektur nur schlecht geeignet.
Die folgenden Zahlen zeigen, welche Clustergröße für die jeweilige Anwendung sinnvoll ist, damit Berechnungen (auf dem menschlichem Genom) im
Stundenbereich ausgeführt werden können.
• CloudBurst, Crossbow, MSA (Alignment): 100-500 Kerne
• Contrail: > 1000 Kerne
5
Schluss
Die dargestellten Anwendungen haben gezeigt, dass sich MapReduce auf eine Vielzahl von Problemstellungen anwenden lässt. Dabei sind die Ansätze
bei der Umsetzung in Map-und Reduce-Phasen sehr vielfältig. Theoretisch
muss sich der Entwickler nicht mit den Problemen der Parallelisierung auseinandersetzen. In der Realität sollte der Entwickler dennoch einen Einblick
in die Thematik besitzen, da bei mehreren Anwendungen kleine Anpassungen nötig waren, um die Parallelität voll ausnutzen zu können. Als Beispiel
sei CloudBurst genannt, wo Teile der Daten repliziert werden, um bessere
Load Balance zu erhalten. Dennoch zeigt die hohe Verbreitung von Hadoop,
dass MapReduce die Verwendung von Clustern deutlich erleichtert und somit parallele Programmierung einem größeren Personenkreis ermöglicht.
18
Abbildungsverzeichnis
1
2
3
4
5
MapReduce Ablaufübersicht . . . . .
De Bruijn Graph Konstruktion . . .
Crossbow Speedup . . . . . . . . . .
Ablauf innerhalb einer Permutation .
State-Effect Pattern in MapReduce .
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.
5
8
10
12
14
Literatur
[1] Jeffrey Dean, Ghemawat Sanjay: MapReduce: Simplified Data Processing on Large Clusters. OSDI: Sixth Symposium on Operating System
Design and Implementation, San Francisco, CA (2004)
[2] Michael C. Schatz: CloudBurst: highly sensitive read mapping with
MapReduce. Oxford Journals Bioinformatics 25 no. 11 (2009) 1363–
1369
[3] Michael C. Schatz, Dan Sommer, David Kelley, Mihai Pop:
Contrail: Assembly of Large Genomes using Cloud Computing. (2010) URL: http://sourceforge.net/apps/mediawiki/contrailbio/index.php?title=Contrail
[4] Michael C. Schatz: High Performance Computing for DNA Sequence
Alignment and Assembly. Ph.D. Dissertation, Department of Computer
Science, University of Maryland (2010)
[5] Ben Langmead, Michael C. Schatz, Jimmy Lin, Mihai Pop, Steven L.
Salzberg: Searching for SNPs with cloud computing. Genome Biology
10 R134 (2009)
[6] P. Ferragina, G. Manzini: Opportunistic Data Structures with Applications. Proceedings of the 41st Annual Symposium on Foundations of
Computer Science, IEEE Computer Society, USA (2000) 390-398
[7] Sudha Sadasivam, G. Baktavatchalam: A Novel Approach to Multiple Sequence Alignment using Hadoop Data Grids. Proceedings of the
Workshop on Massive Data Analytics on the Cloud (2010)
[8] Guozhang Wang, Marcos Vaz Salles, Benjamin Sowell, Xun Wang, Tuan
Cao, Alan Demers, Johannes Gehrke, Walker White: Behavioral Simulations in MapReduce. Proceedings of the VLDB Endowment 3 Issue
1-2 (2010)
19