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

MapReduce二次排序

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

标签：

本文主要介绍下二次排序的实现方式

 

我们知道MapReduce是按照key来进行排序的，那么如果有个需求就是先按照第一个字段排序，在第一个字段相等的情况下，按照第二个字段排序，这就是传说中的二次排序。

 

下面就具体说一下二次排序的实现方式

 

主要就是4点

 

1.自定义一个Key

为什么要自定义一个Key，我们知道MapReduce中排序就是按照Key来排序的，我们既然想要实现按照两个字段进行排序，默认的方式肯定是不行的，所以自定义一个新的Key，Key里面有两个属性，也就是我们要排序的两个字段。

首先，实现WritableComparable接口，因为Key是可序列化并且可以比较的。其次，重载相关的方法，例如序列化、反序列化相关的方法write()、readFields()。重载在分区的时候要用到的hashCode()方法，注意后面会说到一个Partitioner类，也是用来分区的，用hashCode()方法和Partitioner类进行分区都是可以的，使用其中的一个即可。重载排序用的compareTo()方法。

 

2.分区函数类

上面定义了一个新的Key，那么我现在做分发，到底按照什么样的规则进行分发是在分区函数中定义的，这个类要继承Partitioner类，重载其中的分区方法getPartition()，在main()函数里面给job添加上即可，例如：job.setPartitionerClass(XXX.class);

注：这个类的作用和新Key中的hashCode()方法作用一样，所以如果在新key的hashCode()方法中写了分区的实现，这个分区类是可以省略的。

 

3.比较函数类

这个类决定着Key的排序规则，是一个比较器，需要继承WritableComparator类，并且重载其中的compare()方法。在main()函数里给job添加上即可，例如：job.setSortComparatorClass(XXX.class);

注：这个类的作用跟自定义Key的compareTo()方法一样，如果在自定义的Key中重载了compareTo()方法，这个类是可以省略的。

 

4.分组函数类

通过分区类，我们重新定义了key的分区规则，但是多个key不同的也可以进入一个reduce中，这不是我们想要的，我们需要分区函数来定义什么样的key可以进入相应的reduce来执行，因为也涉及到比较，所以这个类也需要继承WritableComparator，也可以实现RawComparator，并且重载其中的compare()方法，在main()函数中给job加上即可，如：job.setGroupingComparatorClass(XXX.class)。

 

下面我们来重新简化一下上一篇文章中提到了例子：

#需求：第一列升序，第一列相同时，第二列升序，其第一列相同的放在一个分区中输出

 

[java] view plain copy
 
  1. sort1   1  
  2. sort2   3  
  3. sort2   88  
  4. sort2   54  
  5. sort1   2  
  6. sort6   22  
  7. sort6   888  
  8. sort6   58  

#预期输出结果

 

#part-r-00000文件

 

[java] view plain copy
 
  1. sort1   1  
  2. sort1   2  

#part-r-00001文件

 

 

[java] view plain copy
 
  1. sort2   3  
  2. sort2   54  
  3. sort2   88  


#part-r-00002

 

 

[java] view plain copy
 
  1. sort6   22  
  2. sort6   58  
  3. sort6   888  





 

1.自定义组合键

 

[java] view plain copy
 
  1. public class CombinationKey implements WritableComparable<CombinationKey>{  
  2.   
  3.     private Text firstKey;  
  4.     private IntWritable secondKey;  
  5.       
  6.     //无参构造函数  
  7.     public CombinationKey() {  
  8.         this.firstKey = new Text();  
  9.         this.secondKey = new IntWritable();  
  10.     }  
  11.       
  12.     //有参构造函数  
  13.     public CombinationKey(Text firstKey, IntWritable secondKey) {  
  14.         this.firstKey = firstKey;  
  15.         this.secondKey = secondKey;  
  16.     }  
  17.   
  18.     public Text getFirstKey() {  
  19.         return firstKey;  
  20.     }  
  21.   
  22.     public void setFirstKey(Text firstKey) {  
  23.         this.firstKey = firstKey;  
  24.     }  
  25.   
  26.     public IntWritable getSecondKey() {  
  27.         return secondKey;  
  28.     }  
  29.   
  30.     public void setSecondKey(IntWritable secondKey) {  
  31.         this.secondKey = secondKey;  
  32.     }  
  33.   
  34.     public void write(DataOutput out) throws IOException {  
  35.         this.firstKey.write(out);  
  36.         this.secondKey.write(out);  
  37.     }  
  38.   
  39.     public void readFields(DataInput in) throws IOException {  
  40.         this.firstKey.readFields(in);  
  41.         this.secondKey.readFields(in);  
  42.     }  
  43.   
  44.     /** 
  45.      * 自定义比较策略 
  46.      * 注意：该比较策略用于MapReduce的第一次默认排序 
  47.      * 也就是发生在Map端的sort阶段 
  48.      * 发生地点为环形缓冲区(可以通过io.sort.mb进行大小调整) 
  49.      */  
  50.     public int compareTo(CombinationKey combinationKey) {  
  51.         int minus = this.getFirstKey().compareTo(combinationKey.getFirstKey());  
  52.         if (minus != 0){  
  53.             return minus;  
  54.         }  
  55.         return this.getSecondKey().get() - combinationKey.getSecondKey().get();  
  56.     }  
  57.       
  58.     /*  public int compareTo(CombinationKey combinationKey) { 
  59.         System.out.println("------------------------CombineKey flag-------------------"); 
  60.         return this.firstKey.compareTo(combinationKey.getFirstKey()); 
  61.     }*/  
  62.   
  63.     @Override  
  64.     public int hashCode() {  
  65.         final int prime = 31;  
  66.         int result = 1;  
  67.         result = prime * result + ((firstKey == null) ? 0 : firstKey.hashCode());  
  68.         return result;  
  69.     }  
  70.   
  71.     @Override  
  72.     public boolean equals(Object obj) {  
  73.         if (this == obj)  
  74.             return true;  
  75.         if (obj == null)  
  76.             return false;  
  77.         if (getClass() != obj.getClass())  
  78.             return false;  
  79.         CombinationKey other = (CombinationKey) obj;  
  80.         if (firstKey == null) {  
  81.             if (other.firstKey != null)  
  82.                 return false;  
  83.         } else if (!firstKey.equals(other.firstKey))  
  84.             return false;  
  85.         return true;  
  86.     }  
  87.   
  88.       
  89. }  


2.自定义分组

 

 

[java] view plain copy
 
  1. /** 
  2.  * 自定义分组有中方式，一种是继承WritableComparator 
  3.  * 另外一种是实现RawComparator接口 
  4.  * @author 廖钟民 
  5.  * time : 2015年1月19日下午3:30:11 
  6.  * @version 
  7.  */  
  8. public class DefinedGroupSort extends WritableComparator{  
  9.   
  10.   
  11.     protected DefinedGroupSort() {  
  12.         super(CombinationKey.class,true);  
  13.     }  
  14.   
  15.     @Override  
  16.     public int compare(WritableComparable a, WritableComparable b) {  
  17.         System.out.println("---------------------进入自定义分组---------------------");  
  18.         CombinationKey combinationKey1 = (CombinationKey) a;  
  19.         CombinationKey combinationKey2 = (CombinationKey) b;  
  20.         System.out.println("---------------------分组结果：" + combinationKey1.getFirstKey().compareTo(combinationKey2.getFirstKey()));  
  21.         System.out.println("---------------------结束自定义分组---------------------");  
  22.         //自定义按原始数据中第一个key分组  
  23.         return combinationKey1.getFirstKey().compareTo(combinationKey2.getFirstKey());  
  24.     }  
  25.   
  26.   
  27. }  


3.自定义分区

 

 

[java] view plain copy
 
  1. /** 
  2.  * 自定义分区 
  3.  *  
  4.  * @author 廖钟*民 time : 2015年1月19日下午12:13:54 
  5.  * @version 
  6.  */  
  7. public class DefinedPartition extends Partitioner<CombinationKey, IntWritable> {  
  8.   
  9.     /** 
  10.      * 数据输入来源：map输出 我们这里根据组合键的第一个值作为分区 如果不自定义分区的话，MapReduce会根据默认的Hash分区方法 将整个组合键相等的分到一个分区中，这样的话显然不是我们要的效果 
  11.      *  
  12.      * @param key 
  13.      *            map输出键值 
  14.      * @param value 
  15.      *            map输出value值 
  16.      * @param numPartitions 
  17.      *            分区总数，即reduce task个数 
  18.      */  
  19.     public int getPartition(CombinationKey key, IntWritable value, int numPartitions) {  
  20.           
  21.         if (key.getFirstKey().toString().endsWith("1")){  
  22.             return 0;  
  23.         } else if (key.getFirstKey().toString().endsWith("2")){  
  24.             return 1;  
  25.         } else {  
  26.             return 2;  
  27.         }  
  28.     }  
  29.   
  30. }  


4.主类

 

 

[java] view plain copy
 
  1. public class SecondSortMapReduce extends Configured implements Tool{  
  2.   
  3.     // 定义输入路径  
  4.         private String INPUT_PATH = "";  
  5.         // 定义输出路径  
  6.         private String OUT_PATH = "";  
  7.   
  8.         public static void main(String[] args) {  
  9.   
  10.             try {  
  11.                 ToolRunner.run(new SecondSortMapReduce(), args);  
  12.             } catch (Exception e) {  
  13.                 e.printStackTrace();  
  14.             }  
  15.         }  
  16.   
  17.           
  18.     public static class SecondSortMapper extends Mapper<Text, Text, CombinationKey, IntWritable>{  
  19.         /** 
  20.          * 这里要特殊说明一下，为什么要将这些变量写在map函数外边 
  21.          * 对于分布式的程序，我们一定要注意到内存的使用情况，对于MapReduce框架 
  22.          * 每一行的原始记录的处理都要调用一次map()函数，假设，这个map()函数要处理1一亿 
  23.          * 条输入记录，如果将这些变量都定义在map函数里面则会导致这4个变量的对象句柄 
  24.          * 非常的多(极端情况下将产生4*1亿个句柄，当然java也是有自动的GC机制的，一定不会达到这么多) 
  25.          * 导致栈内存被浪费掉，我们将其写在map函数外面，顶多就只有4个对象句柄 
  26.          */  
  27.         private CombinationKey combinationKey = new CombinationKey();  
  28.         Text sortName = new Text();  
  29.         IntWritable score = new IntWritable();  
  30.         String[] splits = null;  
  31.         protected void map(Text key, Text value, Mapper<Text, Text, CombinationKey, IntWritable>.Context context) throws IOException, InterruptedException {  
  32.             System.out.println("---------------------进入map()函数---------------------");  
  33.             //过滤非法记录(这里用计数器比较好)  
  34.             if (key == null || value == null || key.toString().equals("")){  
  35.                 return;  
  36.             }  
  37.             //构造相关属性  
  38.             sortName.set(key.toString());  
  39.             score.set(Integer.parseInt(value.toString()));  
  40.             //设置联合key  
  41.             combinationKey.setFirstKey(sortName);  
  42.             combinationKey.setSecondKey(score);  
  43.               
  44.             //通过context把map处理后的结果输出  
  45.             context.write(combinationKey, score);  
  46.             System.out.println("---------------------结束map()函数---------------------");  
  47.         }  
  48.           
  49.     }  
  50.       
  51.       
  52.     public static class SecondSortReducer extends Reducer<CombinationKey, IntWritable, Text, Text>{  
  53.           
  54.         StringBuffer sb = new StringBuffer();  
  55.         Text score = new Text();  
  56.         /** 
  57.          * 这里要注意一下reduce的调用时机和次数： 
  58.          * reduce每次处理一个分组的时候会调用一次reduce函数。 
  59.          * 所谓的分组就是将相同的key对应的value放在一个集合中 
  60.          * 例如：<sort1,1> <sort1,2> 
  61.          * 分组后的结果就是 
  62.          * <sort1,{1,2}>这个分组会调用一次reduce函数 
  63.          */  
  64.         protected void reduce(CombinationKey key, Iterable<IntWritable> values, Reducer<CombinationKey, IntWritable, Text, Text>.Context context)  
  65.                 throws IOException, InterruptedException {  
  66.               
  67.             //将联合Key的第一个元素作为新的key，遍历values将结果写出去  
  68.             for (IntWritable val : values){  
  69.                 context.write(key.getFirstKey(), new Text(String.valueOf(val.get())));  
  70.             }  
  71.               
  72.               
  73.         }  
  74.     }  
  75.   
  76.   
  77.     public int run(String[] args) throws Exception {  
  78.         // 给路径赋值  
  79.         INPUT_PATH = args[0];  
  80.         OUT_PATH = args[1];  
  81.         try {  
  82.             // 创建配置信息  
  83.             Configuration conf = new Configuration();  
  84.             conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, "\t");  
  85.   
  86.             // 创建文件系统  
  87.             FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);  
  88.             // 如果输出目录存在，我们就删除  
  89.             if (fileSystem.exists(new Path(OUT_PATH))) {  
  90.                 fileSystem.delete(new Path(OUT_PATH), true);  
  91.             }  
  92.   
  93.             // 创建任务  
  94.             Job job = new Job(conf, SecondSortMapReduce.class.getName());  
  95.             //设置成jar运行型  
  96.             job.setJarByClass(SecondSortMapReduce.class);  
  97.             // 1.1  设置输入目录和设置输入数据格式化的类  
  98.             FileInputFormat.setInputPaths(job, INPUT_PATH);  
  99.             job.setInputFormatClass(KeyValueTextInputFormat.class);  
  100.   
  101.             // 部1.2 设置自定义Mapper类和设置map函数输出数据的key和value的类型  
  102.             job.setMapperClass(SecondSortMapper.class);  
  103.             job.setMapOutputKeyClass(CombinationKey.class);  
  104.             job.setMapOutputValueClass(IntWritable.class);  
  105.   
  106.             // 1.3  设置分区和reduce数量(reduce的数量，和分区的数量对应)  
  107.             job.setPartitionerClass(DefinedPartition.class);  
  108.             job.setNumReduceTasks(3);  
  109.               
  110.             //设置自定义分组策略  
  111.             job.setGroupingComparatorClass(DefinedGroupSort.class);  
  112.             //设置自定义比较策略(因为我的CombineKey重写了compareTo方法，所以这个可以省略)  
  113.             //job.setSortComparatorClass(DefinedComparator.class);  
  114.               
  115.             // 1.4  排序  
  116.             //1.5   归约  
  117.             // 2.1  Shuffle把数据从Map端拷贝到Reduce端。  
  118.             // 2.2  指定Reducer类和输出key和value的类型  
  119.             job.setReducerClass(SecondSortReducer.class);  
  120.             job.setOutputKeyClass(Text.class);  
  121.             job.setOutputValueClass(Text.class);  
  122.   
  123.             //2.3   指定输出的路径和设置输出的格式化类  
  124.             FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));  
  125.             job.setOutputFormatClass(TextOutputFormat.class);  
  126.   
  127.   
  128.             // 提交作业 退出  
  129.             System.exit(job.waitForCompletion(true) ? 0 : 1);  
  130.           
  131.         } catch (Exception e) {  
  132.             e.printStackTrace();  
  133.         }  
  134.         return 0;  
  135.     }  
  136. }  

注：使用jar包的方式运行程序一定不要忘记很重要的一句：job.setJarByClass(XXX);

 

程序运行的结果：

技术分享

MapReduce二次排序

标签：

原文地址：http://www.cnblogs.com/thinkpad/p/5173743.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
迷上了代码！