利用hadoop mapreduce 做数据排序

     我们的需求是想统计一个文件中用IK分词后每个词出现的次数，然后按照出现的次数降序排列。也就是高频词统计。

由于hadoop在reduce之后就不能对结果做什么了，所以只能分为两个job完成，第一个job统计次数，第二个job对第一个job的结果排序。 第一个job的就是hadoop最简单的例子countwords，我要说的是用hadoop对结果排序。 假设第一个job的结果输出如下：

   part-r-0000文件内容：

a    5
b    4
c    74
d    78
e    1
r    64
f    4

要做的就是按照每个词出现的次数降序排列。

   <!-- lang: lua -->
**********************************分割线*****************************************

首先可能会出现这样的问题：

1.可能上一个job为多个reduce，也就是会产生多个结果文件，因为一个reduce就会生成一个结果文件，结果存放在上一个job输出目录下类似part-r-00的文件里。

2.需要排序的文件内容很大，所以需要考虑多个reduce的情况。

   <!-- lang: lua -->
*********************************分割线*******************************

怎么去设计mapreduce

1.在map阶段按照行读取文本，然后调用map方法时把上一个job的结果颠倒，也就是map后结果应该是这样的

   <!-- lang: java -->
5    a
4    b
74    c
................
.........................
4    f

2.然后map后，hadoop会对结果进行分组，这时结果就会变成 <!-- lang: lua -->

   (5：a)
(4:b,f)
(74:c)

3.因为hadoop对数据分组后默认是按照key升序排序的，所以需要自定义排序函数将分组数据降序排序。

4.然后按照reduce数目的大小自定义分区函数，让结果形成多个区间，比如我认为大于50的应该在一个区间，一共3个reduce，那么最后的 数据应该是三个区间，大于50的直接分到第一个分区0，25到50之间的分到第二个分区1，小于25的分到第三个分区2.因为分区数和reduce数是相 同的，所以不同的分区对应不同的reduce，因为分区是从0开始的，数据分区到分区0的会被分到第一个reduce处理,分区是1的会分到第2个 reduce处理，依次类推。并且reduce对应着输出文件，所以，第一个reduce生成的文件就会是part-r-0000，第二个reduce对 应的生成文件就会是part-r-0001，依次类推，所以reduce处理时只需要把key和value再倒过来直接输出。这样最后就会形成数目最大的 字符串就会在第一个生成文件里，排好序的数据就是按照文件命名的顺序存放的。

   **其实就是利用了hadoop分组的特点，会把key相同的字符串放到一个组里，然后我们把分组的数据用自己定义的排序函数按照key排序后，再按照分区函数分到不同的reduce，固然会是第一个reduce结果文件里面是最大数字的已排序集合，也就是说需要排好序的数据时只需要依次遍历reduce的结果文件part-r-0000，part-r0001,part-r-0002...。当然，如果只有一个reduce，那就正好是一个排好序的结果文件。**

代码如下：

   <!-- lang: lua -->
*******************************分割线*****************************************

map：

   <!-- lang: java -->
/**
 * 把上一个mapreduce的结果的key和value颠倒，调到后就可以按照key排序了。
 * 
 * @author zhangdonghao
 * 
 */
    public class SortIntValueMapper extends
    Mapper<LongWritable, Text, IntWritable, Text> {
private final static IntWritable wordCount = new IntWritable(1);

private Text word = new Text();

public SortIntValueMapper() {
    super();
}

@Override
public void map(LongWritable key, Text value, Context context)
        throws IOException, InterruptedException {

    StringTokenizer tokenizer = new StringTokenizer(value.toString());
    while (tokenizer.hasMoreTokens()) {
        word.set(tokenizer.nextToken().trim());
        wordCount.set(Integer.valueOf(tokenizer.nextToken().trim()));
        context.write(wordCount, word);
    }
}
}

reudce:

   <!-- lang: java -->
/**
 * 把key和value颠倒过来输出
 * @author zhangdonghao
 * 
 */
    public class SortIntValueReduce extends
    Reducer<IntWritable, Text, Text, IntWritable> {

private Text result = new Text();

@Override
public void reduce(IntWritable key, Iterable<Text> values, Context context)
        throws IOException, InterruptedException {
    for (Text val : values) {
        result.set(val.toString());
        context.write(result, key);
    }

}
}

Partitioner:

   <!-- lang: java -->
/**
 * 按照key的大小来划分区间，当然，key 是 int值
 * 
 * @author zhangdonghao
 * 
 */
public class KeySectionPartitioner<K, V> extends Partitioner<K, V> {

public KeySectionPartitioner() {
}

@Override
public int getPartition(K key, V value, int numReduceTasks) {
    /**
     * int值的hashcode还是自己本身的数值
     */
    //这里我认为大于maxValue的就应该在第一个分区
    int maxValue = 50;
    int keySection = 0;
    // 只有传过来的key值大于maxValue 并且numReduceTasks比如大于1个才需要分区，否则直接返回0
    if (numReduceTasks > 1 && key.hashCode() < maxValue) {
        int sectionValue = maxValue / (numReduceTasks - 1);
        int count = 0;
        while ((key.hashCode() - sectionValue * count) > sectionValue) {
            count++;
        }
        keySection = numReduceTasks - 1 - count;
    }
    return keySection;
}

}

Comparator：

   <!-- lang: java -->
/**
 * int的key按照降序排列
 * 
 * @author zhangdonghao
 * 
 */
public class IntKeyDescComparator extends WritableComparator {

protected IntKeyDescComparator() {
    super(IntWritable.class, true);

}

@Override
public int compare(WritableComparable a, WritableComparable b) {
    return -super.compare(a, b);
}

}

job的关键设置：

   <!-- lang: java -->
            /**
     * 这里是map输出的key和value类型
     */
    job.setOutputKeyClass(IntWritable.class);
    job.setOutputValueClass(Text.class);

    job.setMapperClass(SortIntValueMapper.class);
    // job.setCombinerClass(WordCountReduce.class);
    job.setReducerClass(SortIntValueReduce.class);
    // key按照降序排列
    job.setSortComparatorClass(IntKeyAscComparator.class);

    job.setPartitionerClass(KeySectionPartitioner.class);

    job.setInputFormatClass(TextInputFormat.class);
    job.setOutputFormatClass(TextOutputFormat.class);
            /**
              *这里可以放输入目录数组，也就是可以把上一个job所有的结果都放进去
             **/
            FileInputFormat.setInputPaths(job, inputPath);

    FileOutputFormat.setOutputPath(job,outputPath);



http://my.oschina.net/132722/blog/168022