Hadoop（三）自定义combiner和partitioner

103 阅读 0 评论 68 点赞

我是靠谱客的博主欢喜玉米，最近开发中收集的这篇文章主要介绍Hadoop（三）自定义combiner和partitioner，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

Hadoop提供了缺省的Partition来完成map的输出向reduce分发处理。有时也需要自定义partition来将相同key值的数据分发到同一个reduce处理，为了减少map过程输出的中间结果键值对的数量，降低网络数据通信开销，用户也可以自定制combiner过程。

自定制Partition过程：

在mapreduce中，partition用于决定Map节点输出将被分到哪个Reduce节点，MapReduce提供的缺省Partition是HashPartition,他根据每条数据的主键值进行hash操作，获得一个hash码，然后对当前的分区数量进行取模计算，以此决定分发到哪个reduce节点。用户自定制的Partition，重载了getPartition()方法，对于上篇的倒排索引复合键，将<itr.nextToken()+":"+split.getPath().toString().substring(splitIndex),"1">拆开.代码：

[cpp] 
view plain
copy

 public class NewPartition extends HashPartitioner<Text,Text>{  
             String keyinfo;  
             public int getPartition(Text key,Text value,int numReducerTasks){  
                 keyinfo = key.toString().split(":")[0];  
                 return super.getPartition(new Text(keyinfo), value, numReducerTasks);  
             }  
         }  

自定制的partition使用：

job.setPartitionClass(NewPartition.class);
而用户自定制的combiner，借鉴上篇的倒排索引的部分代码：

[java] 
view plain
copy

 public static class combiner extends Reducer<Text,Text,Text,Text>{  
         private Text info = new Text(); //为了拆分 key值 准备存储新的value值  
         public void reduce(Text key,Iterable<Text>values,Context context) throws IOException, InterruptedException{  
             int sum = 0;  
             for(Text val:values){  
                 sum += Integer.parseInt(val.toString());  
             }  
             int splitIndex = key.toString().indexOf(":");  
             info.set(key.toString().substring(splitIndex+1)+":"+sum);   //新的value值  
             key.set(key.toString().substring(0, splitIndex));  
             context.write(key, info);  
         }  
     }  