首页 > 代码库 > 第八篇：经典案例 - 排序
第八篇：经典案例 - 排序
前言
在计算机领域，排序的重要性不用多说。而排序的算法，效率分析等也一直是研究的热点。
本文将给出使用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)。
代码实现
1 package org.apache.hadoop.examples; 2 3 import java.io.IOException; 4 5 //导入各种Hadoop包 6 import org.apache.hadoop.conf.Configuration; 7 import org.apache.hadoop.fs.Path; 8 import org.apache.hadoop.io.IntWritable; 9 import org.apache.hadoop.io.Text; 10 import org.apache.hadoop.mapreduce.Job; 11 import org.apache.hadoop.mapreduce.Mapper; 12 import org.apache.hadoop.mapreduce.Partitioner; 13 import org.apache.hadoop.mapreduce.Reducer; 14 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 15 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 16 import org.apache.hadoop.util.GenericOptionsParser; 17 18 // 主类 19 public class Sort { 20 21 // Mapper类 22 public static class Map extends Mapper<Object, Text, IntWritable, IntWritable>{ 23 24 // new一个值为1的IntWritable对象 25 private static IntWritable data = http://www.mamicode.com/new IntWritable(1); 26 27 // 实现map函数 28 public void map(Object key, Text value, Context context) throws IOException, InterruptedException { 29 30 // 将切分后的value作为中间输出的key，然后value值为1。 31 String line = value.toString(); 32 data.set(Integer.parseInt(line)); 33 context.write(data, new IntWritable(1)); 34 } 35 } 36 37 // Reducer类 38 public static class Reduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> { 39 40 // new一个值为空的IntWritable对象 41 private static IntWritable linenum = new IntWritable(); 42 43 // 实现reduce函数 44 public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { 45 46 // 写入结果键值对 47 for (IntWritable val : values) { 48 context.write(linenum, key); 49 linenum = new IntWritable(linenum.get()+1); 50 } 51 } 52 } 53 54 // 重写Partitioner类 55 public static class Partition extends Partitioner <IntWritable, IntWritable> { 56 57 // 重载getPartition方法。下面的三个参数分别为map中间输出的键，值，以及分割区间的个数。 58 public int getPartition(IntWritable key, IntWritable value, int numPartitions) { 59 60 // 依次将键值对分配到各个分割区间 61 int MaxNumber = 65223; 62 int bound = MaxNumber/numPartitions + 1; 63 int keynumber = key.get(); 64 65 for (int i=0; i<numPartitions; i++) { 66 if (keynumber < bound * (i+1) && keynumber >= bound*i) { 67 68 // 返回的 i 就是分配到的区间号 69 return i; 70 } 71 } 72 73 return -1; 74 } 75 } 76 77 // 主函数 78 public static void main(String[] args) throws Exception { 79 80 // 获取配置参数 81 Configuration conf = new Configuration(); 82 String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); 83 84 // 检查命令语法 85 if (otherArgs.length != 2) { 86 System.err.println("Usage: Dedup <in> <out>"); 87 System.exit(2); 88 } 89 90 // 定义作业对象 91 Job job = new Job(conf, "Sort"); 92 // 注册分布式类 93 job.setJarByClass(Sort.class); 94 // 注册Mapper类 95 job.setMapperClass(Map.class); 96 // 注册Reducer类 97 job.setReducerClass(Reduce.class); 98 // 注册Partition类 99 job.setPartitionerClass(Partition.class);100 // 注册输出格式类101 job.setOutputKeyClass(IntWritable.class);102 job.setOutputValueClass(IntWritable.class);103 // 设置输入输出路径104 FileInputFormat.addInputPath(job, new Path(otherArgs[0]));105 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));106 107 // 运行程序108 System.exit(job.waitForCompletion(true) ? 0 : 1);109 }110 }
运行结果
输入文件1,2分别为:
小结
1. 掌握Partitioner方法的重写技巧，这是本程序最核心的部分。
2. 熟悉hadoop的key默认有序的性质。
3. 本文采取的是伪分布式，故只有1个reduce节点，体现不出hadoop的优越性。当对海量数据进行排序的时候，它的速度价值才能真正体现出来。
第八篇：经典案例 - 排序