如何使用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一样。
目前舍利云服务器的主要特色是适合seo和建站,性价比方面非常不错,舍利云的产品以BGP线路速度优质稳定而著称,对于产品的线路和带宽有着极其严格的讲究,这主要表现在其对母鸡的超售有严格的管控,与此同时舍利云也尽心尽力为用户提供完美服务。目前,香港cn2云服务器,5M/10M带宽,价格低至30元/月,可试用1天;;美国cera云服务器,原生ip,低至28元/月起。一、香港CN2云服务器香港CN2精品线...
艾云怎么样?艾云是一家去年年底成立的国人主机商家,商家主要销售基于KVM虚拟架构的VPS服务,机房目前有美国洛杉矶、圣何塞和英国伦敦,目前商家推出了一些年付特价套餐,性价比非常高,洛杉矶套餐低至85元每年,给500M带宽,可解奈飞,另外圣何塞也有特价机器;1核/1G/20G SSD/3T/2.5Gbps,有需要的朋友以入手。点击进入:艾云官方网站艾云vps促销套餐:KVM虚拟架构,自带20G的防御...
企鹅小屋怎么样?企鹅小屋最近针对自己的美国cn2 gia套餐推出了2个优惠码:月付7折和年付6折,独享CPU,100%性能,三网回程CN2 GIA网络,100Mbps峰值带宽,用完优惠码1G内存套餐是年付240元,线路方面三网回程CN2 GIA。如果新购IP不能正常使用,请在开通时间60分钟内工单VPS技术部门更换正常IP;特价主机不支持退款。点击进入:企鹅小屋官网地址企鹅小屋优惠码:年付6折优惠...
fileinputformat为你推荐
performclick模拟点击事件 如何禁用声音 mView.performClick();应用雷达雷达是干什么用的?vga接口定义电脑显示屏VGA是什么意思12种颜色12种颜色的英语怎么写,用中文怎么读xcelsiussap bi是什么kjavaJava与KJava有什么区别吗.?河北云办税厅用小度怎么打开河北教育资讯云平台?软件开发技术文档软件开发技术的目录病毒分析网站谁能给我个防电脑病毒的网站?着急!tokenstream微信公众平台上,Token是什么?怎么填写?
域名注册使用godaddy 北京服务器租用 edis 名片模板psd 服务器是干什么的 中国电信测速网 服务器监测 paypal注册教程 免费私人服务器 德讯 美国迈阿密 万网注册 免费网络空间 新网dns cx域名 windowsserverr2 iptables 什么是云主机 主机响 web服务器软件下载 更多