第八篇：经典案例

前言

       在计算机领域，排序的重要性不用多说。而排序的算法，效率分析等也一直是研究的热点。

       本文将给出使用Hadoop分布式方案进行排序的例子，这能极大提高排序的速度，是需要重点掌握的一个案例。

需求

       对输入文件中的数据进行排序。

       输入文件中的每行内容都是一个数字，要求在输出文件中每行有两个数字，第一个数字代表位次，第二个数字为原始数据。

       比如文件1包含以下数据：

       1

       3

       5

       2

       4

       6

       文件2包含以下数据：

       2

       4

       6

       3

       1

       5

       那么输出文件应当为：

       1　　1

       2　　1

       3　　2

       4　　2

       ...

方案制定

       表面上看，这是一个非常简单的例子 - Hadoop中存放的键值对本身就是有序的，直接将输入存放进来然后再取出来就完成排序了。

       但事实上，直接这样做行不通。为何？因为默认的排序过程是在单个的节点上完成的。也就是说，每个reduce节点收到键值对是在该节点局部有序，而不是在所有reduce节点里全局有序。

       解决之道是重写Partition方法，请仔细阅读以下内容：

       在shuffle阶段之后(或者说是shuffle最后)，将根据map中间输出键值对中的key值来决定将此键值对划分给哪个Partition区间，或者说哪个reduce节点。

       可以根据数据的最大最小值将数据划分为多个区间，这样，每个reduce节点就能获取到某个数据段的完整的数据，而且根据hadoop特性，这些数据在单个的reduce节点之内都是有序存放的。

       因此每个reduce节点的任务很简单，输出结果就可以了。

       至于说位次，只需要在reduce类中声明一个static变量，让这个static变量在不同的reduce调用之间共享就可以了。

       要说明的是这里统计的只是数据在每个reduce节点之内的位次，如果要获得全局位次，则需要再遍历一次所有reduce输出文件。时间复杂度仅为O(n)。

代码实现

package org.apache.hadoop.examples;

import java.io.IOException;

//导入各种Hadoop包
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

// 主类
public class Sort {
        
    // Mapper类
    public static class Map extends Mapper<Object, Text, IntWritable, IntWritable>{
        
        // new一个值为1的IntWritable对象
        private static IntWritable data = new IntWritable(1);
                
        // 实现map函数
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            
            // 将切分后的value作为中间输出的key，然后value值为1。
            String line = value.toString();
            data.set(Integer.parseInt(line));
            context.write(data, new IntWritable(1));
        }
    }
        
    // Reducer类
    public static class Reduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
    
        // new一个值为空的IntWritable对象
        private static IntWritable linenum = new IntWritable();
                
        // 实现reduce函数
        public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
                
            // 写入结果键值对
            for (IntWritable val : values) {
                context.write(linenum, key);
                linenum = new IntWritable(linenum.get()+1);
            }
        }
    }

    // 重写Partitioner类
    public static class Partition extends Partitioner <IntWritable, IntWritable> {
        
        // 重载getPartition方法。下面的三个参数分别为map中间输出的键，值，以及分割区间的个数。
        public int getPartition(IntWritable key, IntWritable value, int numPartitions) {
            
            // 依次将键值对分配到各个分割区间
            int MaxNumber = 65223;
            int bound = MaxNumber/numPartitions + 1;
            int keynumber = key.get();
            
            for (int i=0; i<numPartitions; i++) {
                if (keynumber < bound * (i+1) && keynumber >= bound*i) {
                    
                    // 返回的 i 就是分配到的区间号
                    return i;
                }
            }
            
            return -1;
        }
    }
    
    // 主函数
    public static void main(String[] args) throws Exception {
    
        // 获取配置参数
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
                
        // 检查命令语法
        if (otherArgs.length != 2) {
            System.err.println("Usage: Dedup <in> <out>");
            System.exit(2);
        }

        // 定义作业对象
        Job job = new Job(conf, "Sort");
        // 注册分布式类
        job.setJarByClass(Sort.class);
        // 注册Mapper类
        job.setMapperClass(Map.class);
        // 注册Reducer类
        job.setReducerClass(Reduce.class);
        // 注册Partition类
        job.setPartitionerClass(Partition.class);
        // 注册输出格式类
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);
        // 设置输入输出路径
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
                
        // 运行程序
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

运行结果

       输入文件1,2分别为:

       

       

小结

1. 掌握Partitioner方法的重写技巧，这是本程序最核心的部分。

2. 熟悉hadoop的key默认有序的性质。

3. 本文采取的是伪分布式，故只有1个reduce节点，体现不出hadoop的优越性。当对海量数据进行排序的时候，它的速度价值才能真正体现出来。