sorting - Hadoopのシャッフル/ソート段階で数値的にソートする方法は?

Question

データは次のようになります。最初のフィールドは数値です。

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

そして、これらの行を最初のフィールドに従ってアルファベット順ではなく数値で並べ替えたいと思います。つまり、並べ替え後は次のようになります。

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

しかし、hadoop は私にこれを与え続けます。

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

どのように修正しますか？

Answer 1

Hadoop Streamingを使用していると仮定すると、 KeyFieldBasedComparatorクラスを使用する必要があります。

1. -D mapred.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator をストリーミング コマンドに追加する必要があります

2. mapred.text.key.comparator.options を使用して、必要な並べ替えの種類を指定する必要があります。-n : 数値ソート、 -r : 逆ソート

例:

次のコードを使用して ID マッパーとリデューサーを作成します

これはmapper.pyとreducer.pyです

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

これはinput.txtです

1
11
2
20
7
3
40

これがストリーミングコマンドです。

$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

そして、必要な出力が得られます

1   
2   
3   
7   
11  
20  
40

注:

1. シンプルなワンキー入力を使用しました。ただし、複数のキーやパーティションがある場合は、必要に応じて mapred.text.key.comparator.options を編集する必要があります。私はあなたのユースケースを知らないので、私の例はこれに限定されています

2. MR ジョブを実行するには少なくとも 1 つのマッパーが必要になるため、ID マッパーが必要です。

3. 純粋なマップのみのジョブの場合、シャッフル/ソート フェーズが機能しないため、Identity Reducer が必要です。

Answer 2

WritableHadoop のデフォルト コンパレータは、使用するタイプ (より正確には )に基づいてキーを比較しますWritableComparable。IntWritableorを扱っている場合は、LongWritableそれらを数値順に並べ替えます。

私はあなたがあなたの例で使用していると仮定しているので、自然なソート順Textになるでしょう。

ただし、特殊なケースでは、独自のコンパレータを作成することもできます。
例:テストのみを目的として、テキスト キーの並べ替え順序を変更する簡単なサンプルを次に示します。これにより、キーが整数として扱われ、数値による並べ替え順序が生成されます。

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);
            }
        }
    }

jobrunner クラス セット内:

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

Answer 3

順序付き Hadoop (構成-jobconfの代わりに使用される可能性があり-Dます) を使用したスト​​リーミングの場合、キーで並べ替えることができます。

-jobconf stream.num.map.output.key.fields=2\
-jobconf mapreduce.partition.keycomparator.options="-k2,2nr"\
-jobconf mapred.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator

によってstream.num.map.output.key.fields、1 番目と 2 番目の列はkey 1とkey 2です。

mapreduce.partition.keycomparator.options="-k2,2nr"2番目のキー（2番目から2番目のキーまで）を数値として逆順にソートすることを意味します。

これは、Linuxsortコマンドによく似ています。