# MapReduce

MapReduce ist ein vom Unternehmen Google Inc. eingeführtes Programmiermodell für nebenläufige Berechnungen über (mehrere Petabyte) große Datenmengen auf Computerclustern. MapReduce ist auch der Name einer Implementierung des Programmiermodells in Form einer Software-Bibliothek. Beim MapReduce-Verfahren werden die Daten in drei Phasen verarbeitet (Map, Shuffle, Reduce), von denen zwei durch den Anwender spezifiziert werden (Map und Reduce). Dadurch lassen sich Berechnungen parallelisieren und auf mehrere Rechner verteilen. Bei sehr großen Datenmengen ist die Parallelisierung unter Umständen schon deshalb erforderlich, weil die Datenmengen für einen einzelnen Prozess (und das ausführende Rechnersystem) zu groß sind.
Das Programmiermodell wurde durch die in der funktionalen Programmierung häufig verwendeten Funktionen map und reduce inspiriert, auch wenn die Arbeitsweise der Bibliothek davon abweicht. 2010 wurde für MapReduce ein US-Patent erteilt. Der wesentliche Beitrag von MapReduce ist jedoch das zu Grunde liegende System, das die Berechnungen stark parallelisiert, die Reorganisation der Daten im Shuffle-Schritt optimiert, und automatisch auf Fehler im Cluster reagieren kann, wie beispielsweise den Ausfall von kompletten Knoten.

## Arbeitsweise
### Illustration des Datenflusses

Das obige Bild illustriert den Datenfluss bei der MapReduce-Berechnung.
* Map-Phase:
** Die Eingabedaten (D, A, T, A) werden auf eine Menge von Map-Prozessen verteilt (illustriert durch bunte Rechtecke), welche jeweils die vom Nutzer bereitgestellte Map-Funktion berechnen.
** Die Map-Prozesse werden idealerweise parallel ausgeführt.
** Jede dieser Map-Instanzen legt Zwischenergebnisse ab (illustriert durch pinkfarbene Sterne).
** Von jeder Map-Instanz fließen Daten in eventuell verschiedene Zwischenergebnisspeicher.
* Shuffle-Phase:
** Die Zwischenergebnisse werden gemäß den Ausgabeschlüsseln, die von der Map-Funktion produziert wurden, neu verteilt, sodass alle Zwischenergebnisse mit demselben Schlüssel im nächsten Schritt auf demselben Computersystem verarbeitet werden.
* Reduce-Phase:
** Für jeden Satz an Zwischenergebnissen berechnet jeweils genau ein Reduce-Prozess (illustriert durch violette Rechtecke) die vom Nutzer bereitgestellte Reduce-Funktion und damit die Ausgabedaten (illustriert durch violette Kreise X, Y und Z).
** Die Reduce-Prozesse werden idealerweise ebenfalls parallel ausgeführt.

### Definition der MapReduce-Funktion

Die MapReduce-Bibliothek realisiert eine Funktion, welche aus einer Liste von Schlüssel-Wert-Paaren (Eingabeliste) eine neue Liste von Schlüssel-Wert-Paaren (Ausgabeliste) berechnet:
    ${\begin{aligned}\mathrm {MapReduce} :(K\times V)^{*}&\to (L\times W)^{*}\\{\lbrack (k_{1},v_{1}),\ldots ,(k_{n},v_{n})\rbrack }&\mapsto \lbrack (l_{1},w_{1}),\ldots ,(l_{m},w_{m})\rbrack \end{aligned}}$
Erläuterung:
* Die Mengen $K$ und $L$ enthalten Schlüssel, die Mengen $V$ und $W$ enthalten Werte.
* Alle Schlüssel $k\in K$ sind vom gleichen Typ, z. B. Strings.
* Alle Schlüssel $l\in L$ sind vom gleichen Typ, z. B. ganze Zahlen.
* Alle Werte $v\in V$ sind vom gleichen Typ, z. B. Atome.
* Alle Werte $w\in W$ sind vom gleichen Typ, z. B. Gleitkommazahlen.
* Wenn $A$ und $B$ Mengen sind, so ist mit $A\times B$ die Menge aller Paare $(a,b)$ gemeint, wobei $a\in A$ und $b\in B$ (kartesisches Produkt).
* Wenn $M$ eine Menge ist, so ist mit $M^{*}$ die Menge aller endlichen Listen mit Elementen aus $M$ gemeint (angelehnt an den Kleene-Stern) – die Liste kann auch leer sein.

### Definition der Map- und Reduce-Funktionen

Der Nutzer konfiguriert die Bibliothek über die Bereitstellung der beiden Funktionen Map und Reduce, die wie folgt definiert sind:
    ${\begin{aligned}\mathrm {Map} :K\times V&\to (L\times W)^{*}\\(k,v)&\mapsto \lbrack (l_{1},x_{1}),\ldots ,(l_{r_{k}},x_{r_{k}})\rbrack \end{aligned}}$
bzw.
    ${\begin{aligned}\mathrm {Reduce} :L\times W^{*}&\to X^{*}\\\left(l,{\lbrack y_{1},\ldots ,y_{s_{l}}\rbrack }\right)&\mapsto \lbrack w_{1},\ldots ,w_{m_{l}}\rbrack \end{aligned}}$

### Map-Phase

* Map bildet ein Paar, bestehend aus einem Schlüssel $k$ und einem Wert $v$, auf eine Liste von neuen Paaren $(l_{r},x_{r})$ ab, welche die Rolle von Zwischenergebnissen spielen. Die Werte $x_{r}$ sind vom gleichen Typ wie die Endergebnisse $w_{i}$.
* Bei einem neuen Paar $(l,x)$ verweist der von Map vergebene Schlüssel $l$ dabei auf eine Liste $T_{l}$ von Zwischenergebnissen, in welcher der von Map berechnete Wert $x$ gesammelt wird.
* Die Bibliothek ruft für jedes Paar in der Eingabeliste die Funktion Map auf.
* All diese Map-Berechnungen sind voneinander unabhängig, so dass man sie nebenläufig und verteilt auf einem Computercluster ausführen kann.

### Shuffle-Phase

* Bevor die Reduce-Phase starten kann, müssen die Ergebnisse der Map-Phase nach ihrem neuen Schlüssel $l$ in Listen $T_{l}$ gruppiert werden.
* Wenn Map- und Reduce-Funktionen nebenläufig und verteilt ausgeführt werden, wird hierfür ein koordinierter Datenaustausch notwendig.
* Die Performanz eines Map-Reduce-Systems hängt maßgeblich davon ab, wie effizient die Shuffle-Phase implementiert ist.
* Der Nutzer wird in der Regel nur über die Gestaltung der Schlüssel $l$ auf die Shuffle-Phase Einfluss nehmen. Daher reicht es, sie einmalig gut zu optimieren, und zahlreiche Anwendungen können hiervon profitieren.

### Reduce-Phase

* Sind alle Map-Aufrufe erfolgt bzw. liegen alle Zwischenergebnisse in $T_{l}$ vor, so ruft die Bibliothek für jede Zwischenwertliste die Funktion Reduce auf, welche daraus eine Liste von Ergebniswerten $w_{j}$ berechnet, die von der Bibliothek in der Ausgabeliste als Paare $(l,w_{j})$ gesammelt werden.
* Auch die Aufrufe von Reduce können unabhängig auf verschiedene Prozesse im Computercluster verteilt werden.
Anmerkung: Diese Darstellung war etwas vereinfacht, denn in der Regel wird die Steuerung des MapReduce Verfahrens eine Anzahl $R$ von Reduce-Prozessen anstreben, so dass, wenn es für mehr als $R$ verschiedene Schlüssel $l$ Zwischenergebnisse $(l,x)$ gibt, Zwischenergebnisse $(l,x)$ mit verschiedenen Schlüsseln $l$ in einer gemeinsamen Liste gespeichert werden. Die entsprechenden Paare werden vor der Reduce-Berechnung nach Schlüsseln sortiert.

### Combine-Phase

Optional kann vor der Shuffle-Phase noch eine Combine-Phase erfolgen. Diese hat in der Regel die gleiche Funktionalität wie die Reducefunktion, wird aber auf dem gleichen Knoten wie die Map-Phase ausgeführt. Dabei geht es darum, die Datenmenge, die in der Shuffle-Phase verarbeitet werden muss, und damit die Netzwerklast zu reduzieren. Der Sinn der Combine-Phase erschließt sich sofort bei der Betrachtung des Wordcount-Beispiels: Auf Grund der unterschiedlichen Häufigkeit von Wörtern in natürlicher Sprache, würde bei einem deutschen Text beispielsweise sehr oft eine Ausgabe der Form ("und", 1) erzeugt (gleiches gilt für Artikel und Hilfsverben). Durch die Combine-Phase wird nun aus 100 Nachrichten der Form ("und", 1) lediglich eine Nachricht der Form ("und", 100). Dies kann die Netzwerkbelastung signifikant reduzieren, ist aber nicht in allen Anwendungsfällen möglich.

## Beispiel: Verteilte Häufigkeitsanalyse mit MapReduce
### Problem

Man möchte für umfangreiche Texte herausfinden, wie oft welche Wörter vorkommen.

### Map-Phase

* Map bekommt jeweils einen Dokumentnamen name und ein Dokument document als Zeichenkette übergeben.
* Map durchläuft das Dokument Wort für Wort.
* Jedes Mal, wenn ein Wort w angetroffen wird, wandert eine 1 in die w-Zwischenergebnisliste (falls diese noch nicht existiert, wird sie angelegt).
* Ist man mit allen Wörtern durch und hat der Text insgesamt n verschiedene Wörter, so endet die Map-Phase mit n Zwischenergebnislisten, jede für ein anderes Wort sammelnd, welche so viele 1-Einträge enthält, wie das entsprechende Wort im Dokument gefunden wurde.
* Eventuell liefen viele Map-Instanzen gleichzeitig, falls der Bibliothek mehrere Wörter und Dokumente übergeben wurden.

### Shuffle-Phase

* Die Zwischenergebnislisten von mehreren Prozessen / Systemen für das gleiche Wort w werden zusammengefasst, und auf die Systeme für die Reducer verteilt.

### Reduce-Phase

* Reduce wird für das Wort word und die Zwischenergebnisliste partialCounts aufgerufen.
* Reduce durchläuft die Zwischenergebnisliste und addiert alle gefundenen Zahlen auf.
* Die Summe result wird an die Bibliothek zurückgegeben, sie enthält, wie oft das Wort word in allen Dokumenten gefunden wurde.
* Die Zwischenergebnisse konnten parallel, durch gleichzeitige Reduce-Aufrufe, berechnet werden.

### Insgesamt

* Aus einer Liste von Dokumentnamen und Dokumenten wird eine Liste von Worten und Worthäufigkeiten generiert.

### Beispielhafte Berechnung

Zum Beispiel wäre folgende Berechnung auf einem klassischen Text denkbar:
Der Text wird in Sätze aufgeteilt, dabei bietet sich eine Normalisierung an, indem man alles klein schreibt und die Satzzeichen entfernt:
Die Eingabeliste hat drei Paare als Elemente, wir können daher drei Map-Prozesse starten:
Die Map-Aufrufe generieren diese Zwischenergebnispaare:
Die Map-Prozesse liefern ihre Paare an die MapReduce-Bibliothek, welche diese in den Zwischenergebnislisten sammelt. Parallel könnte folgendes geschehen (Die gleiche Taktung der 3 Map-Prozesse ist unrealistisch, tatsächlich überlappen sich die Ausführungen. Die T_wort-Listen sind lokal pro Map-Prozess vorhanden und werden nicht zwischen den Schritten synchronisiert):
Im vierten Schritt sieht man, dass Zwischenergebnislisten lokal für jeden Map-Prozess existieren und nicht global wiederverwendet werden können:
Im siebten Schritt kommt dann zum ersten Mal vor, dass ein weiteres Vorkommen in einer bereits angelegten Zwischenergebnisliste gesammelt wird:
usw.
Nach 21 Schritten sind alle drei Map-Prozesse mit ihrer Arbeit fertig, die Map-Phase endet und es beginnt die Reduce-Phase. Die Zwischenergebnislisten, die von verschiedenen Map-Prozessen zu demselben Wort angelegt wurden, werden zusammengefügt. Für jede der entstandenen Zwischenergebnislisten (hier sortiert aufgeführt)
können wir parallel einen Reduce-Prozess starten, der jeweils die Elemente aufzählt. Das Ergebnis von MapReduce sieht in etwa so aus:

## Weitere Beispiele

| Verfahren | Map-Funktion | Reduce-Funktion |
| Verteiltes grep | Gibt die gefundene Zeile (hit) in einen Zwischenergebnisspeicher | Reicht durch (Identische Abbildung, genauer: Projektion auf die 2. Komponente) |
| Umsatzauswertung | Schreibt für jeden Beleg die Artikelnummer und den Betrag in einen Zwischenspeicher | Addiert für jede unterschiedliche Artikelnummer die Beträge zusammen |
| Datenbanksystem | Liest, filtert und verarbeitet Teilmengen von Datensätzen | Führt Aggregatfunktionen aus |

## Verallgemeinerung

Nachdem das Verfahren 2014 bereits zehn Jahre alt ist, bietet Google seit kurzem eine Erweiterung Cloud Dataflow an, die größere Flexibilität bietet und das Cloud Computing noch stärker vorantreiben soll.