码迷，mamicode.com
首页 > 编程语言 > 详细

MapReduce排序之 二次排序

时间：2016-01-31 21:31:22      阅读：559      评论：0      收藏：0      [点我收藏+]

标签：

一：背景

Hadoop中虽然有自动排序和分组，由于自带的排序是按照Key进行排序的，有些时候，我们希望同时对Key和Value进行排序。自带的排序功能就无法满足我们了，还好Hadoop提供了一些组件可以让开发人员进行二次排序。

 

二：技术实现

我们先来看案例需求

#需求1： 首先按照第一列数字升序排列，当第一列数字相同时，第二列数字也升序排列(列之间用制表符\t隔开)

 

[java] view plain copy
 
  1. 3   3  
  2. 3   2  
  3. 3   1  
  4. 2   2  
  5. 2   1  
  6. 1   1  

MapReduce计算之后的结果应该是：

 

 

[java] view plain copy
 
  1. 1   1  
  2. 2   1  
  3. 2   2  
  4. 3   1  
  5. 3   2  
  6. 3   3  


#需求2：第一列不相等时，第一列按降序排列，当第一列相等时，第二列按升序排列

 

 

[java] view plain copy
 
  1. 3   3  
  2. 3   2  
  3. 3   1  
  4. 2   2  
  5. 2   1  
  6. 1   1  

MapReduce计算之后的结果应该是：

[java] view plain copy
 
  1. 3   1  
  2. 3   2  
  3. 3   3  
  4. 2   1  
  5. 2   2  
  6. 1   1  



下面是实现代码，实现两种需求的关键是compareTo()方法的实现不同：

 

[java] view plain copy
 
    1. public class SecondSortTest {  
    2.   
    3.     // 定义输入路径  
    4.         private static final String INPUT_PATH = "hdfs://liaozhongmin:9000/data";  
    5.         // 定义输出路径  
    6.         private static final String OUT_PATH = "hdfs://liaozhongmin:9000/out";  
    7.   
    8.         public static void main(String[] args) {  
    9.   
    10.             try {  
    11.                 // 创建配置信息  
    12.                 Configuration conf = new Configuration();  
    13.                   
    14.                 /**********************************************/  
    15.                 //对Map端输出进行压缩  
    16.                 //conf.setBoolean("mapred.compress.map.output", true);  
    17.                 //设置map端输出使用的压缩类  
    18.                 //conf.setClass("mapred.map.output.compression.codec", GzipCodec.class, CompressionCodec.class);  
    19.                 //对reduce端输出进行压缩  
    20.                 //conf.setBoolean("mapred.output.compress", true);  
    21.                 //设置reduce端输出使用的压缩类  
    22.                 //conf.setClass("mapred.output.compression.codec", GzipCodec.class, CompressionCodec.class);  
    23.                 // 添加配置文件(我们可以在编程的时候动态配置信息，而不需要手动去改变集群)  
    24.                 /* 
    25.                  * conf.addResource("classpath://hadoop/core-site.xml");  
    26.                  * conf.addResource("classpath://hadoop/hdfs-site.xml"); 
    27.                  * conf.addResource("classpath://hadoop/hdfs-site.xml"); 
    28.                  */  
    29.   
    30.                 // 创建文件系统  
    31.                 FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);  
    32.                 // 如果输出目录存在，我们就删除  
    33.                 if (fileSystem.exists(new Path(OUT_PATH))) {  
    34.                     fileSystem.delete(new Path(OUT_PATH), true);  
    35.                 }  
    36.   
    37.                 // 创建任务  
    38.                 Job job = new Job(conf, SecondSortTest.class.getName());  
    39.   
    40.                 //1.1   设置输入目录和设置输入数据格式化的类  
    41.                 FileInputFormat.setInputPaths(job, INPUT_PATH);  
    42.                 job.setInputFormatClass(TextInputFormat.class);  
    43.   
    44.                 //1.2   设置自定义Mapper类和设置map函数输出数据的key和value的类型  
    45.                 job.setMapperClass(MySecondSortMapper.class);  
    46.                 job.setMapOutputKeyClass(CombineKey.class);  
    47.                 job.setMapOutputValueClass(LongWritable.class);  
    48.   
    49.                 //1.3   设置分区和reduce数量(reduce的数量，和分区的数量对应，因为分区为一个，所以reduce的数量也是一个)  
    50.                 job.setPartitionerClass(HashPartitioner.class);  
    51.                 job.setNumReduceTasks(1);  
    52.   
    53.                 //1.4   排序、分组  
    54.                 //1.5   归约  
    55.                 //2.1   Shuffle把数据从Map端拷贝到Reduce端。  
    56.                 //2.2   指定Reducer类和输出key和value的类型  
    57.                 job.setReducerClass(MySecondSortReducer.class);  
    58.                 job.setOutputKeyClass(LongWritable.class);  
    59.                 job.setOutputValueClass(LongWritable.class);  
    60.   
    61.                 //2.3   指定输出的路径和设置输出的格式化类  
    62.                 FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));  
    63.                 job.setOutputFormatClass(TextOutputFormat.class);  
    64.   
    65.   
    66.                 // 提交作业 退出  
    67.                 System.exit(job.waitForCompletion(true) ? 0 : 1);  
    68.               
    69.             } catch (Exception e) {  
    70.                 e.printStackTrace();  
    71.             }  
    72.         }  
    73.       
    74.     public static class MySecondSortMapper extends Mapper<LongWritable, Text, CombineKey, LongWritable>{  
    75.           
    76.         //定义联合的key  
    77.         private CombineKey combineKey = new CombineKey();  
    78.           
    79.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, CombineKey, LongWritable>.Context context) throws IOException,  
    80.                 InterruptedException {  
    81.             //对输入的value进行切分  
    82.             String[] splits = value.toString().split("\t");  
    83.             //设置联合的key  
    84.             combineKey.setComKey(Long.parseLong(splits[0]));  
    85.             combineKey.setComVal(Long.parseLong(splits[1]));  
    86.               
    87.             //通过context写出去  
    88.             context.write(combineKey, new LongWritable(Long.parseLong(splits[1])));  
    89.         }  
    90.     }  
    91.       
    92.       
    93.     public static class MySecondSortReducer extends Reducer<CombineKey, LongWritable, LongWritable, LongWritable>{  
    94.         @Override  
    95.         protected void reduce(CombineKey combineKey, Iterable<LongWritable> values, Reducer<CombineKey, LongWritable, LongWritable, LongWritable>.Context context)  
    96.                 throws IOException, InterruptedException {  
    97.             //因为输入的CombineKey已经排好序了，所有我们只要获取其中的两个成员变量写出去就可以了。values在这个例子中没有什么作用  
    98.             context.write(new LongWritable(combineKey.getComKey()), new LongWritable(combineKey.getComVal()));  
    99.         }  
    100.     }  
    101.   
    102. }  
    103.   
    104. /** 
    105.  * 重新组合成一个key，实现二次排序 
    106.  * @author 廖*民 
    107.  * time : 2015年1月18日下午7:27:52 
    108.  * @version 
    109.  */  
    110. class CombineKey implements WritableComparable<CombineKey>{  
    111.   
    112.     public long comKey;  
    113.     public long comVal;  
    114.       
    115.     //必须提供无参构造函数，否则hadoop反射机制会出错  
    116.     public CombineKey() {  
    117.           
    118.     }  
    119.     //有参构造函数  
    120.     public CombineKey(long comKey, long comVal) {  
    121.         this.comKey = comKey;  
    122.         this.comVal = comVal;  
    123.     }  
    124.   
    125.       
    126.       
    127.     public long getComKey() {  
    128.         return comKey;  
    129.     }  
    130.     public void setComKey(long comKey) {  
    131.         this.comKey = comKey;  
    132.     }  
    133.     public long getComVal() {  
    134.         return comVal;  
    135.     }  
    136.     public void setComVal(long comVal) {  
    137.         this.comVal = comVal;  
    138.     }  
    139.       
    140.     public void write(DataOutput out) throws IOException {  
    141.         out.writeLong(comKey);  
    142.         out.writeLong(comVal);  
    143.     }  
    144.   
    145.     public void readFields(DataInput in) throws IOException {  
    146.         this.comKey = in.readLong();  
    147.         this.comVal = in.readLong();  
    148.     }  
    149.   
    150.     /** 
    151.      * 这个方法一定要实现 
    152.      * java里面排序默认是小的放在前面，即返回负数的放在前面，这样就是所谓的升序排列 
    153.      * 我们在下面的方法中直接返回一个差值，也就相当于会升序排列。 
    154.      * 如果我们要实现降序排列，那么我们就可以返回一个正数 
    155.      */  
    156.     /*public int compareTo(CombineKey o) { 
    157.         //第一列不相同时按升序排列，当第一列相同时第二列按升序排列 
    158.         long minus = this.comKey - o.comKey; 
    159.         //如果第一个值不相等时，我们就先对第一列进行排序 
    160.         if (minus != 0){ 
    161.             return (int) minus; 
    162.         } 
    163.         //如果第一列相等时，我们就对第二列进行排序 
    164.         return (int) (this.comVal - o.comVal); 
    165.     }*/  
    166.       
    167.     /** 
    168.      * 为了实现第一列不同时按降序排序，第一列相同时第二列按升序排列 
    169.      * 第一列：降序，当第一列相同时，第二列：升序 
    170.      * 为了实现降序， 
    171.      */  
    172.     public int compareTo(CombineKey o) {  
    173.         //如果a-b<0即,a小于b，按这样 的思路应该是升序排列，我们可以返回一个相反数使其降序  
    174.         long tmp = this.comKey - o.comKey;  
    175.         //如果第一个值不相等时，我们就先对第一列进行排序  
    176.         if (tmp != 0){  
    177.             return (int) (-tmp);  
    178.         }  
    179.         //如果第一列相等时，我们就对第二列进行升序排列  
    180.         return (int) (this.comVal - o.comVal);  
    181.     }  
    182.       
    183.       
    184.     @Override  
    185.     public int hashCode() {  
    186.         final int prime = 31;  
    187.         int result = 1;  
    188.         result = prime * result + (int) (comKey ^ (comKey >>> 32));  
    189.         return result;  
    190.     }  
    191.     @Override  
    192.     public boolean equals(Object obj) {  
    193.         if (this == obj)  
    194.             return true;  
    195.         if (obj == null)  
    196.             return false;  
    197.         if (getClass() != obj.getClass())  
    198.             return false;  
    199.         CombineKey other = (CombineKey) obj;  
    200.         if (comKey != other.comKey)  
    201.             return false;  
    202.         return true;  
    203.     }  
    204.       
    205. }  

MapReduce排序之 二次排序

标签：

原文地址：http://www.cnblogs.com/thinkpad/p/5173736.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！