5、自定义FileOutputFormat分类输出

#应用场景：
我们的map和reduce的输出路径默认就是FileOutPutFormat.setOutPutPath()；指定的那个路径，
但是有的时候我们的代码需要对结果进行分类输出，比如错误的信息输出到一个文件中，而正确的输出
到另一个文件中。这样就需要自己去改写一个FileOutPutFormat类来分类指定。

#实现：
public class EnhanceOutputFormat extends FileOutputFormat<Text, NullWritable>{
@Override
public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
Path tocrawlPath = new Path("d:/flow/crawlout/tocrawl.log");
Path enhancedPath = new Path("d:/flow/enhanced/enhanced.log");
FileSystem fs = FileSystem.get(context.getConfiguration());
FSDataOutputStream tocrawlOs = fs.create(tocrawlPath);
FSDataOutputStream enhancedOs = fs.create(enhancedPath);

return new EnhanceRecordWriter(tocrawlOs,enhancedOs);
}

static class EnhanceRecordWriter extends RecordWriter<Text, NullWritable>{

FSDataOutputStream tocrawlOs = null;
FSDataOutputStream enhancedOs = null;
public EnhanceRecordWriter(FSDataOutputStream tocrawlOs, FSDataOutputStream enhancedOs) {
this.tocrawlOs = tocrawlOs;
this.enhancedOs = enhancedOs;
}

/**
* write方法是把mr程序输出的最后的结果kv写入外部存储系统的实现方法
*/
@Override
public void write(Text key, NullWritable value) throws IOException, InterruptedException {
if(key.toString().contains("tocrawl")){
tocrawlOs.write(key.toString().getBytes());
}else{
enhancedOs.write(key.toString().getBytes());
}
}
@Override
public void close(TaskAttemptContext context) throws IOException, InterruptedException {
if(tocrawlOs!=null) tocrawlOs.close();
if(enhancedOs!=null) enhancedOs.close();
}
}
}

#应用：



##第二句还是需要的？因为fileinputformat的最终结果还有一个secusses的文件，需要默认指定到哪里。