MapReduce mit Hadoop denken

Apache Hadoop verspricht "eine Softwareplattform, mit der man auf einfache Weise Anwendungen schreiben und ausführen kann, die große Datenmengen verarbeiten". Wenn Sie die Dokumentation lesen, werden Sie sicherlich Beschreibungen wie:

(Eingabe)  <k1, v1>  ->  Karte -> <k2, v2>  ->  kombinieren -> <k2, v2>  ->  reduzieren -> <k3, v3>  (Ausgabe)

sind einfach genug zu lesen und zu verstehen, aber wie wenden Sie MapReduce auf ein Problem an, das Sie in einem realen Projekt haben? Dieser Blog versucht, einen Einblick in die Anwendung von MapReduce mit Hadoop zu geben.

Was ist Hadoop?

Hadoop besteht im Wesentlichen aus zwei Dingen:

1. Ein verteiltes Dateisystem - HDFS
2. Ein MapReduce-Framework, das es Algorithmen ermöglicht, parallel auf Daten im verteilten Dateisystem zu arbeiten

Das Hadoop Distributed File System (HDFS) ist das eigentliche Herzstück von Hadoop. Es bietet Skalierbarkeit, Zuverlässigkeit und Leistung zu geringen Kosten. Das System ist so konzipiert, dass es auf handelsüblicher Hardware läuft. Obwohl das System in Java geschrieben ist, gibt es auch andere Möglichkeiten, darauf zuzugreifen und es zu nutzen. MapReduce ist ein Software-Framework, mit dem Berechnungen in einem Cluster ausgeführt werden können. Es verwendet HDFS für die Datennähe: Die Berechnungen werden verteilt und parallel auf dem Cluster ausgeführt, wobei jeder Prozess auf Daten zugreift und diese verarbeitet, die lokal auf seinem Knoten verfügbar sind. Dies sorgt für einen erheblichen Leistungsschub. Außerdem bietet das Framework Zuverlässigkeit, da Prozesse, die ausfallen, automatisch auf anderen Knoten neu gestartet werden.

Wann sollten Sie MapReduce anwenden?

MapReduce ist nur für Systeme sinnvoll, die große Datenmengen verarbeiten. Es gibt einen Overhead für das Starten von Aufgaben und es gibt immer einen Netzwerk-Overhead. Wenn wir von großen Datenmengen sprechen, meinen wir eher GBs und mehr als MBs. Ein weiterer wichtiger Aspekt ist das Datenverwendungsmuster in Ihrer Anwendung. Wenn Sie Daten 'zufällig' lesen und schreiben müssen, z.B. auf der Grundlage einer bestimmten eingehenden Anfrage, kann MapReduce Ihnen nicht helfen. MapReduce glänzt vor allem dann, wenn die Daten stapelweise gelesen werden.Die letzte Voraussetzung für die Verwendung von MapReduce ist, dass der Algorithmus als Map-and-Reduce-Prozess beschrieben werden kann. In diesem Blog möchte ich mich auf diesen letzten Aspekt konzentrieren.

Wie wendet man MapReduce an? - Ein weiteres Beispiel

Wie beschreiben Sie also einen Algorithmus auf MapReduce-Art? Um das zu veranschaulichen, eignet sich nichts besser als ein Beispiel. Wie bei jedem Beispiel, das ich für Hadoop gesehen habe, ist es ein wenig akademisch. Was ich zu erklären versuche, ist, wie sich ein MapReduce-Algorithmus von einem normalen Ansatz unterscheidet und wie man bei der Entwicklung dieses Algorithmus vorgeht. Das Wichtigste bei einem MapReduce-Algorithmus ist, dass er über < key , value > Paare die ganze Zeit über vom Eingabeformat bis zum Ausgabeformat, wenn nötig mit synthetischen Schlüsseln, begründet. Wenn Ihre Eingabe eine einfache flache Datei ist, wird sie standardmäßig an den Zeilenenden aufgebrochen und der Offset in der Datei als Schlüssel und die Zeile als Wert angegeben. Die Hauptstärke des Algorithmus liegt darin, dass er die Daten zwischen der Map- und der Reduce-Phase nach Schlüsseln sortiert. Das Framework liefert dann alle Daten mit demselben Schlüssel an dieselbe Reduziererinstanz. Jeder erfolgreiche MapReduce-Algorithmus sollte sich diesen Mechanismus zunutze machen. Das Problem - Finden von Anagrammen Nehmen wir an, Sie müssen Anagramme in einer sehr großen Eingabedatei finden. Wie würden Sie das implementieren? Ich denke, ein erster Versuch wäre eine Art Funktion wie diese:

  public static boolean isAnagram(String first, String second) {
  // Prüft, ob die beiden Eingaben Anagramme sind, indem er überprüft, ob sie alle die gleichen Zeichen enthalten.
  // Bleibt als Übung für den Benutzer...
  }

Die Anwendung müsste diese Funktion irgendwie auf alle Wortpaare in der Eingabe anwenden. So schnell diese Methode auch sein mag, die gesamte Ausführung würde immer noch eine ganze Weile dauern. Hadoopifying... Wie können Sie nun einen MapReduce-Algorithmus entwerfen, der die gewünschte Antwort liefert? Der Schlüssel liegt darin, eine Funktion zu finden, die für alle Wörter, die Anagramme sind, denselben Schlüssel liefert. Wenn Sie dies in der MapReduce-Phase anwenden, nutzen Sie die Leistungsfähigkeit des MapReduce-Frameworks, um alle Wörter, die Anagramme sind, an denselben Reducer zu liefern. Die Lösung, wenn sie gefunden ist, ist wie üblich täuschend einfach:

  public static String sortCharacters(String input) {
  char[] cs = input.toCharArray();
  Arrays.sort(cs);
  return new String(cs);
  }

Wenn Sie alle Zeichen in allen Wörtern der Eingabe sortieren, haben alle Anagramme den gleichen Schlüssel:

  angestrebt ->  Adeiprs
  Verzweiflung ->  Adeiprs

Jetzt hat die Liste der Zeichen auf der rechten Seite keine Bedeutung mehr, aber alle Anagramme werden für diese Funktion genau das gleiche Ergebnis liefern. Implementierung Sobald der Algorithmus gefunden ist, ist die Implementierung mit Hadoop recht unkompliziert und einfach (wenn auch ziemlich lang...). [java] public class AnagramFinder extends Configured implements Tool { public static class Mapper extends org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, Text, Text> { private Text sortedText = new Text(); private Text outputValue = new Text(); protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer tokenizer = new StringTokenizer(value.toString(), " tnrf,.:()!?", false); while (tokenizer.hasMoreTokens()) { String token = tokenizer.nextToken().trim().toLowerCase(); sortedText.set(sort(token)); outputValue.set(token); context.write(sortedText, outputValue); } } protected String sort(String input) { char[] cs = input.toCharArray(); Arrays.sort(cs); return new String(cs); } } public static class Combiner extends org.apache.hadoop.mapreduce.Reducer<Text, Text, Text, Text> { protected void reduzieren(Text Schlüssel, Iterable<Text> values, Context context) throws IOException, InterruptedException { Setzen Sie<Text> uniques = new HashSet<Text>(); for (Text Wert : Werte) { if (uniques.add(value)) { context.write(Schlüssel, Wert); } } } } public static class Reducer extends org.apache.hadoop.mapreduce.Reducer<Text, Text, IntWritable, Text> { private IntWritable count = new IntWritable(); private Text outputValue = new Text(); protected void reduzieren(Text Schlüssel, Iterable<Text> values, Context context) throws IOException, InterruptedException { Setzen Sie<Text> uniques = new HashSet<Text>(); int size = 0; StringBuilder builder = new StringBuilder(); for (Text Wert : Werte) { if (uniques.add(value)) { Größe++; builder.append(wert.toString()); builder.append(','); } } builder.setLength(builder.length() - 1); wenn (Größe > 1) { count.set(size); outputValue.set(builder.toString()); context.write(count, outputValue); } } } public int run(String[] args) throws Exception { Path inputPath = new Path(args[0]); Path outputPath = new Path(args[1]); Job job = new Job(getConf(), "Anagram Finder"); job.setJarByClass(AnagramFinder.class); FileInputFormat.setInputPaths(job, inputPath); FileOutputFormat.setOutputPath(job, outputPath); job.setMapperClass(Mapper.class); job.setCombinerClass(Combiner.class); job.setReducerClass(Reducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); return job.waitForCompletion(false) ? 0 : -1; } public static void main(String[] args) throws Exception { System.exit(ToolRunner.run(new Konfiguration(), new AnagramFinder(), args)); } } [/java] Die wichtigsten Teile der Implementierung sind die folgenden: Mapper - Zerlegt den Eingabetext in Token (wobei einige gängige Satzzeichen herausgefiltert werden) und wendet die Zeichensortierung an, um den gewünschten Schlüssel zu finden. Combiner (optional) - Entfernt doppelte Werte aus der Eingabe. Reducer - Sammelt Anagramme und gibt die Anzahl der Anagramme (Schlüssel) und alle verketteten Wörter (Wert) aus. Main und Run - Dieser Code konfiguriert den Auftrag für die Ausführung auf dem MapReduce-Framework. Der Combiner wird verwendet, um einige Vorverarbeitungen für den Reducer vorzunehmen. Der Hauptgrund dafür ist, dass die Ergebnisse von Mappern, die auf verschiedenen Knoten ausgeführt werden, auf demselben Knoten verarbeitet werden und somit das Netzwerk durchlaufen müssen. Um die Netzwerkbelastung zu minimieren, kann der Combiner die Anzahl der < key , value > Paare, die verarbeitet werden, reduzieren, wie hier gezeigt, indem er Duplikate herausfiltert. Der Prozess kann sich jedoch nicht darauf verlassen, dass alle < Schlüssel , Wert > Paare von einem einzigen Combiner verarbeitet werden, so dass der Reducer ebenfalls Duplikate entfernen muss. Es ist interessant zu sehen, dass das Konzept Anagramm in diesem Code nirgends vorkommt. Die Tatsache, dass der Code Anagramme findet, ergibt sich aus der Tatsache, dass alle Anagramme denselben Wert aus der Sortierfunktion haben und dieser als Schlüssel für die Map-Ausgabe verwendet wird. Das könnte für die Leser ziemlich verwirrend sein.

Fazit

Die größte Herausforderung bei Hadoop besteht darin, einen guten Algorithmus für MapReduce-Anwendungen zu finden. Der Algorithmus ist in der Regel das Ergebnis des gesamten MapReduce-Prozesses und ist möglicherweise nicht leicht aus dem Code zu verstehen. Das liegt daran, dass einige der Funktionen, die das Framework bietet, für den Algorithmus entscheidend sein können. Eine gute Dokumentation, die den gesamten Prozess beschreibt, ist unerlässlich, um dieses Problem zu lösen. Sobald ein Algorithmus entworfen ist, ist seine Implementierung in Hadoop recht einfach.