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;
  }

应用场景：假如，我们事前已经对我们的数据以及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 = value.toString();
		// 假设value值以A开头。那么就把数据发送到当中一个reducer，否则发送到另外的一个;
		if(tmpValue!=null&&tmpValue.indexOf("A")==0){
			return 0;
		}
		return 1;
	}

}

定义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);
	}
}

执行MR程序，查看结果：

通过上面的结果对照，能够发现，（在mapper中并没有不论什么的逻辑操作）。输出的数据仅仅是设置了Partitioner，然后就能够达到不同数据输出到不同的reducer的效果。

总结：假设对数据的总体有非常好的了解。能够使用自己定义Partitioner来达到reducer的负载均衡，提高工作效率。

分享，成长，高兴

转载请注明blog地址：http://blog.csdn.net/fansy1990