如何使用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一样。
v5net当前对香港和美国机房的走优质BGP+CN2网络的云服务器进行7折终身优惠促销,每个客户进线使用优惠码一次,额外有不限使用次数的终身9折优惠一枚!V5.NET Server提供的都是高端网络线路的机器,特别优化接驳全世界骨干网络,适合远程办公、跨境贸易、网站建设等用途。 官方网站:https://v5.net/cloud.html 7折优惠码:new,仅限新客户,每人仅限使用一次 9...
wordpress高级全行业大气外贸主题,wordpress通用全行业高级外贸企业在线询单自适应主题建站程序,完善的外贸企业建站功能模块 + 高效通用的后台自定义设置,更实用的移动设备特色功能模块 + 更适于欧美国外用户操作体验 大气简洁的网站风格设计 + 高效优化的网站程序结构,更利于Goolge等SEO搜索优化和站点收录排名。点击进入:wordpress高级全行业大气外贸主题主题价格:¥398...
SugarHosts 糖果主机商我们算是比较熟悉的,早年学会建站的时候开始就用的糖果虚拟主机,目前他们家还算是为数不多提供虚拟主机的商家,有提供香港、美国、德国等虚拟主机机房。香港机房CN2速度比较快,美国机房有提供优化线路和普通线路适合外贸业务。德国欧洲机房适合欧洲业务的虚拟主机。糖果主机商一般是不会发布黑五活动的,他们在圣圣诞节促销活动是有的,我们看到糖果主机商发布的圣诞节促销虚拟主机低至6折...
fileinputformat为你推荐
迅雷地址转换下载地址转换 怎么把别人的资源转换成迅雷地址 并且下载文件名改成自己想要的名字excel计算公式求excel计算公式大全请求超时请求超时是怎么回事请求超时DNS请求超时,怎么办?winhttp请问winhttp.dl是什么文件??winhttp什么是winhttp.dii匹配函数Excel中vlookup函数数据匹配怎么用应用雷达雷达是什么东西at89s52单片机有谁知道单片机如AT89c52,AT89s52具体是指什么含义啊?chrome系统Chromenbsp;OS是操作系统吗?
出租服务器 美国vps 万网免费域名 512m内存 中国特价网 七夕促销 常州联通宽带 yundun 河南移动梦网 免费的asp空间 德隆中文网 电信宽带测速软件 网络速度 广州服务器托管 学生机 ping值 paypal兑换 性能测试工具 ddos攻击小组 新浪轻博客 更多