如何使用Hadoop的Partitioner
Hadoop里面的MapReduce编程模型,非常灵活,大部分环节我们都可以重写它的API,来灵活定制我们自己的一些特殊需求。
今天散仙要说的这个分区函数Partitioner,也是一样如此,下面我们先来看下Partitioner的作用:
对map端输出的数据key作一个散列,使数据能够均匀分布在各个reduce上进行后续操作,避免产生热点区。
Hadoop默认使用的分区函数是Hash Partitioner,源码如下:
/** Partition keys by their {@link Object#hashCode()}. */
public class HashPartitioner<K, V> extends Partitioner<K, V> {
/** Use {@link Object#hashCode()} to partition. */
public int getPartition(K key, V value,
int numReduceTasks) {
//默认使用key的hash值与上int的最大值,避免出现数据溢出 的情况
return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
}
}
大部分情况下,我们都会使用默认的分区函数,但有时我们又有一些,特殊的需求,而需要定制Partition来完成我们的业务,案例如下:
对如下数据,按字符串的长度分区,长度为1的放在一个,2的一个,3的各一个。
河南省;1
河南;2
中国;3
中国人;4
大;1
小;3
中;11
这时候,我们使用默认的分区函数,就不行了,所以需要我们定制自己的Partition,首先分析下,我们需要3个分区输出,所以在设置reduce的个数时,一定要设置为3,其次在partition里,进行分区时,要根据长度具体分区,而不是根据字符串的hash码来分区。
核心代码如下:
/**
* Partitioner
*
*
* */
public static class PPartition extends Partitioner<Text, Text>{
@Override
public int getPartition(Text arg0, Text arg1, int arg2) {
/**
* 自定义分区,实现长度不同的字符串,分到不同的reduce里面
*
* 现在只有3个长度的字符串,所以可以把reduce的个数设置为3
* 有几个分区,就设置为几
* */
String key=arg0.toString();
if(key.length()==1){
return 1%arg2;
}else if(key.length()==2){
return 2%arg2;
}else if(key.length()==3){
return 3%arg2;
}
return 0;
}
}
全部代码如下:
.partition.test;
import java.io.IOException;
.apache.hadoop.fs.FileSystem;
.apache.hadoop.fs.Path;
.apache.hadoop.io.LongWritable;
.apache.hadoop.io.Text;
.apache.hadoop.mapred.JobConf;
.apache.hadoop.mapreduce.Job;
.apache.hadoop.mapreduce.Mapper;
.apache.hadoop.mapreduce.Partitioner;
.apache.hadoop.mapreduce.Reducer;
.apache.hadoop.mapreduce.lib.db.DBConfiguration;
.apache.hadoop.mapreduce.lib.db.DBInputFormat;
.apache.hadoop.mapreduce.lib.input.FileInputFormat;
.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
.qin.operadb.PersonRecoder;
.qin.operadb.ReadMapDB;
/**
* @author qindongliang
*
* 大数据交流群:376932160
*
*
* **/
public class MyTestPartition {
/**
* map任务
*
* */
public static class PMapper extends Mapper<LongWritable, Text, Text, Text>{
@Override
protected void map(LongWritable key, Text value,Context context)
throws IOException, InterruptedException {
// System.out.println("进map了");
//mos.write(namedOutput, key, value);
String ss[]=value.toString().split(";");
context.write(new Text(ss[0]), new Text(ss[1]));
}
}
/**
* Partitioner
*
*
* */
public static class PPartition extends Partitioner<Text, Text>{
@Override
public int getPartition(Text arg0, Text arg1, int arg2) {
/**
* 自定义分区,实现长度不同的字符串,分到不同的reduce里面
*
* 现在只有3个长度的字符串,所以可以把reduce的个数设置为3
* 有几个分区,就设置为几
* */
String key=arg0.toString();
if(key.length()==1){
return 1%arg2;
}else if(key.length()==2){
return 2%arg2;
}else if(key.length()==3){
return 3%arg2;
}
return 0;
}
}
/***
* Reduce任务
*
* **/
public static class PReduce extends Reducer<Text, Text, Text, Text>{
@Override
protected void reduce(Text arg0, Iterable<Text> arg1, Context arg2)
throws IOException, InterruptedException {
String key=arg0.toString().split(",")[0];
System.out.println("key==> "+key);
for(Text t:arg1){
//System.out.println("Reduce: "+arg0.toString()+" "+t.toString());
arg2.write(arg0, t);
}
}
}
public static void main(String[] args) throws Exception{
JobConf conf=new JobConf(ReadMapDB.class);
//Configuration conf=new Configuration();
conf.set("mapred.job.tracker","192.168.75.130:9001");
//读取person中的数据字段
conf.setJar("tt.jar");
//注意这行代码放在最前面,进行初始化,否则会报
/**Job任务**/
Job job=new Job(conf, "testpartion");
job.setJarByClass(MyTestPartition.class);
System.out.println("模式: "+conf.get("mapred.job.tracker"));;
// job.setCombinerClass(PCombine.class);
job.setPartitionerClass(PPartition.class);
job.setNumReduceTasks(3);//设置为3
job.setMapperClass(PMapper.class);
// MultipleOutputs.addNamedOutput(job, "hebei", TextOutputFormat.class, Text.class, Text.class);
// MultipleOutputs.addNamedOutput(job, "henan", TextOutputFormat.class, Text.class, Text.class);
job.setReducerClass(PReduce.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
String path="hdfs://192.168.75.130:9000/root/outputdb";
FileSystem fs=FileSystem.get(conf);
Path p=new Path(path);
if(fs.exists(p)){
fs.delete(p, true);
System.out.println("输出路径存在,已删除!");
}
FileInputFormat.setInputPaths(job, "hdfs://192.168.75.130:9000/root/input");
FileOutputFormat.setOutputPath(job,p );
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}如何使用eclipse调试Hadoop作业
将hadoop开发包里面的相关jar导进工程就行,
至于想调试,就看hadoop计数器返回到eclipse里的内容就可以了.
不过有一点,
如果调试的是MapReduce,速度可能不快.Hadoop,Combiner有什么用?
Combiner,Combiner号称本地的Reduce,Reduce最终的输入,是Combiner的输出。
Combiner是用reducer来定义的,多数的情况下Combiner和reduce处理的是同一种逻辑,所以job.setCombinerClass()的参数可以直接使用定义的reduce。
当然也可以单独去定义一个有别于reduce的Combiner,继承Reducer,写法基本上定义reduce一样。
官方网站:点击访问王小玉网络官网活动方案:买美国云服务器就选MF.0220.CN 实力 强 强 强!!!杭州王小玉网络 旗下 魔方资源池 “我亏本你引流活动 ” mf.0220.CNCPU型号内存硬盘美国CERA机房 E5 2696v2 2核心8G30G总硬盘1个独立IP19.9元/月 续费同价mf.0220.CN 购买湖北100G防御 E5 2690v2 4核心4G...
易探云怎么样?易探云香港云服务器比较有优势,他家香港BGP+CN2口碑不错,速度也很稳定。尤其是今年他们动作很大,推出的香港云服务器有4个可用区价格低至18元起,试用过一个月的用户基本会续费,如果年付的话还可以享受8.5折或秒杀价格。今天,云服务器网(yuntue.com)小编推荐一下易探云国内云服务器优惠活动,北京和深圳这二个机房的云服务器2核2G5M带宽低至330.66元/年,还有高配云服务器...
HostKvm是一家成立于2013年的国外主机服务商,主要提供基于KVM架构的VPS主机,可选数据中心包括日本、新加坡、韩国、美国、中国香港等多个地区机房,均为国内直连或优化线路,延迟较低,适合建站或者远程办公等。目前商家发布了夏季特别促销活动,针对香港国际/韩国机房VPS主机提供7折优惠码,其他机房全场8折,优惠后2GB内存套餐月付5.95美元起。下面分别列出几款主机套餐配置信息。套餐:韩国KR...
fileinputformat为你推荐
刘建平现在赫章县委书记是谁宝应中学宝应初级中学有哪些iso20000认证什么是ISO20000-IT服务管理体系iso20000认证为什么ISO20000认证能够风靡全球?它对整个企业的发展有什么好处?以图搜人怎样人肉搜人?只知道他名字和当初居住的地址华为总裁女儿为啥姓孟孟姜女为什么不姓孟?vga接口定义VGA接口的15针分别接什么?seo优化技术什么是SEO优化,seo优化有什么用?防火墙排名什么防火墙最好数据统计分析表如何用Excel做数据分析?
主机点评 rackspace 美国主机代购 xfce 大容量存储 java虚拟主机 网站卫士 世界测速 台湾google web应用服务器 百度云空间 服务器防火墙 域名转入 购买空间 广州服务器托管 789电视剧网 SmartAXMT800 phpwind论坛 域名商城 vpsaa 更多