如何在hadoop的shuffle/sort阶段进行数字排序？

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:反向排序

示例:

使用以下代码创建标识映射器和reducer

这是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

这是Streaming命令

$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作业.

3. 需要身份缩减器,因为如果它是纯粹的仅映射作业,则无处理/排序阶段将不起作用.

Answer 2

Hadoop的默认比较器根据您使用的Writable类型(更准确地WritableComparable)比较您的键.如果您正在处理IntWritable或者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);