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一样。

v5server:香港+美国机房,优质CN2网络云服务器,7折优惠,低至35元/月

v5net当前对香港和美国机房的走优质BGP+CN2网络的云服务器进行7折终身优惠促销,每个客户进线使用优惠码一次,额外有不限使用次数的终身9折优惠一枚!V5.NET Server提供的都是高端网络线路的机器,特别优化接驳全世界骨干网络,适合远程办公、跨境贸易、网站建设等用途。 官方网站:https://v5.net/cloud.html 7折优惠码:new,仅限新客户,每人仅限使用一次 9...

wordpress外贸企业主题 wordpress高级全行业大气外贸主题

wordpress高级全行业大气外贸主题,wordpress通用全行业高级外贸企业在线询单自适应主题建站程序,完善的外贸企业建站功能模块 + 高效通用的后台自定义设置,更实用的移动设备特色功能模块 + 更适于欧美国外用户操作体验 大气简洁的网站风格设计 + 高效优化的网站程序结构,更利于Goolge等SEO搜索优化和站点收录排名。点击进入:wordpress高级全行业大气外贸主题主题价格:¥398...

SugarHosts糖果主机圣诞节促销 美国/香港虚拟主机低至6折

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攻击小组 新浪轻博客 更多