fileinputformatHadoop,Combiner有什么用?

fileinputformat  时间:2021-06-08  阅读:()

如何使用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一样。

CloudCone,美国洛杉矶独立服务器特价优惠,美国洛杉矶MC机房,100Mbps带宽不限流量,可选G口,E3-1270 v2处理器32G内存1Gbps带宽,69美元/月

今天CloudCone发布了最新的消息,推送了几款特价独立服务器/杜甫产品,美国洛杉矶MC机房,分配100Mbps带宽不限流量,可以选择G口限制流量计划方案,存储分配的比较大,选择HDD硬盘的话2TB起,MC机房到大陆地区线路还不错,有需要美国特价独立服务器的朋友可以关注一下。CloudCone怎么样?CloudCone服务器好不好?CloudCone值不值得购买?CloudCone是一家成立于2...

HTTPS加密协议端口默认是多少且是否支持更换端口访问

看到群里网友们在讨论由于不清楚的原因,有同学的网站无法访问。他的网站是没有用HTTPS的,直接访问他的HTTP是无法访问的,通过PING测试可以看到解析地址已经比较乱,应该是所谓的DNS污染。其中有网友提到采用HTTPS加密证书试试。因为HTTP和HTTPS走的不是一个端口,之前有网友这样测试过是可以缓解这样的问题。这样通过将网站绑定设置HTTPS之后,是可以打开的,看来网站的80端口出现问题,而...

knownhost西雅图/亚特兰大/阿姆斯特丹$5/月,2个IP1G内存/1核/20gSSD/1T流量

美国知名管理型主机公司,2006年运作至今,虚拟主机、VPS、云服务器、独立服务器等业务全部采用“managed”,也就是人工参与度高,很多事情都可以人工帮你处理,不过一直以来价格也贵。也不知道knownhost什么时候开始运作无管理型业务的,估计是为了扩展市场吧,反正是出来较长时间了。闲来无事,那就给大家介绍下“unmanaged VPS”,也就是无管理型VPS,低至5美元/月,基于KVM虚拟,...

fileinputformat为你推荐
excel计算公式请教在excel中如何用求和公式mac地址克隆路由器的Mac地址克隆有什么作用?网络审计网络安全审计系统的背景搜索引擎的概念7 什么是搜索引擎?如何在Internet上搜索图片和文字资料的?国产操作系统下载国产操作系统要钱吗12种颜色油画的基本12种颜色是什么数据挖掘项目将来想从事统计分析和数据挖掘,需要哪些技能labelforhtml中label是什么意思啊?腾讯贴吧QQ贴吧如何发帖什么是网络地址什么是IP地址?
免费二级域名申请 中文域名交易中心 typecho 商家促销 论坛空间 免费mysql 空间出租 169邮箱 卡巴斯基试用版 福建铁通 太原网通测速平台 免费私人服务器 彩虹云 空间首页登陆 空间登录首页 上海电信测速 帽子云排名 浙江服务器 lamp架构 镇江高防 更多