fileinputformatmapreduce 键值对怎么定义的

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

hadoop怎么重新启动tasktracker

如果是某个TaskTracker挂掉了想要重启,比较稳妥的方式是在master直接运行一次start-mapred.sh(或者start-all.sh也可以)脚本。

脚本会忽略正在正常运行的hadoop进程,并在没有运行TaskTracker的节点上启动TaskTracker。

如果是想停止某个节点上的Taskracker再启动,直接登录到该节点,kill掉TaskTracker进程,再返回master运行start-mapred.sh。

Hadoop,怎么实现多个输入路径的输入?

1.多路径输入 1)FileInputFormat.addInputPath 多次调用加载不同路径 .apache.hadoop.mapreduce.lib.input.FileInputFormat; .apache.hadoop.mapreduce.lib.output.FileOutputFormat; String?in0?=?args[0]; String?in1?=?args[1]; String?out?=?args[2]; FileInputFormat.addInputPath(job,new?Path(in0)); FileInputFormat.addInputPath(job,new?Path(in1)); FileOutputFormat.setOutputPath(job,new?Path(out));2)FileInputFormat.addInputPaths一次调用加载 多路径字符串用逗号隔开 FileInputFormat.addInputPaths(job, "hdfs://RS5-112:9000/cs/path1,hdfs://RS5-112:9000/cs/path2"); 2.多种输入 MultipleInputs可以加载不同路径的输入文件,并且每个路径可用不同的maper MultipleInputs.addInputPath(job, new Path("hdfs://RS5-112:9000/cs/path1"), TextInputFormat.class,MultiTypeFileInput1Mapper.class); MultipleInputs.addInputPath(job, new Path("hdfs://RS5-112:9000/cs/path3"), TextInputFormat.class,MultiTypeFileInput3Mapper.class);

Hadoop是怎么分块的

hadoop的分块有两部分,其中第一部分更为人熟知一点。

第一部分就是数据的划分(即把File划分成Block),这个是物理上真真实实的进行了划分,数据文件上传到HDFS里的时候,需要划分成一块一块,每块的大小由hadoop-default.xml里配置选项进行划分。

<property> <name>dfs.block.size</name> <value>67108864</value> <description>The default block size for new files.</description> </property> 这个就是默认的每个块64MB。

数据划分的时候有冗余,个数是由 <property> <name>dfs.replication</name> <value>3</value> <description>Default block replication. The actual number of replications can be specified when the file is created. The default is used if replication is not specified in create time. </description> </property> 指定的。

具体的物理划分步骤要看Namenode,这里要说的是更有意思的hadoop中的第二种划分。

在hadoop中第二种划分是由InputFormat这个接口来定义的,其中有个getSplits方法。

这里就有了一个新的不为人熟知的概念:Split。

Split的作用是什么,Split和Block是什么关系,下面就可以说明清楚。

在Hadoop0.1中,split划分是在JobTracker端完成的,发生在JobInitThread对JobInProgress调用inittasks()的时候;而在0.18.3中是由JobClient完成的,JobClient划分好后,把split.file写入hdfs里,到时候jobtracker端只需要读这个文件,就知道Split是怎么划分的了。

第二种划分只是一种逻辑上划分,目的是为了让Map Task更好的获取数据输入,仔细分析如下这个场景: File 1 : Block11, Block 12, Block 13, Block 14, Block 15 File 2 : Block21, Block 22, Block 23 File1有5个Block,最后一个Block当然可能小于64MB;File2有3个Block 如果用户在程序中指定map tasks的个数,比如说是2(如果不指定的话maptasks个数默认是1),那么在 FileInputFormat(最常见的InputFormat实现)的getSplits方法中,首先会计算totalSize=8(可以对照源码看看,注意getSplits这个函数里的计量单位是Block个数,而不是Byte个数,后面有个变量叫bytesremaining仍然表示剩余的Block个数,有些变量名让人无语),然后会计算goalSize=totalSize/numSplits=4,对于File1,计算一个Split有多少个Block是这样计算的 long splitSize =puteSplitSize(goalSize, minSize, blockSize); protected puteSplitSize(long goalSize, long minSize, long blockSize) { return Math.max(minSize, Math.min(goalSize, blockSize)); } 这里minSize是1(说明了一个Split至少包含一个Block,不会出现一个Split包含零点几个Block的情况),计算得出splitSize=4,所以接下来Split划分是这样分的: Split 1: Block11, Block12, Block13,Block14 Split 2: Block15 Split 3: Block21, Block22, Block23 那用户指定的map个数是2,出现了三个split怎么办?在JobInProgress里其实maptasks的个数是根据Splits的长度来指定的,所以用户指定的map个数只是个参考。

可以参看JobInProgress: initTasks() 里的代码: try { splits = JobClient.readSplitFile(splitFile); } finally { splitFile.close(); } numMapTasks = splits.length; maps = new TaskInProgress[numMapTasks]; 所以问题就很清晰了,还如果用户指定了20个map作业,那么最后会有8个Split(每个Split一个Block),所以最后实际上就有8个MapTasks,也就是说maptask的个数是由splits的长度决定的。

几个简单的结论: 1. 一个split不会包含零点几或者几点几个Block,一定是包含大于等于1个整数个Block 2. 一个split不会包含两个File的Block,不会跨越File边界 3. split和Block的关系是一对多的关系 4. maptasks的个数最终决定于splits的长度 还有一点需要说明,在FileSplit类中,有一项是private String[] hosts; 看上去是说明这个FileSplit是放在哪些机器上的,实际上hosts里只是存储了一个Block的冗余机器列表。

比如上面例子中的Split 1: Block11, Block12, Block13,Block14,这个FileSplit中的hosts里最终存储的是Block11本身和其冗余所在的机器列表,也就是说Block12,Block13,Block14存在哪些机器上没有在FileSplit中记录。

FileSplit中的这个属性有利于调度作业时候的数据本地性问题。

如果一个tasktracker前来索取task,jobtracker就会找个task给他,找到一个maptask,得先看这个task的输入的FileSplit里hosts是否包含tasktracker所在机器,也就是判断和该tasktracker同时存在一个机器上的datanode是否拥有FileSplit中某个Block的备份。

但总之,只能牵就一个Block,其他Block就从网络上传罢。

mapreduce怎么导入hbace

1、先看一个标准的hbase作为数据读取源和输出源的样例: [java] view plaincopy在CODE上查看代码片派生到我的代码片 Configuration conf = HBaseConfiguration.create(); Job job = new Job(conf, "job name "); job.setJarByClass(test.class); Scan scan = new Scan(); TableMapReduceUtil.initTableMapperJob(inputTable, scan, mapper.class, Writable.class, Writable.class, job); TableMapReduceUtil.initTableReducerJob(outputTable, reducer.class, job); job.waitForCompletion(true); 首先创建配置信息和作业对象,设置作业的类。

这些和正常的mapreduce一样, [java] view plaincopy在CODE上查看代码片派生到我的代码片 Configuration conf = HBaseConfiguration.create(); Job job = new Job(conf, "job name "); job.setJarByClass(test.class); Scan scan = new Scan(); TableMapReduceUtil.initTableMapperJob(inputTable, scan, mapper.class, Writable.class, Writable.class, job); TableMapReduceUtil.initTableReducerJob(outputTable, reducer.class, job); job.waitForCompletion(true); 唯一不一样的就是数据源的说明部分,TableMapReduceUtil的initTableMapperJob和initTableReducerJob方法来实现。

用如上代码: 数据输入源是hbase的inputTable表,执行mapper.class进行map过程,输出的key/value类型是ImmutableBytesWritable和Put类型,最后一个参数是作业对象。

需要指出的是需要声明一个扫描读入对象scan,进行表扫描读取数据用,其中scan可以配置参数,这里为了例子简单不再详述。

数据输出目标是hbase的outputTable表,输出执行的reduce过程是reducer.class类,操作的作业目标是job。

与map比缺少输出类型的标注,因为他们不是必要的,看过源代码就知道mapreduce的TableRecordWriter中write(key,value)方法中,key值是没有用到的。

value只能是Put或者Delete两种类型,write方法会自行判断并不用用户指明。

接下来就是mapper类: [java] view plaincopy在CODE上查看代码片派生到我的代码片 public class mapper extends TableMapper { public void map(Writable key, Writable value, Context context) throws IOException, InterruptedException { //mapper逻辑 context.write(key, value); } } } 继承的是hbase中提供的TableMapper类,其实这个类也是继承的MapReduce类。

后边跟的两个泛型参数指定类型是mapper输出的数据类型,该类型必须继承自Writable类,例如可能用到的put和delete就可以。

需要注意的是要和initTableMapperJob方法指定的数据类型一直。

该过程会自动从指定hbase表内一行一行读取数据进行处理。

然后reducer类: [java] view plaincopy在CODE上查看代码片派生到我的代码片 public class countUniteRedcuer extends TableReducer { public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { //reducer逻辑 context.write(null, put or delete); } } reducer继承的是TableReducer类。

后边指定三个泛型参数,前两个必须对应map过程的输出key/value类型,第三个必须是put或者delete。

write的时候可以把key写null,它是不必要的。

这样reducer输出的数据会自动插入outputTable指定的表内。

2、有时候我们需要数据源是hdfs的文本,输出对象是hbase。

这时候变化也很简单: [java] view plaincopy 你会发现只需要像平常的mapreduce的作业声明过程一样,指定mapper的执行类和输出key/value类型,指定FileInputFormat.setInputPaths的数据源路径,输出声明不变。

便完成了从hdfs文本读取数据输出到hbase的命令声明过程。

mapper和reducer如下: [java] view plaincopy在CODE上查看代码片派生到我的代码片 Configuration conf = HBaseConfiguration.create(); Job job = new Job(conf, "job name "); job.setJarByClass(test.class); job.setMapperClass(mapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(LongWritable.class); FileInputFormat.setInputPaths(job, path); TableMapReduceUtil.initTableReducerJob(tableName, reducer.class, job); [java] view plaincopy在CODE上查看代码片派生到我的代码片 public class mapper extends Mapper { public void map(LongWritable key, Text line, Context context) { //mapper逻辑 context.write(k, one); } } public class redcuer extends TableReducer { public void reduce(Writable key, Iterable values, Context context) throws IOException, InterruptedException { //reducer逻辑 context.write(null, put or delete); } } [java] view plaincopy mapper还依旧继承原来的MapReduce类中的Mapper即可。

同样注意这前后数据类型的key/value一直性。

[java] view plaincopy 3、最后就是从hbase中的表作为数据源读取,hdfs作为数据输出,简单的如下: [java] view plaincopy在CODE上查看代码片派生到我的代码片 Configuration conf = HBaseConfiguration.create(); Job job = new Job(conf, "job name "); job.setJarByClass(test.class); Scan scan = new Scan(); TableMapReduceUtil.initTableMapperJob(inputTable, scan, mapper.class, Writable.class, Writable.class, job); job.setOutputKeyClass(Writable.class); job.setOutputValueClass(Writable.class); FileOutputFormat.setOutputPath(job, Path); job.waitForCompletion(true); mapper和reducer简单如下: [java] view plaincopy在CODE上查看代码片派生到我的代码片Configuration conf = HBaseConfiguration.create(); Job job = new Job(conf, "job name "); job.setJarByCl ass(test.class); Scan scan = new Scan(); TableMapReduceUtil.initTableMapperJob(inputTable, scan, mapper.class, Writable.class, Writable.class, job); TableMapReduceUtil.initTableReducerJob(outputTable, reducer.class, job); job.waitForCompletion(true); Configuration conf = HBaseConfiguration.create(); Job job = new Job(conf, "job name "); job.setJarByClass(test.class); Scan scan = new Scan(); TableMapReduceUtil.initTableMapperJob(inputTable, scan, mapper.class, Writable.class, Writable.class, job); TableMapReduceUtil.initTableReducerJob(outputTable, reducer.class, job); job.waitForCompletion(true); 唯一不一样的就是数据源的说明部分,TableMapReduceUtil的initTableMapperJob和initTableReducerJob方法来实现。

public class mapper extends TableMapper { public void map(Writable key, Writable value, Context context) throws IOException, InterruptedException { //mapper逻辑 context.write(key, value); } } } public class countUniteRedcuer extends TableReducer { public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { //reducer逻辑 context.write(null, put or delete); } }

mapreduce 键值对怎么定义的

一般情况下Mapreduce输出的键值对是以制表符 为分隔符的,如下图所示: 但有时候我们像将其设置为其它的分隔符输出,比如",",如下图所示: 此时可以在Mapreduce的主函数中添加如下的两行代码: [java] view plain copy print? conf.set("mapred.textoutputformat.ignoreseparator","true"); conf.set("mapred.textoutputformat.separator",","); 具体如下的WordCount程序: [java] view plain copy print? import java.io.IOException; import java.util.StringTokenizer; .apache.hadoop.conf.Configuration; .apache.hadoop.fs.Path; .apache.hadoop.io.IntWritable; .apache.hadoop.io.Text; .apache.hadoop.mapreduce.Job; .apache.hadoop.mapreduce.Mapper; .apache.hadoop.mapreduce.Reducer; .apache.hadoop.mapreduce.lib.input.FileInputFormat; .apache.hadoop.mapreduce.lib.output.FileOutputFormat; .apache.hadoop.util.GenericOptionsParser; public class WordCount { public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } } public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); //设置MapReduce的输出的分隔符为逗号 conf.set("mapred.textoutputformat.ignoreseparator", "true"); conf.set("mapred.textoutputformat.separator", ","); String[] otherArgs = new GenericOptionsParser(conf, args) .getRemainingArgs(); if (otherArgs.length != 2) { System.err.println("Usage: wordcount <in> <out>"); System.exit(2); } Job job = new Job(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }

美国云服务器 1核 1G 30M 50元/季 兆赫云

【双十二】兆赫云:全场vps季付六折优惠,低至50元/季,1H/1G/30M/20G数据盘/500G流量/洛杉矶联通9929商家简介:兆赫云是一家国人商家,成立2020年,主要业务是美西洛杉矶联通9929线路VPS,提供虚拟主机、VPS和独立服务器。VPS采用KVM虚拟架构,线路优质,延迟低,稳定性强。是不是觉得黑五折扣力度不够大?还在犹豫徘徊中?这次为了提前庆祝双十二,特价推出全场季付六折优惠。...

spinservers:圣何塞物理机7.5折,$111/月,2*e5-2630Lv3/64G内存/2T SSD/10Gbps带宽

spinservers美国圣何塞机房的独立服务器补货120台,默认接入10Gbps带宽,给你超高配置,这价格目前来看好像真的是无敌手,而且可以做到下单后30分钟内交货,都是预先部署好了的。每一台机器用户都可以在后台自行安装、重装、重启、关机操作,无需人工参与! 官方网站:https://www.spinservers.com 比特币、信用卡、PayPal、支付宝、webmoney、Payssi...

HostYun 新上美国CN2 GIA VPS 月15元

HostYun 商家以前是玩具主机商,这两年好像发展还挺迅速的,有点在要做点事情的味道。在前面也有多次介绍到HostYun商家新增的多款机房方案,价格相对还是比较便宜的。到目前为止,我们可以看到商家提供的VPS主机包括KVM和XEN架构,数据中心可选日本、韩国、香港和美国的多个地区机房,电信双程CN2 GIA线路,香港和日本机房,均为国内直连线路。近期,HostYun上线低价版美国CN2 GIA ...

fileinputformat为你推荐
csonline2反恐精英online2什么时候出eagleeye《鹰眼》的男主角是谁?知识分享平台知识付费平台有哪些?oa办公系统下载办公软件下载哪里可以下载啊?主要是公司准备搞这个东西,我先看看什么情况vga接口定义VGA接口通常用来连接哪些设备,各个脚代表什么意思,它的连线是如何焊接的?腾讯年终奖腾讯外聘员工与正式员工的区别是什么?国产操作系统下载国产操作系统要钱吗模式识别算法算法是编程么?考研学模式识别方向,编程多么?assemblyinfoasp.net这几个文件是干什么的?鄂n鄂N的车牌是那里的?
网址域名注册 万网域名代理 132邮箱 dns是什么 simcentric highfrequency 500m空间 阿里云浏览器 hostker 699美元 南通服务器 免费高速空间 银盘服务是什么 游戏服务器出租 免费稳定空间 黑科云 杭州电信 netvigator godaddy中文 rsync 更多