首页 > 代码库 > hadoop编程小技巧(3)---自定义分区类Partitioner
hadoop编程小技巧(3)---自定义分区类Partitioner
Hadoop代码测试环境:Hadoop2.4
原理:在Hadoop的MapReduce过程中,Mapper读取处理完成数据后,会把数据发送到Partitioner,由Partitioner来决定每条记录应该送往哪个reducer节点,默认使用的是HashPartitioner,其核心代码如下:
/** Use {@link Object#hashCode()} to partition. */ public int getPartition(K2 key, V2 value, int numReduceTasks) { return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks; }
getPartition的输出参数就是Mapper输出的key和value,然后针对这样的输入,采用key的hash值来判断当前记录应该被分为哪个reducer中(如果numReduceTasks为1,那么其实就只有一个分组,这里使用的是取模运算)。
应用场景:假如,我们事前已经对我们的数据以及Mapper处理后的输出数据都有一个很好的了解,那么其实我们可以控制记录应该送往哪个reducer进行处理,这样方便我们采取某种策略,来使reducer处理的数据量基本相同,达到一种均衡的效果。这样,对我们数据处理的效率也会有很大的提高。当然,这种策略需要我们对数据的了解会比较高。
实例:
首先自定义Partitioner(假设,我们需要把值value以A开头的数据分入一个reducer,那么可以使用下面的Partitioner),可以参考HashPartitioner,:
package fz.partitioner; import org.apache.hadoop.mapreduce.Partitioner; public class MyPartitioner<K1, V1> extends Partitioner<K1, V1> { @Override public int getPartition(K1 key, V1 value, int numPartitions) { String tmpValue = http://www.mamicode.com/value.toString();>
接着定义一个什么都不做的MR任务,只是简单的读取数据,调用自定义的MyPartitioner,然后查看输出结果,是否是我们需要的。定义driver:
package fz.partitioner; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; public class PartitionerDriver extends Configured implements Tool { @Override public int run(String[] arg0) throws Exception { Configuration conf = getConf(); if(arg0.length!=2){ System.err.println("Usage:\nfz.partitioner.PartitionerDriver <in> <out>"); return -1; } // System.out.println(conf.get("fs.defaultFS")); Path in = new Path(arg0[0]); Path out= new Path(arg0[1]); out.getFileSystem(conf).delete(out, true); Job job = Job.getInstance(conf,"test partitioner"); job.setJarByClass(getClass()); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); job.setPartitionerClass(MyPartitioner.class); job.setMapperClass(Mapper.class); job.setMapOutputKeyClass(LongWritable.class); job.setMapOutputValueClass(Text.class); job.setOutputKeyClass(LongWritable.class); job.setOutputValueClass(Text.class); job.setReducerClass(Reducer.class); job.setNumReduceTasks(2); FileInputFormat.setInputPaths(job, in); FileOutputFormat.setOutputPath(job, out); return job.waitForCompletion(true)?0:-1; } public static void main(String[] args) throws Exception { ToolRunner.run(new Configuration(), new PartitionerDriver(),args); } }这里设置了自定义的MyPartitioner,同时设置reducer的个数为2。运行MR程序,查看结果:
通过上面的结果对比,可以发现,(在mapper中并没有任何的逻辑操作),输出的数据仅仅是设置了Partitioner,然后就可以达到不同数据输出到不同的reducer的效果。
总结:如果对数据的整体有很好的了解,可以使用自定义Partitioner来达到reducer的负载均衡,提高效率。
分享,成长,快乐
转载请注明blog地址:http://blog.csdn.net/fansy1990
声明:以上内容来自用户投稿及互联网公开渠道收集整理发布,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任,若内容有误或涉及侵权可进行投诉: 投诉/举报 工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。