码迷，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）
登录后才能评论！
分享档案
更多>
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
迷上了代码！