本篇内容介绍了“hadoop多文件输出新旧API的方法是什么”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处理这些情况吧!希望大家仔细阅读,能够学有所成!
一般来说Map/Reduce都是输出一组文件,但是有些情况下需要我们输出多组文件,比如我上面提到的需求,接下来我用新旧API分别说明如何实现多文件输出
旧API:
MultipleTextOutputFormat 这个类很重要,我们其实只要写个类继承MultipleTextOutputFormat,并且重写generateFileNameForKeyValue(Object key, Object value, String name)方法就好了。因为MultipleTextOutputFormat中有个write方法,即将记录写到hdfs上,在这个方法中,会调用generateFileNameForKeyValue。废话不多说,上代码:
public class MultiFileOutputFormat extends MultipleTextOutputFormat<Object, Object>{ @Override protected String generateFileNameForKeyValue(Object key, Object value, String name) { if(key instanceof OutputFileName){ return ((OutputFileName) key).getPath()+"/"+name; }else{ return super.generateFileNameForKeyValue(key, value, name); } } }
其中OutputFileName是我自己定义的枚举类,便于管理而已,这里也可以return一个路径,以下是OutputFileName的代码
public enum OutputFileName { ERRORLOG("errorlog","logtype=errorlog"), APIREQUEST("apiRequest","logtype=apiRequest"), FIRSTINTOTIME("firstIntoTime","logtype=firstIntoTime"), TABFLUSHTIME("tabFlushTime","logtype=tabFlushTime"), PERFORMANCE("performance","logtype=performance"), FILEREQUEST("fileRequest","logtype=fileRequest"); private String name; private String path; private String tempPath; private OutputFileName(String name,String path){ this.name = name; this.path = path; } public String getName(){ return this.name; } public String getPath(){ if(!StringUtil.isEmpty(tempPath)){ String temp = this.tempPath; this.tempPath = null; return temp; }else{ return this.path; } } }
如何使用MultiFileOutputFormat这个自己写的类呢?就这么用
//job所在类的main方法中 JobConf conf = new JobConf(config,XXX.class); conf.setOutputFormat(MultiFileOutputFormat.class); //map函数中 collector.collect(OutputFileName.ERRORLOG, new Text(log));
此示例做了以上 的操作就可以将数据写到logtype=errorlog目录下了,当然可以根据不同的日志去设置输出目录了
新API:
对于新的API,我没发现MultipleTextOutputFormat这个类,很头疼,我甚至看了源码,仿照旧API自己写了MultipleTextOutputFormat,这就需要做很多事情,必须写个集成RecordWriter的类,重写里面的方法,当时确实可以做到将数据写到不同的路径下,但是也有bug,数据很多的时候,路径下的数据只有一部分保留,做了一下测试,确实把所有的记录都写了,但却只是把最后写的一部分保留在设定好的路径下了,至今都没发现原因,这里就不给代码了,只能保留60多万行的记录
当然我还是有办法的,经过百般折磨,终于在网上找到相关资料,使用这个类MultipleOutputs,查查API,还真有,只不过是在org.apache.hadoop.mapreduce.lib.output包下,这个类相当于把旧的API东西又重新整理了一遍,我们不用再去写其他的类集成MultipleTextOutputFormat。具体使用方法看代码吧
public static class MapperClass extends Mapper<Object, Text, Text, NullWritable> { private Text outkey = new Text(""); private MultipleOutputs<Text, NullWritable> mos; public void map(Object key, Text value, Context context) throws IOException,InterruptedExceptio{ String log = value.toString(); outkey.set(log); int begin = log.indexOf("@[#("); if(begin != -1){ String logForSplit = log.substring(begin+"@".length()); String [] split = logForSplit.split("#"); if(split != null && split.length >0){ String cType = split[0]; if(!StringUtil.isEmpty(cType)){ if("apiRequest".equals(cType)){ mos.write("apiRequest", outkey, NullWritable.get()); }else if("errlog".equals(cType)){ mos.write("errorlog", outkey, NullWritable.get()); } } } } } @Override protected void cleanup(Context context) throws IOException, InterruptedException { mos.close(); super.cleanup(context); } @Override protected void setup(Context context) throws IOException, InterruptedException { mos = new MultipleOutputs<Text, NullWritable>(context); super.setup(context); } }
public class TestJob { public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException { Configuration conf = new Configuration(); Job job = new Job(conf, "ss"); job.setInputFormatClass(TrackInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); job.setJarByClass(TestJob.class); job.setMapperClass(TestJob.MapperClass.class); job.setNumReduceTasks(0); job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class); if(inputPaths.length > 0){ Path[] paths = new Path[inputPaths.length]; for(int i = 0 ; i < inputPaths.length ; i++){ paths[i] = new Path(inputPaths[i]); } FileInputFormat.setInputPaths(job, paths); }else{ FileInputFormat.setInputPaths(job, new Path(args[0])); } FileOutputFormat.setOutputPath(job, new Path(args[1])); MultipleOutputs.addNamedOutput(job, "errorlog", TextOutputFormat.class, Text.class, NullWritable.class); MultipleOutputs.addNamedOutput(job, "apiRequest", TextOutputFormat.class, Text.class, NullWritable.class); } }
OK,这就可以了,总结一下需要注意的问题,首先在我们的map类中一定要定义MultipleOutputs的对象,并且重写cleanup和setup方法,分别用来关闭和创建MultipleOutputs对象,最重要的是在job所在的类中注册我们的文件名,比如errorlog,apiRequest等
上述的两个例子有点区别,第一个是将数据写到不同的目录下,而第二个是写到同一个目录下,但是会分成不同类型的文件,如我截取的记录
-rw-r--r-- 2 hadoop supergroup 10569073 2014-06-06 11:50 /test/aa/fileRequest-m-00063.lzo
-rw-r--r-- 2 hadoop supergroup 10512656 2014-06-06 11:50 /test/aa/fileRequest-m-00064.lzo
-rw-r--r-- 2 hadoop supergroup 68780 2014-06-06 11:51 /test/aa/firstIntoTime-m-00000.lzo
-rw-r--r-- 2 hadoop supergroup 67901 2014-06-06 11:51 /test/aa/firstIntoTime-m-00001.lzo
至于怎么样输出到不同的目录下,有待研究,这种方式有个不好的地方, 会产生很多的
-rw-r--r-- 2 hadoop supergroup 42 2014-06-06 11:50 /test/aa/part-m-00035.lzo 空文件
“hadoop多文件输出新旧API的方法是什么”的内容就介绍到这里了,感谢大家的阅读。如果想了解更多行业相关的知识可以关注亿速云网站,小编将为大家输出更多高质量的实用文章!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。