为了实现控制最终文件的输出路径和输出格式，可以自定义OutputFormat。
    例如：要在一个MapReduce程序中根据数据的不同输出两类结果到不同目录，这类灵活的输出需求可以通过自定义类来实现。
    步骤：

    1. 自定义一个类继承FileOutputFormat
    2. 改写RecordWriter,具体改写输出数据的方法write()
    3. 在主程序中设置输出类为自定义类

    案例：
    （需求）过滤输入的log日志，包含atguigu的网址输出到/log/atguigu.log，不包含atguigu的网站输出到/log/other.log
    (1) 数据
    log.txt

    1. package com.BigData.MapReduce;
    2. import org.apache.hadoop.conf.Configuration;
    3. import org.apache.hadoop.fs.FSDataOutputStream;
    4. import org.apache.hadoop.fs.FileSystem;
    5. import org.apache.hadoop.fs.Path;
    6. import org.apache.hadoop.io.*;
    7. import org.apache.hadoop.mapreduce.Job;
    8. import org.apache.hadoop.mapreduce.Mapper;
    9. import org.apache.hadoop.mapreduce.RecordWriter;
    10. import org.apache.hadoop.mapreduce.TaskAttemptContext;
    11. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    12. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    13. import java.io.IOException;
    14. import java.net.URI;
    15. import java.net.URISyntaxException;
    18. public class MapReduceDemo {
    19.     public static class MyMapper extends Mapper<LongWritable, Text,  Text, NullWritable> {
    20.         protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, Text,NullWritable>.Context context) throws java.io.IOException ,InterruptedException
    21.         {
    22.             String line = value.toString();
    23.             context.write(new Text(line+"\n"),NullWritable.get());
    24.         }
    25.     }
    26.     //    =======分割线=========
    28.     public static void main(String[] args) throws Exception{
    29.         //设置配置参数
    30.         Configuration conf = new Configuration();
    31.         FileSystem fs = FileSystem.get(new URI("hdfs://192.168.142.20:9000"),conf);
    32.         if(fs.exists(new Path("/out")))
    33.             fs.delete(new Path("/out"),true);
    34.         //创建任务
    35.         conf.set("fs.defaultFS","hdfs://192.168.142.20:9000");
    36.         Path input = new Path("/data/log.data");
    37.         Path output = new Path("/out");
    38.         Job job = Job.getInstance(conf, MapReduceDemo.class.getSimpleName());
    39.         //指定jar文件
    40.         job.setJarByClass(MapReduceDemo.class);
    41.         //指定输入路径，数据在hdfs上的输入路径,指定第一个参数是hdfs输入路径
    42.         FileInputFormat.addInputPath(job,input);
    43.         //指定map的类
    44.         job.setMapperClass(MyMapper.class);
    45.         //指定map输出的key和value的数据类型。
    46.         job.setMapOutputKeyClass(OrderBean.class);
    47.         job.setMapOutputValueClass(Text.class);
    48.         //指定reduce类以及输出数据类型。
    49.         job.setNumReduceTasks(0);
    50.         //指定输出路径hdfs
    51.         job.setOutputFormatClass(MyFileOutPutFormat.class);
    52.         FileOutputFormat.setOutputPath(job,output);
    53.         //提交任务，如果是true，会返回任务执行的进度信息等。
    54.         job.waitForCompletion(true);
    55.     }
    56. }
    58. class MyFileOutPutFormat extends FileOutputFormat<Text,NullWritable>{
    60.     @Override
    61.     public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
    62.         try {
    63.             return new MyRecordWriter(taskAttemptContext);
    64.         } catch (URISyntaxException e) {
    65.             e.printStackTrace();
    66.         }
    67.         return null;
    68.     }
    69. }
    70. class MyRecordWriter extends RecordWriter<Text, NullWritable>{
    71.     private Path output1;
    72.     private Path output2;
    73.     private FSDataOutputStream fos1;
    74.     private FSDataOutputStream fos2;
    75.     private FileSystem fs;
    77.     public MyRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException, URISyntaxException {
    78.         output1 = new Path("/log/atguigu.log");
    79.         output2 = new Path("/log/other.log");
    80.         Configuration conf = new Configuration();
    81.         FileSystem fs = FileSystem.get(new URI("hdfs://192.168.142.20:9000"),conf);
    82.         fos1 = fs.create(output1);
    83.         fos2 = fs.create(output2);
    84.     }
    85.     @Override
    86.     public void write(Text key, NullWritable value) throws IOException, InterruptedException {
    87.         String data = key.toString();
    88.         if(data.contains("atguigu"))
    89.             fos1.writeUTF(data);
    90.         else
    91.             fos2.writeUTF(data);
    92.     }
    94.     @Override
    95.     public void close(TaskAttemptContext context) throws IOException, InterruptedException {
    96.         if(fos1!=null)
    97.             fos1.close();
    98.         if(fos2!=null)
    99.             fos2.close();
    100.         if(fs!=null)
    101.             fs.close();
    102.     }
    103. }

    （2）运行结果
    image.png