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

Hadoop二次排序及MapReduce处理流程实例详解

时间:2016-01-31 21:17:38      阅读:243      评论:0      收藏:0      [点我收藏+]

标签:

一、概述

MapReduce框架对处理结果的输出会根据key值进行默认的排序,这个默认排序可以满足一部分需求,但是也是十分有限的,在我们实际的需求当中,往往有要对reduce输出结果进行二次排序的需求。对于二次排序的实现,网络上已经有很多人分享过了,但是对二次排序的实现原理及整个MapReduce框架的处理流程的分析还是有非常大的出入,而且部分分析是没有经过验证的。本文将通过一个实际的MapReduce二次排序的例子,讲述二次排序的实现和其MapReduce的整个处理流程,并且通过结果和Map、Reduce端的日志来验证描述的处理流程的正确性。

 

二、需求描述

1.输入数据

 

[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  

 

2.目标输出

 

[java] view plain copy
 
  1. sort1   1,2  
  2. sort2   3,54,88  
  3. sort6   22,58,888  


三、解决思路

 

1.首先,在思考解决问题思路时,我们应该先深刻的理解MapReduce处理数据的整个流程,这是最基础的,不然的话是不可能找到解决问题的思路的。我描述一下MapReduce处理数据的大概流程:首先,MapReduce框架通过getSplits()方法实现对原始文件的切片之后,每一个切片对应着一个MapTask,InputSplit输入到map()函数进行处理,中间结果经过环形缓冲区的排序,然后分区、自定义二次排序(如果有的话)和合并,再通过Shuffle操作将数据传输到reduce Task端,reduce端也存在着缓冲区,数据也会在缓冲区和磁盘中进行合并排序等操作,然后对数据按照key值进行分组,然后每处理完一个分组之后就会去调用一次reduce()函数,最终输出结果。大概流程 我画了一下,如下图:

技术分享

2.具体解决思路

(1):Map端处理

根据上面的需求,我们有一个非常明确的目标就是要对第一列相同的记录,并且对合并后的数字进行排序。我们都知道MapReduce框架不管是默认排序或者是自定义排序都只是对key值进行排序,现在的情况是这些数据不是key值,怎么办?其实我们可以将原始数据的key值和其对应的数据组合成一个新的key值,然后新的key值对应的value还是原始数据中的valu。那么我们就可以将原始数据的map输出变成类似下面的数据结构:

 

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


那么我们只需要对[]里面的心key值进行排序就OK了,然后我们需要自定义一个分区处理器,因为我的目标不是想将新key相同的记录传到一个reduce中,而是想将新key中第一个字段相同的记录放到同一个reduce中进行分组合并,所以我们需要根据新key值的第一个字段来自定义一个分区处理器。通过分区操作后,得到的数据流如下:

 

 

[java] view plain copy
 
  1. Partition1:{[sort1,1],1}、{[sort1,2],2}  
  2.   
  3. Partition2:{[sort2,3],3}、{[sort2,88],88}、{[sort2,54],54}  
  4.   
  5. Partition3:{[sort6,22],22}、{[sort6,888],888}、{[sort6,58],58}  


分区操作完成之后,我调用自己的自定义排序器对新的key值进行排序。

 

 

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



 

(2).Reduce端处理

经过Shuffle处理之后,数据传输到Reducer端了。在Reducer端按照组合键的第一个字段进行分组,并且每处理完一次分组之后就会调用一次reduce函数来对这个分组进行处理和输出。最终各个分组的数据结果变成类似下面的数据结构:

 

[java] view plain copy
 
  1. sort1   1,2  
  2. sort2   3,54,88  
  3. sort6   22,58,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.     /*public int compareTo(CombinationKey combinationKey) { 
  46.         int minus = this.getFirstKey().compareTo(combinationKey.getFirstKey()); 
  47.         if (minus != 0){ 
  48.             return minus; 
  49.         } 
  50.         return this.getSecondKey().get() - combinationKey.getSecondKey().get(); 
  51.     }*/  
  52.     /** 
  53.      * 自定义比较策略 
  54.      * 注意:该比较策略用于MapReduce的第一次默认排序 
  55.      * 也就是发生在Map端的sort阶段 
  56.      * 发生地点为环形缓冲区(可以通过io.sort.mb进行大小调整) 
  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. }  

说明:在自定义组合键的时候,我们需要特别注意,一定要实现WritableComparable接口,并且实现compareTo()方法的比较策略。这个用于MapReduce的第一次默认排序,也就是发生在Map阶段的sort小阶段,发生地点为环形缓冲区(可以通过io.sort.mb进行大小调整),但是其对我们最终的二次排序结果是没有影响的,我们二次排序的最终结果是由我们的自定义比较器决定的。

 

2.自定义分区器

 

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


3.自定义比较器

 

 

[java] view plain copy
 
  1. public class DefinedComparator extends WritableComparator{  
  2.   
  3.     protected DefinedComparator() {  
  4.         super(CombinationKey.class,true);  
  5.     }  
  6.   
  7.     /** 
  8.      * 第一列按升序排列,第二列也按升序排列 
  9.      */  
  10.     public int compare(WritableComparable a, WritableComparable b) {  
  11.         System.out.println("------------------进入二次排序-------------------");  
  12.         CombinationKey c1 = (CombinationKey) a;  
  13.         CombinationKey c2 = (CombinationKey) b;  
  14.         int minus = c1.getFirstKey().compareTo(c2.getFirstKey());  
  15.           
  16.         if (minus != 0){  
  17.             System.out.println("------------------结束二次排序-------------------");  
  18.             return minus;  
  19.         } else {  
  20.             System.out.println("------------------结束二次排序-------------------");  
  21.             return c1.getSecondKey().get() -c2.getSecondKey().get();  
  22.         }  
  23.     }  
  24. }  


4.自定义分组

 

 

[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. }  


5.主体程序实现

 

 

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


程序运行的结果:

 

技术分享

 

五、处理流程

看到前面的代码,都知道我在各个组件上已经设置好了相应的标志,用于追踪整个MapReduce处理二次排序的处理流程。现在让我们分别看看Map端和Reduce端的日志情况。

(1)Map端日志分析

 

[java] view plain copy
 
  1. 15/01/19 15:32:29 INFO input.FileInputFormat: Total input paths to process : 1  
  2. 15/01/19 15:32:29 WARN snappy.LoadSnappy: Snappy native library not loaded  
  3. 15/01/19 15:32:30 INFO mapred.JobClient: Running job: job_local_0001  
  4. 15/01/19 15:32:30 INFO mapred.Task:  Using ResourceCalculatorPlugin : null  
  5. 15/01/19 15:32:30 INFO mapred.MapTask: io.sort.mb = 100  
  6. 15/01/19 15:32:30 INFO mapred.MapTask: data buffer = 79691776/99614720  
  7. 15/01/19 15:32:30 INFO mapred.MapTask: record buffer = 262144/327680  
  8. ---------------------进入map()函数---------------------  
  9. ---------------------进入自定义分区---------------------  
  10. ---------------------结束自定义分区---------------------  
  11. ---------------------结束map()函数---------------------  
  12. ---------------------进入map()函数---------------------  
  13. ---------------------进入自定义分区---------------------  
  14. ---------------------结束自定义分区---------------------  
  15. ---------------------结束map()函数---------------------  
  16. ---------------------进入map()函数---------------------  
  17. ---------------------进入自定义分区---------------------  
  18. ---------------------结束自定义分区---------------------  
  19. ---------------------结束map()函数---------------------  
  20. ---------------------进入map()函数---------------------  
  21. ---------------------进入自定义分区---------------------  
  22. ---------------------结束自定义分区---------------------  
  23. ---------------------结束map()函数---------------------  
  24. ---------------------进入map()函数---------------------  
  25. ---------------------进入自定义分区---------------------  
  26. ---------------------结束自定义分区---------------------  
  27. ---------------------结束map()函数---------------------  
  28. ---------------------进入map()函数---------------------  
  29. ---------------------进入自定义分区---------------------  
  30. ---------------------结束自定义分区---------------------  
  31. ---------------------结束map()函数---------------------  
  32. ---------------------进入map()函数---------------------  
  33. ---------------------进入自定义分区---------------------  
  34. ---------------------结束自定义分区---------------------  
  35. ---------------------结束map()函数---------------------  
  36. ---------------------进入map()函数---------------------  
  37. ---------------------进入自定义分区---------------------  
  38. ---------------------结束自定义分区---------------------  
  39. ---------------------结束map()函数---------------------  
  40. 15/01/19 15:32:30 INFO mapred.MapTask: Starting flush of map output  
  41. ------------------进入二次排序-------------------  
  42. ------------------结束二次排序-------------------  
  43. ------------------进入二次排序-------------------  
  44. ------------------结束二次排序-------------------  
  45. ------------------进入二次排序-------------------  
  46. ------------------结束二次排序-------------------  
  47. ------------------进入二次排序-------------------  
  48. ------------------结束二次排序-------------------  
  49. ------------------进入二次排序-------------------  
  50. ------------------结束二次排序-------------------  
  51. ------------------进入二次排序-------------------  
  52. ------------------结束二次排序-------------------  
  53. ------------------进入二次排序-------------------  
  54. ------------------结束二次排序-------------------  
  55. ------------------进入二次排序-------------------  
  56. ------------------结束二次排序-------------------  
  57. ------------------进入二次排序-------------------  
  58. ------------------结束二次排序-------------------  
  59. ------------------进入二次排序-------------------  
  60. ------------------结束二次排序-------------------  
  61. ------------------进入二次排序-------------------  
  62. ------------------结束二次排序-------------------  
  63. ------------------进入二次排序-------------------  
  64. ------------------结束二次排序-------------------  
  65. 15/01/19 15:32:30 INFO mapred.MapTask: Finished spill 0  
  66. 15/01/19 15:32:30 INFO mapred.Task: Task:attempt_local_0001_m_000000_0 is done. And is in the process of commiting  
  67. 15/01/19 15:32:30 INFO mapred.LocalJobRunner:   
  68. 15/01/19 15:32:30 INFO mapred.Task: Task ‘attempt_local_0001_m_000000_0‘ done.  
  69. 15/01/19 15:32:30 INFO mapred.Task:  Using ResourceCalculatorPlugin : null  
  70. 15/01/19 15:32:30 INFO mapred.LocalJobRunner:   

从Map端的日志,我们可以很容易的看出来每一条记录开始时进入到map()函数进行处理,处理完了之后立马就自定义分区函数中对其进行分区,当所有输入数据经过map()函数和分区函数处理之后,就调用自定义二次排序函数对其进行排序。

 

(2)Reduce端日志分析

 

[java] view plain copy
 
  1. 15/01/19 15:32:30 INFO mapred.Merger: Merging 1 sorted segments  
  2. 15/01/19 15:32:30 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 130 bytes  
  3. 15/01/19 15:32:30 INFO mapred.LocalJobRunner:   
  4. ---------------------进入自定义分组---------------------  
  5. ---------------------分组结果:0  
  6. ---------------------结束自定义分组---------------------  
  7. ---------------------进入自定义分组---------------------  
  8. ---------------------分组结果:-1  
  9. ---------------------结束自定义分组---------------------  
  10. ---------------------进入reduce()函数---------------------  
  11. ---------------------{[sort1,2],[1,2]}  
  12. ---------------------结束reduce()函数---------------------  
  13. ---------------------进入自定义分组---------------------  
  14. ---------------------分组结果:0  
  15. ---------------------结束自定义分组---------------------  
  16. ---------------------进入自定义分组---------------------  
  17. ---------------------分组结果:0  
  18. ---------------------结束自定义分组---------------------  
  19. ---------------------进入自定义分组---------------------  
  20. ---------------------分组结果:-4  
  21. ---------------------结束自定义分组---------------------  
  22. ---------------------进入reduce()函数---------------------  
  23. ---------------------{[sort2,88],[3,54,88]}  
  24. ---------------------结束reduce()函数---------------------  
  25. ---------------------进入自定义分组---------------------  
  26. ---------------------分组结果:0  
  27. ---------------------结束自定义分组---------------------  
  28. ---------------------进入自定义分组---------------------  
  29. ---------------------分组结果:0  
  30. ---------------------结束自定义分组---------------------  
  31. ---------------------进入reduce()函数---------------------  
  32. ---------------------{[sort6,888],[22,58,888]}  
  33. ---------------------结束reduce()函数---------------------  
  34. 15/01/19 15:32:30 INFO mapred.Task: Task:attempt_local_0001_r_000000_0 is done. And is in the process of commiting  
  35. 15/01/19 15:32:30 INFO mapred.LocalJobRunner:   
  36. 15/01/19 15:32:30 INFO mapred.Task: Task attempt_local_0001_r_000000_0 is allowed to commit now  
  37. 15/01/19 15:32:30 INFO output.FileOutputCommitter: Saved output of task ‘attempt_local_0001_r_000000_0‘ to hdfs://liaozhongmin:9000/out  
  38. 15/01/19 15:32:30 INFO mapred.LocalJobRunner: reduce > reduce  
  39. 15/01/19 15:32:30 INFO mapred.Task: Task ‘attempt_local_0001_r_000000_0‘ done.  
  40. 15/01/19 15:32:31 INFO mapred.JobClient:  map 100% reduce 100%  
  41. 15/01/19 15:32:31 INFO mapred.JobClient: Job complete: job_local_0001  
  42. 15/01/19 15:32:31 INFO mapred.JobClient: Counters: 19  
  43. 15/01/19 15:32:31 INFO mapred.JobClient:   File Output Format Counters   
  44. 15/01/19 15:32:31 INFO mapred.JobClient:     Bytes Written=40  
  45. 15/01/19 15:32:31 INFO mapred.JobClient:   FileSystemCounters  
  46. 15/01/19 15:32:31 INFO mapred.JobClient:     FILE_BYTES_READ=446  
  47. 15/01/19 15:32:31 INFO mapred.JobClient:     HDFS_BYTES_READ=140  
  48. 15/01/19 15:32:31 INFO mapred.JobClient:     FILE_BYTES_WRITTEN=131394  
  49. 15/01/19 15:32:31 INFO mapred.JobClient:     HDFS_BYTES_WRITTEN=40  
  50. 15/01/19 15:32:31 INFO mapred.JobClient:   File Input Format Counters   
  51. 15/01/19 15:32:31 INFO mapred.JobClient:     Bytes Read=70  
  52. 15/01/19 15:32:31 INFO mapred.JobClient:   Map-Reduce Framework  
  53. 15/01/19 15:32:31 INFO mapred.JobClient:     Reduce input groups=3  
  54. 15/01/19 15:32:31 INFO mapred.JobClient:     Map output materialized bytes=134  
  55. 15/01/19 15:32:31 INFO mapred.JobClient:     Combine output records=0  
  56. 15/01/19 15:32:31 INFO mapred.JobClient:     Map input records=8  
  57. 15/01/19 15:32:31 INFO mapred.JobClient:     Reduce shuffle bytes=0  
  58. 15/01/19 15:32:31 INFO mapred.JobClient:     Reduce output records=3  
  59. 15/01/19 15:32:31 INFO mapred.JobClient:     Spilled Records=16  
  60. 15/01/19 15:32:31 INFO mapred.JobClient:     Map output bytes=112  
  61. 15/01/19 15:32:31 INFO mapred.JobClient:     Total committed heap usage (bytes)=391118848  
  62. 15/01/19 15:32:31 INFO mapred.JobClient:     Combine input records=0  
  63. 15/01/19 15:32:31 INFO mapred.JobClient:     Map output records=8  
  64. 15/01/19 15:32:31 INFO mapred.JobClient:     SPLIT_RAW_BYTES=99  
  65. 15/01/19 15:32:31 INFO mapred.JobClient:     Reduce input records=8  

首先,我们看了Reduce端的日志,第一个信息我应该很容易能够很容易看出来,就是分组和reduce()函数处理都是在Shuffle完成之后才进行的。另外一点我们也非常容易看出,就是每次处理完一个分组数据就会去调用一次的reduce()函数对这个分组进行处理和输出。此外,说明一些分组函数的返回值问题,当返回0时才会被分到同一个组中。另外一点我们也可以看出来,一个分组中每合并n个值就会有n-1分组函数返回0值,也就是说进行了n-1次比较。

 

 

六、总结

本文主要从MapReduce框架执行的流程,去分析了如何去实现二次排序,通过代码进行了实现,并且对整个流程进行了验证。另外,要吐槽一下,网络上有很多文章都记录了MapReudce处理二次排序问题,但是对MapReduce框架整个处理流程的描述错漏很多,而且他们最终的流程描述也没有证据可以支撑。所以,对于网络上的学习资源不能够完全依赖,要融入自己的思想,并且要重要的观点进行代码或者实践的验证。另外,今天在一个hadoop交流群上听到少部分人在讨论,有了hive我们就不用学习些MapReduce程序?对这这个问题我是这么认为:我不相信写不好MapReduce程序的程序员会写好hive语句,最起码的他们对整个执行流程是一无所知的,更不用说性能问题了,有可能连最常见的数据倾斜问题的弄不清楚。

Hadoop二次排序及MapReduce处理流程实例详解

标签:

原文地址:http://www.cnblogs.com/thinkpad/p/5173739.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!