Gibt es in Hadoop Map-Reduce eine Klasse nach dem Sortieren und vor der Partitionierung die ganze Liste der Schlüssel?

Question

Ich verwende Hadoop, um eine sehr ungleiche Verteilung von Daten zu analysieren. Einige Schlüssel haben Tausende von Werten, aber die meisten haben nur einen Wert. Zum Beispiel würde Netzwerkverkehr, der mit IP-Adressen verknüpft ist, viele Pakete mit einigen gesprächigen IPs und nur ein paar mit den meisten IPs verbinden. Eine andere Art zu sagen ist, dass die Gini index sehr hoch ist.

Um dies effizient zu verarbeiten, sollte jeder Druckminderer entweder einige Tasten mit hohem Volumen oder viele Tasten mit geringem Volumen erhalten, um eine annähernd gleichmäßige Belastung zu erhalten. Ich weiß, wie ich das machen würde, wenn ich den Partitionierungsprozess schreiben würde: Ich würde die sortierte Liste der keys (einschließlich aller doppelten Schlüssel), die von den Mappern erzeugt wurde, sowie die Anzahl der Reduzierungen N und Put-Splits unter

nehmen

split[i] = keys[floor(i*len(keys)/N)] 

Reducer i bekommen würde Schlüssel k so dass split[i] <= k < split[i+1] für 0 <= i < N-1 und split[i] <= k für i == N-1.

Ich bin bereit, meinen eigenen Partitionierer in Java zu schreiben, aber die Partitioner<KEY,VALUE> Klasse scheint nur Zugriff auf einen Schlüssel-Wert-Datensatz zu einer Zeit zu haben, nicht die ganze Liste. Ich weiß, dass Hadoop die Datensätze sortiert, die von den Mappern erstellt wurden, daher muss diese Liste irgendwo existieren. Es könnte auf mehrere Partitionierungsknoten verteilt sein, in diesem Fall würde ich die Aufteilungsprozedur für eine der Unterlisten durchführen und das Ergebnis irgendwie an alle anderen Partitionierungsknoten kommunizieren. (Angenommen, der ausgewählte Partitionerknoten sieht eine randomisierte Teilmenge, wäre das Ergebnis immer noch annähernd ausgeglichen.) Weiß jemand, wo die sortierte Schlüsselliste gespeichert ist und wie man darauf zugreift?

Ich möchte nicht zwei Map-Reduce-Jobs schreiben, einen, um die Splits zu finden, und einen anderen, um sie tatsächlich zu verwenden, weil das verschwenderisch erscheint. (Die Mapper müssten den gleichen Job zweimal machen.) Dies scheint ein allgemeines Problem zu sein: Ungleichmäßige Verteilungen sind ziemlich üblich.

Answer 1

Soweit ich weiß, gibt es keinen einzigen Ort in der MR-Verarbeitung, wo alle Schlüssel vorhanden sind. Mehr als das - es gibt keine Garantie, dass einzelne Maschinen diese Daten speichern können. Ich denke, dieses Problem hat keine ideale Lösung im aktuellen MR-Framework. Ich denke schon, weil wir eine ideale Lösung haben - wir müssen auf das Ende des letzten Mappers warten und erst dann die Schlüsselverteilung analysieren und den Partitionierer mit diesem Wissen parametrisieren.
Dieser Ansatz wird das System erheblich komplizieren und Latenz erhöhen.
Ich denke, gute Approximation könnte sein, Stichproben über Daten zu tun, um die Idee der Schlüsselverteilung zu bekommen und dann partiotioner zu machen, um danach zu arbeiten.
Soweit ich verstehe Terasort Implementierung macht etwas sehr ähnliches: http://sortbenchmark.org/YahooHadoop.pdf

Answer 2

Ich habe über dieses Problem auch nachgedacht. Dies ist der Ansatz auf höchster Ebene, den ich nehmen würde, wenn mich jemand zwingen würde.

Zusätzlich zu der Mapper-Logik haben Sie die Möglichkeit, Ihr Geschäftsproblem zu lösen, eine Logik zu kodieren, um alle Statistiken zu sammeln, die Sie im Partitionierer benötigen, um Schlüssel-Wert-Paare ausgewogen zu verteilen. Natürlich wird jeder Mapper nur einige der Daten sehen.

Jeder Mapper kann seine Task-ID herausfinden und diese ID verwenden, um einen eindeutigen Dateinamen in einem angegebenen hdfs-Ordner zu erstellen, in dem die gesammelten Statistiken gespeichert werden. Schreiben Sie diese Datei in die cleanup() -Methode, die am Ende der Aufgabe ausgeführt wird.

Verwenden Sie die Lazy-Initialisierung im Partitionierer, um alle Dateien im angegebenen hdfs-Verzeichnis zu lesen. Dadurch erhalten Sie alle während der Mapper-Phase gesammelten Statistiken. Von dort müssen Sie die Partitionierungslogik implementieren, die Sie für die korrekte Partitionierung der Daten benötigen.

Dies alles geht davon aus, dass der Partitionierer nicht aufgerufen wird, bis alle Mapper fertig sind, aber das ist das Beste, was ich bisher konnte.