git clone https://github.com/jrings/hadoopBeispiele
Um die Skripte zu streamen, müssen sie ausführbar sein, ändert als die Rechte wie folgt:
chmod +x alpha*py weight*py
Im Ordner 1-2-Intro findet ihr den Code zu den ersten beiden Artikeln, und in 3-Pagerank die neuen Dateien, mit einem gepackten graph.zip, das ihr entpacken solltet um die Eingabedatei zu erhalten. Wenn ihr euer eigenes Beispiel testen wollt, müsst ihr es als graph.txt speichern und in das folgende Format bringen:
Knotennummer Gewicht Adjazenzliste
Die Knotenummer beginnt mit 0. Das Gewicht für eure Eingabedatei ist gleich 1/(Anzahl Knoten), und die Adjazenzliste definiert den Graph: Es ist eine Liste der Knotennummern, auf die dieser Knoten verlinkt. Der Beginn des Beispiels also etwa ist:
0 3.547e-06 6547 15408
1 3.547e-06 17793 25201 53624 54581 64929 73763 84476 98627 100192 102354 105317 105729 115925 140863 163549 164598 175798 178641 181713 190452 204188 204603 210869 213965 2
25118 241595 243293 246896 251657 252914 280934
2 3.547e-06 74360
Knoten 0 hat ausgehende Links zu Knoten 6547 und 15408, Knoten 1 verlinkt auf eine ganze Reihe Seiten und 2 nur auf eine.
Um zu verstehen, wie ich den Algorithmus parallelisiert habe, gehen wir durch die Skripte. Starten wir mit weightMapper.py:
#!/usr/bin/python
import sys
def read():
for line in sys.stdin:
yield line.strip().split()
def main():
for pp in read():
nodeId = int(pp[0])
nodeValue = float(pp[1])
nodeAdjaList = [int(p) for p in pp[2:]]
print("%i 0"% nodeId)
if len(nodeAdjaList)>0:
wD = nodeValue/float(len(nodeAdjaList))
for idn in nodeAdjaList:
print("%i%8.4g" % (idn, wD))
else:
print("-9999 %8.4g" % nodeValue)
if __name__ == '__main__':
main()
Vom WordCount-Beispiel erinnern wir uns noch dran, dass wir zeilenweise von STDIN lesen. In diesem Fall lesen wir Zeile für Zeile den Graph wie oben. Wir wissen auch, dass der Mapper Schlüssel-Wert-Paare ausgibt. Wir können uns aber nicht auf eine bestimmte Ordnung verlassen, aber darauf dass derselbe Reducer alle Zeilen mit dem gleichen Schlüssel erhalten kann.
Daher teilen wir diesem Mapper die folgende Aufgabe zu: Pro eingelesene Zeile gibt für jede Knotennummer in der Adjazenzliste aus, wieviel Gewicht dieser Knoten aus dieser Zeile erhält, denn Knotennummer in der Zeile und Adjazenzliste definieren einen Link. Und da die Zeile auch das momentane Gewicht eines Knoten enthält, ist mit der Länge der Adjazenzliste ist auch schon definiert, wieviel Gewicht an jeden Knoten weitergegeben wird. Später kann dann der Reducer aufaddieren, welches Gewicht jeder Knoten in Summe erhält.
Es ist ein wenig effektiver, das Lesen als Generatorfunktion zu schreiben (read()). Ein Generator zeichnet sich durch das yield-Schlüsselwort aus, das wie ein return funktioniert. Allerdings richtet Python es so ein, dass immer nur eine Zeile auf einmal ausgegeben wird, statt erst alle Zeilen in den Speicher zu lesen. Es erlaubt, mit for…in über die Zeilen zu iterieren, generiert neue aber immer nur “on demand”. Wichtig, wenn man mit großen Eingabemengen rechnen muss. Ich lasse den Generator Zeilen schon splitten, also eine Liste [knotenummer, gewicht, …] zurückgeben.
In der Hauptfunktion main() iteriere ich über jede Zeile, stelle Knotenummer des Ausgangsknoten, Gewicht und Adjazenzliste fest und gebe für jeden Knoten in der Adjazenzliste dessen Knotennummer (=Schlüssel) und den Anteil am Gewicht des Ausgangsknoten aus (=Wert), der weitergegeben wird. Zuerst aber schreibe ich schonmal einen Eintrag mit Gewicht 0 für jeden Knoten. Das stellt sicher, dass jeder Knoten, auch die, die keine eingehenden Links haben, im Reducerschritt und dem folgenden MapReduce auch aufgeführt sind. Und Gewicht 0 zu anderen Gewicht addieren wird nichts am Resultat ändern.
Dann gibt es aber noch ein else, falls die Adjazenzliste Länge Null hat. Das ist der Fall eines hängenden Knotens, der keine ausgehenden Links hat. Um das zu handhaben, verwende ich einen Trick und gebe das gesamte Gewicht mit Knotennummer -9999 aus (also einer die sonst nie vorkommen kann). So kann später aufaddiert werden, wieviel Gewicht aus hängenden Knoten weitergegeben werden muss.
Kommentare (1)