概念
MapTask和ReduceTask均会对数据按照key进行排序。该操作属于Hadoop的默认行为。任何应用程序中的数
据均会被排序，而不管逻辑上是否需要。
默认排序是按照字典顺序排序，且实现该排序的方法是快速排序。
对于MapTask，它会将处理的结果暂时放到环形缓冲区中，当环形缓冲区使用率达到一定阈值后，再对缓冲区中的数据进行一次快速排序，并将这些有序数据溢写到磁盘上，而当数据处理完毕后，它会对磁盘上所有文件进行归并排序。
对于ReduceTask，它从每个MapTask上远程拷贝相应的数据文件，如果文件大小超过一定阈值，则溢写磁盘上，否则存储在内存中。如果磁盘上文件数目达到一定阈值，则进行一次归并排序以生成一个更大文件；如果内存中文件大小或者数目超过一定阈值，则对于ReduceTask，它从每个MapTask上远程拷贝相应的数据文件，如果文件大小超过一定阈值，则溢写磁盘上，否则存储在内存中。如果磁盘上文件数目达到一定阈值，则进行一次归并排序以生成一个更大文件；如果内存中文件大小或者数目超过一定阈值，则
MapReduce根据输入记录的键对数据集排序。保证输出的每个文件内部有序。
2.全排序
最终输出结果只有一个文件，且文件内部有序。实现方式是只设置一个ReduceTask。但该方法在处理大型文件时效率极低，因为一台机器处理所有文件，完全丧失了MapReduce所提供的并行架构。（比如wordcount最终将所有数据放到一个文件内再总的进行一次全排序）
3.二次排序
在自定义排序过程中，如果compareTo中的判断条件为两个即为二次排序。（比如先按照总流量使用降序，如果两个用户总流量相同的情况下再根据上行流量进行降序排列...）。
自定义排序WritableComparable
全排序
需求
根据上次的【流量统计案例】的结果，要求结果按照总流量进行降序排列。（默认是按照key（手机号）进行排序）。
输入数据
13480253104 180 180 360 13502468823 7335 110349 117684 13560436666 1116 954 2070 13560439658 2034 5892 7926 13602846565 1938 2910 4848 13660577991 6960 690 7650 13719199419 240 0 240 13726230503 2481 24681 27162 13726238888 2481 24681 27162 13760778710 120 120 240 13826544101 264 0 264 13922314466 3008 3720 6728 13925057413 11058 48243 59301 13926251106 240 0 240 13926435656 132 1512 1644 15013685858 3659 3538 7197 15920133257 3156 2936 6092 15989002119 1938 180 2118 18211575961 1527 2106 3633 18320173382 9531 2412 11943 84138413 4116 1432 5548
思路
让FlowBean来作为key（因为hadoop框架中的key必须支持可排序，如果让手机号作为key，结果仍然是按照手机号的字典序进行排序）
实现
1.修改FlowBean继承Writablepmparable接口,泛型的类型为需要排序的类
public class FlowBean implements WritableComparable<FlowBean> { //... }
2.重写compareTo方法
@Override public int compareTo(FlowBean o) { //按照总流量进行降序排列 if (this.sumFlow > o.sumFlow){ return -1; //降序 }else if (this.sumFlow < o.sumFlow){ return 1; //升序 }else { return 0; } }
3.Mapper
package com.lyh.mapreduce.writableComparable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.util.StringUtils; import java.io.IOException; public class FlowMapper extends Mapper<LongWritable, Text,FlowBean,Text> { private FlowBean outKey = new FlowBean(); private Text outValue = new Text(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //1.获取一行数据 String line = value.toString(); //2.切割 String[] words = StringUtils.split(line, '\t'); //3.封装value outValue.set(words[0]); outKey.setUpFlow(Long.parseLong(words[1]));//对上一次的运行结果进行排序 outKey.setDownFlow(Long.parseLong(words[2])); outKey.setSumFlow(); //4.写出 context.write(outKey,outValue); } }
4.Reducer
public class FlowReducer extends Reducer<FlowBean, Text,Text,FlowBean> { @Override protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException { for (Text value : values) { context.write(value,key); } } }
5.Runner类
主要修改MapOutputKeyClass 和 MapOutputValueClass。
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; public class FlowRunner extends Configured implements Tool { public static void main(String[] args) throws Exception { ToolRunner.run(new Configuration(),new FlowRunner(),args); } @Override public int run(String[] args) throws Exception { //1.获取job Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "flow compu"); //2.配置jar包路径 job.setJarByClass(FlowRunner.class); //3.关联mapper和reducer job.setMapperClass(FlowMapper.class); job.setReducerClass(FlowReducer.class); //4.设置map、reduce输出的k、v类型 job.setMapOutputKeyClass(FlowBean.class); job.setMapOutputValueClass(Text.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(FlowBean.class); //5.设置数据输入的路径 FileInputFormat.setInputPaths(job,new Path("D:\\MapReduce_Data_Test\\writable\\output1")); //6.设置输出路径-输出目录不可存在 FileOutputFormat.setOutputPath(job,new Path("D:\\MapReduce_Data_Test\\writable\\output2")); //7.提交job return job.waitForCompletion(true) ? 0 : 1;//verbose：是否监控并打印job的信息 } }
运行结果
全部按照总流量（最后一列）排序
13502468823 7335 110349 117684 13925057413 11058 48243 59301 13726230503 2481 24681 27162 13726238888 2481 24681 27162 18320173382 9531 2412 11943 13560439658 2034 5892 7926 13660577991 6960 690 7650 15013685858 3659 3538 7197 13922314466 3008 3720 6728 15920133257 3156 2936 6092 84138413 4116 1432 5548 13602846565 1938 2910 4848 18211575961 1527 2106 3633 15989002119 1938 180 2118 13560436666 1116 954 2070 13926435656 132 1512 1644 13480253104 180 180 360 13826544101 264 0 264 13719199419 240 0 240 13760778710 120 120 240 13926251106 240 0 240
二次排序
我们发现，刚才的输出文件中有三位用户的总流量相同，我们希望如果用户总流量相同的情况下，我们再根据上行流量去降序排列。
13719199419 240 0 240 13760778710 120 120 240 13926251106 240 0 240
只需要再FlowBean中修改compareTo方法
@Override public int compareTo(FlowBean o) { //按照总流量进行降序排列 if (this.sumFlow > o.sumFlow){ return -1; //降序 }else if (this.sumFlow < o.sumFlow){ return 1; //升序 }else {//如果总流量相同 //按照上行流量降序排列 if (this.upFlow > o.upFlow){ return -1; }else if (this.upFlow < o.upFlow){ return 1; }else { return 0; } } }
运行结果
... 13719199419 240 0 240 13926251106 240 0 240 13760778710 120 120 240
区内排序
需求：根据手机号（前三位数）进行分区，并要求每个分区内的
输入：
13502468823 7335 110349 117684 13925057413 11058 48243 59301 13726230503 2481 24681 27162 13726238888 2481 24681 27162 13820173382 9531 2412 11943 13560439658 2034 5892 7926 13760577991 6960 690 7650 13813685858 3659 3538 7197 13922314466 3008 3720 6728 13920133257 3156 2936 6092 84138413 4116 1432 5548 13502846565 1938 2910 4848 13911575961 1527 2106 3633 13789002119 1938 180 2118 13560436666 1116 954 2070 13926435656 132 1512 1644 13880253104 180 180 360 13826544101 264 0 264 13719199419 240 0 240 13926251106 240 0 240 13760778710 120 120 240
只需要重写一个分区类
import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Partitioner; public class MyPartitioner extends Partitioner<FlowBean,Text> { @Override public int getPartition(FlowBean flowBean, Text text, int partitioner) { //获取手机号 String line = text.toString(); String phone = line.substring(0,3); if (phone.equals("135")){ partitioner = 0; }else if (phone.equals("137")){ partitioner = 1; }else if (phone.equals("138")){ partitioner = 2; }else if (phone.equals("139")){ partitioner = 3; }else { partitioner = 0; } return partitioner; } }
运行结果
生成4个分区，分别存储135、137、138、139开头的手机号。
