Hadoop-MapReduce-自定义OutputFormat案例实操-连载中

系统管理员 2023-03-01 11:49 48阅读 0赞

## OutputFormat数据输出 ##

### 文章目录 ###

*   *  OutputFormat数据输出
     *   *  1 OutputFormat接口实现类
         *  2 自定义OutputFormat
         *  3 自定义OutputFormat案例实操

### 1 OutputFormat接口实现类 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70]

### 2 自定义OutputFormat ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 1]

### 3 自定义OutputFormat案例实操 ###

**1）需求**

过滤输入的log日志，包含qinjl的网站输出到e:/qinjl.log，不包含qinjl的网站输出到e:/other.log。

**2）需求分析**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 2]  
**3）案例实操**

（1）编写FilterMapper类

public class LogMapper extends Mapper<LongWritable, Text,Text, NullWritable> {
        
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        
            //不做任何处理,直接写出一行log数据
            context.write(value,NullWritable.get());
        }
    }

（2）编写FilterReducer类

public class LogReducer extends Reducer<Text, NullWritable,Text, NullWritable> {
        
        @Override
        protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        
            //防止有相同的数据,迭代写出
            for (NullWritable value : values) {
        
                context.write(key,NullWritable.get());
            }
        }
    }

（3）自定义一个OutputFormat类

public class LogOutputFormat extends FileOutputFormat<Text, NullWritable> {
        
        @Override
        public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
        
            //创建一个自定义的RecordWriter返回
            LogRecordWriter logRecordWriter = new LogRecordWriter(job);
            return logRecordWriter;
        }
    }

（4）编写RecordWriter类

public class LogRecordWriter extends RecordWriter<Text, NullWritable> {
        
    
        private FSDataOutputStream qinjlOut;
        private FSDataOutputStream otherOut;
    
        public LogRecordWriter(TaskAttemptContext job) {
        
            try {
        
                //获取文件系统对象
                FileSystem fs = FileSystem.get(job.getConfiguration());
                //用文件系统对象创建两个输出流对应不同的目录
                qinjlOut = fs.create(new Path("d:/hadoop/qinjl.txt"));
                otherOut = fs.create(new Path("d:/hadoop/other.txt"));
            } catch (IOException e) {
        
                e.printStackTrace();
            }
        }
    
        @Override
        public void write(Text key, NullWritable value) throws IOException, InterruptedException {
        
            String log = key.toString();
            //根据一行的log数据是否包含qinjl,判断两条输出流输出的内容
            if (log.contains("qinjl")) {
        
                qinjlOut.writeBytes(log + "\n");
            } else {
        
                otherOut.writeBytes(log + "\n");
            }
        }
    
        @Override
        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        
            //关流
            IOUtils.closeStream(qinjlOut);
            IOUtils.closeStream(otherOut);
        }
    }

（5）编写FilterDriver类

public class LogDriver {
        
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        
    
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
    
            job.setJarByClass(LogDriver.class);
            job.setMapperClass(LogMapper.class);
            job.setReducerClass(LogReducer.class);
    
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(NullWritable.class);
    
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(NullWritable.class);
    
            //设置自定义的outputformat
            job.setOutputFormatClass(LogOutputFormat.class);
    
            FileInputFormat.setInputPaths(job, new Path("D:\\input"));
            //虽然我们自定义了outputformat，但是因为我们的outputformat继承自fileoutputformat
            //而fileoutputformat要输出一个_SUCCESS文件，所以在这还得指定一个输出目录
            FileOutputFormat.setOutputPath(job, new Path("D:\\logoutput"));
    
            boolean b = job.waitForCompletion(true);
            System.exit(b ? 0 : 1);
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70]: /images/20230208/88f81557c28547b5a5f3997ec5c6da74.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 1]: /images/20230208/b9a517089d224fa18fe993c68c8a761d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 2]: /images/20230208/22d7dc3d13d94c65b957145feeca4be9.png