Wie sortiere ich in Hadoops Shuffle/Sortier-Phase?

Question

Die Daten wie folgt aussehen, erstes Feld ist eine Zahl,

3 ... 
1 ... 
2 ... 
11 ... 

Und ich mag diese Zeilen sortieren, nach dem ersten Feld numerisch und nicht alphabetisch von, die nach dem Sortieren bedeutet, es sollte so aussehen,

1 ... 
2 ... 
3 ... 
11 ... 

Aber hadoop hält mich zu geben, diese

1 ... 
11 ... 
2 ... 
3 ... 

Wie behebe?

Answer 1

Angenommen, Sie Hadoop Streaming verwenden, müssen Sie die KeyFieldBasedComparator Klasse verwenden.

1. -D mapred.output.key.comparator.class = org.apache.hadoop.mapred.lib.KeyFieldBasedComparator

2. auf Streaming-Befehl hinzugefügt werden, sollten Sie müssen mit Art schaffen, die Sortierung erforderlich mapred.text.key.comparator.options. Einige nützlichsten sind -n: numerische Sortierung, -r: reverse sort

Beispiel:

Eine Identität Mapper und Minderer mit dem folgenden Code

Dies ist der Mapper. py & reducer.py

#!/usr/bin/env python 
import sys 
for line in sys.stdin:  
    print "%s" % (line.strip()) 

Dies ist der Eingang.txt

1 
11 
2 
20 
7 
3 
40 

Dies ist der Streaming Befehl

$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar 
-D mapred.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator 
-D mapred.text.key.comparator.options=-n 
-input /user/input.txt 
-output /user/output.txt 
-file ~/mapper.py 
-mapper ~/mapper.py 
-file ~/reducer.py 
-reducer ~/reducer.py 

Und Sie werden die erforderliche Leistung

1 
2 
3 
7 
11 
20 
40 

HINWEIS erhalten:

1. Ich habe einen einfachen Ein-Tasten-Eingang verwendet. Wenn Sie jedoch mehrere Schlüssel und/oder Partitionen haben, müssen Sie mapred.text.key.comparator.options nach Bedarf bearbeiten. Da ich Ihren Anwendungsfall nicht kenne, ist mein Beispiel darauf beschränkt

2. Identity Mapper wird benötigt, da Sie mindestens einen Mapper für einen MR-Job benötigen, um zu laufen.

3. Der Identity Reducer wird benötigt, da die Shuffle/Sort-Phase nicht funktioniert, wenn es sich um einen reinen Map-Job handelt.

Answer 2

Der Standardkomparator von Hadoop vergleicht Ihre Tasten basierend auf dem Typ Writable (genauer WritableComparable), den Sie verwenden. Wenn Sie mit oder LongWritable zu tun haben, dann sortiert sie numerisch.

Ich nehme an, Sie verwenden in Ihrem Beispiel daher werden Sie am Ende natürliche Sortierreihenfolge haben.

In besonderen Fällen können Sie jedoch auch einen eigenen Vergleicher schreiben.
ZB: für Testzwecke nur, hier eine kurze Probe, wie die Sortierreihenfolge von Text Schlüsseln ändern: das sich als ganze Zahlen behandeln und numerische Sortierreihenfolge erzeugen:

public class MyComparator extends WritableComparator { 

     public MyComparator() { 
      super(Text.class); 
     } 

     @Override 
     public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { 

      try { 

       String v1 = Text.decode(b1, s1, l1); 
       String v2 = Text.decode(b2, s2, l2); 

       int v1Int = Integer.valueOf(v1.trim()); 
       int v2Int = Integer.valueOf(v2.trim()); 

       return (v1Int < v2Int) ? -1 : ((v1Int > v2Int) ? 1 : 0); 

      } 
      catch (IOException e) { 
       throw new IllegalArgumentException(e); 
      } 
     } 
    } 

Im jobrunner Klassensatz :

Job job = new Job(); 
... 
job.setSortComparatorClass(MyComparator.class);