devbox
空白诗007
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

自定义InputFormat、自定义OutputFormat_自定义inputformat实现

作者：空白诗007 | 2024-07-01 22:20:40

赞

踩

自定义inputformat实现

1.自定义inputFormat合并小文件

需求

无论hdfs还是mapreduce，对于小文件都有损效率，实践中，又难免面临处理大量小文件的场景，此时，就需要有相应解决方案, 将多个小文件合并成一个文件 SequenceFile.SequenceFile 里面存储着多个文件。存储的形式为文件名称为 key,文件内容为 value。

分析

小文件的优化无非以下几种方式：

  1. 在数据采集的时候，就将小文件或小批数据合成大文件再上传HDFS
  2. 在业务处理之前，在HDFS上使用mapreduce程序对小文件进行合并
  3. 在mapreduce处理时，可采用combineInputFormat提高效率

实现

本节实现的是上述第二种方式

程序的核心机制：

自定义一个InputFormat

改写RecordReader，实现一次读取一个完整文件封装为KV

在输出时使用SequenceFileOutPutFormat输出合并文件

第一步:自定义InputFormat类

  1. import org.apache.hadoop.fs.Path;
  2. import org.apache.hadoop.io.BytesWritable;
  3. import org.apache.hadoop.io.NullWritable;
  4. import org.apache.hadoop.mapreduce.InputSplit;
  5. import org.apache.hadoop.mapreduce.JobContext;
  6. import org.apache.hadoop.mapreduce.RecordReader;
  7. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  8. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  9.  
  10. import java.io.IOException;
  11.  
  12. public class Custom_FileInputFormat extends FileInputFormat<NullWritable,BytesWritable> {
  13.     @Override
  14.     protected boolean isSplitable(JobContext context, Path filename) {
  15.         return false;
  16.     }
  17.  
  18.     @Override
  19.     public RecordReader<NullWritable, BytesWritable> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
  20.         Custom_RecordReader reader = new Custom_RecordReader();
  21.         reader.initialize(inputSplit,taskAttemptContext);
  22.         return reader;
  23.     }
  24. }

第二步:自定义RecordReader类

  1. import org.apache.commons.io.IOUtils;
  2. import org.apache.hadoop.conf.Configuration;
  3. import org.apache.hadoop.fs.FSDataInputStream;
  4. import org.apache.hadoop.fs.FileSystem;
  5. import org.apache.hadoop.fs.Path;
  6. import org.apache.hadoop.io.BytesWritable;
  7. import org.apache.hadoop.io.NullWritable;
  8. import org.apache.hadoop.mapreduce.InputSplit;
  9. import org.apache.hadoop.mapreduce.RecordReader;
  10. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  11. import org.apache.hadoop.mapreduce.lib.input.FileSplit;
  12.  
  13. import java.io.IOException;
  14.  
  15. public class Custom_RecordReader extends RecordReader<NullWritable,BytesWritable> {
  16.  
  17.     private FileSplit fileSplit;
  18.     private Configuration conf;
  19.     private BytesWritable bytesWritable = new BytesWritable();
  20.     private boolean processed = false;
  21.  
  22.     @Override
  23.     public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
  24.         fileSplit = (FileSplit) inputSplit;
  25.         conf = taskAttemptContext.getConfiguration();
  26.     }
  27.  
  28.     @Override
  29.     public boolean nextKeyValue() throws IOException, InterruptedException {
  30.         if (!processed) {
  31.             Path path = fileSplit.getPath();
  32.             FileSystem fileSystem = null;
  33.             FSDataInputStream inputStream = null;
  34.             try {
  35.                 fileSystem = FileSystem.get(conf);
  36.                 //读取文件
  37.                 inputStream = fileSystem.open(path);
  38.                 //初始化一个字节数组，长度为读取内容的大小
  39.                 byte[] bytes = new byte[(int) fileSplit.getLength()];
  40.                 //读取到数组中
  41.                 IOUtils.readFully(inputStream, bytes, 0, bytes.length);
  42.                 //把字节数组转换成bytesWritable对象
  43.                 bytesWritable.set(bytes, 0, bytes.length);
  44.             } catch (Exception e) {
  45.                 e.printStackTrace();
  46.             } finally {
  47.                 fileSystem.close();
  48.                 inputStream.close();
  49.             }
  50.             processed = true;
  51.             return true;
  52.         }else {
  53.             return false;
  54.         }
  55.     }
  56.  
  57.     @Override
  58.     public NullWritable getCurrentKey() throws IOException, InterruptedException {
  59.         return NullWritable.get();
  60.     }
  61.  
  62.     @Override
  63.     public BytesWritable getCurrentValue() throws IOException, InterruptedException {
  64.         return bytesWritable;
  65.     }
  66.  
  67.     @Override
  68.     public float getProgress() throws IOException, InterruptedException {
  69.         return processed?1.0F:0.0F;
  70.     }
  71.  
  72.     @Override
  73.     public void close() throws IOException {
  74.  
  75.     }
  76. }

第三步:编写Mapper类

  1. import org.apache.hadoop.io.BytesWritable;
  2. import org.apache.hadoop.io.NullWritable;
  3. import org.apache.hadoop.io.Text;
  4. import org.apache.hadoop.mapreduce.Mapper;
  5. import org.apache.hadoop.mapreduce.lib.input.FileSplit;
  6.  
  7. import java.io.IOException;
  8.  
  9. public class Custom_Mapper extends Mapper<NullWritable,BytesWritable,Text,BytesWritable> {
  10.     @Override
  11.     protected void map(NullWritable key, BytesWritable value, Context context) throws IOException, InterruptedException {
  12.         FileSplit fileSplit = (FileSplit) context.getInputSplit();
  13.         String name = fileSplit.getPath().getName();
  14.         context.write(new Text(name),value);
  15.  
  16.  
  17.     }
  18. }

 第四步:编写Driver主类:

  1. import org.apache.hadoop.conf.Configuration;
  2. import org.apache.hadoop.conf.Configured;
  3. import org.apache.hadoop.fs.Path;
  4. import org.apache.hadoop.io.BytesWritable;
  5. import org.apache.hadoop.io.Text;
  6. import org.apache.hadoop.mapreduce.Job;
  7. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  8. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  9. import org.apache.hadoop.util.Tool;
  10. import org.apache.hadoop.util.ToolRunner;
  11. import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
  12.  
  13. public class Custom_Driver extends Configured implements Tool {
  14.     @Override
  15.     public int run(String[] args) throws Exception {
  16.         Job job = Job.getInstance(new Configuration());
  17.  
  18.         job.setInputFormatClass(Custom_FileInputFormat.class);
  19.         TextInputFormat.addInputPath(job,new Path("E:\\自定义InputFormat\\input\\"));
  20.  
  21.         job.setMapperClass(Custom_Mapper.class);
  22.         job.setMapOutputKeyClass(Text.class);
  23.         job.setMapOutputValueClass(BytesWritable.class);
  24.  
  25.         job.setOutputKeyClass(Text.class);
  26.         job.setOutputValueClass(BytesWritable.class);
  27.  
  28.         job.setOutputFormatClass(SequenceFileOutputFormat.class);
  29.         TextOutputFormat.setOutputPath(job,new Path("E:\\自定义InputFormat\\output\\"));
  30.         return job.waitForCompletion(true)?0:1;
  31.     }
  32.  
  33.     public static void main(String[] args) throws Exception {
  34.         ToolRunner.run(new Custom_Driver(),args);
  35.     }
  36. }

 

2.自定义OutputFormat

需求

现在有一些订单的评论数据，需求，将订单的好评与差评进行区分开来，将最终的数据分开到不同的文件夹下面去，数据内容参见资料文件夹，其中数据第九个字段表示好评，中评，差评。0：好评，1：中评，2：差评

分析

程序的关键点是要在一个mapreduce程序中根据数据的不同输出两类结果到不同目录，这类灵活的输出需求可以通过自定义outputformat来实现

实现

实现要点：

  1. 在mapreduce中访问外部资源
  2. 自定义outputformat，改写其中的recordwriter，改写具体输出数据的方法write()

第一步：自定义outputformat类

  1. import org.apache.hadoop.conf.Configuration;
  2. import org.apache.hadoop.fs.FSDataOutputStream;
  3. import org.apache.hadoop.fs.FileSystem;
  4. import org.apache.hadoop.fs.Path;
  5. import org.apache.hadoop.io.NullWritable;
  6. import org.apache.hadoop.io.Text;
  7. import org.apache.hadoop.mapreduce.RecordWriter;
  8. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  9. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  10.  
  11. import java.io.IOException;
  12.  
  13. public class Custom_OutputFormat extends FileOutputFormat<Text,NullWritable> {
  14.  
  15.     @Override
  16.     public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
  17.  
  18.         Configuration conf = taskAttemptContext.getConfiguration();
  19.  
  20.         //获取fileSystem
  21.         FileSystem fileSystem = FileSystem.get(conf);
  22.         FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("M:\\自定义outputformat\\output\\good_cmment\\good_cmment.txt"));
  23.         FSDataOutputStream fsDataOutputStream1 = fileSystem.create(new Path("M:\\自定义outputformat\\output\\bad_cmment\\bad_cmment.txt"));
  24.         return new Custom_RecordWriter(fsDataOutputStream,fsDataOutputStream1); //调用自定义的RecordWriter的满参构造传递两个输出流
  25.     }
  26. }

第二步：自定义RecordWrite类

  1. import org.apache.hadoop.fs.FSDataOutputStream;
  2. import org.apache.hadoop.io.NullWritable;
  3. import org.apache.hadoop.io.Text;
  4. import org.apache.hadoop.mapreduce.RecordWriter;
  5. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  6.  
  7. import java.io.IOException;
  8.  
  9. public class Custom_RecordWriter extends RecordWriter<Text,NullWritable> {
  10.  
  11.     FSDataOutputStream out1;
  12.     FSDataOutputStream out2;
  13.  
  14.     public Custom_RecordWriter() {
  15.     }
  16.  
  17.     public Custom_RecordWriter(FSDataOutputStream out1, FSDataOutputStream out2) {
  18.  
  19.         this.out1 = out1;
  20.         this.out2 = out2;
  21.     }
  22.  
  23.     public FSDataOutputStream getOut1() {
  24.  
  25.         return out1;
  26.     }
  27.  
  28.     public void setOut1(FSDataOutputStream out1) {
  29.         this.out1 = out1;
  30.     }
  31.  
  32.     public FSDataOutputStream getOut2() {
  33.         return out2;
  34.     }
  35.  
  36.     public void setOut2(FSDataOutputStream out2) {
  37.         this.out2 = out2;
  38.     }
  39.  
  40.     @Override
  41.     public void write(Text key, NullWritable value) throws IOException, InterruptedException {
  42.         if (key.toString().split("\t")[9].equals("0")){
  43.             //索引为9的是好评
  44.             out1.write(key.toString().getBytes());
  45.             out1.write("\r\n".getBytes());
  46.         }else{
  47.             //中评和差评
  48.             out2.write(key.toString().getBytes());
  49.             out2.write("\r\n".getBytes());
  50.         }
  51.     }
  52.  
  53.     @Override
  54.     public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
  55.         if (null != out1){
  56.             out1.close();
  57.         }
  58.         if (null != out2){
  59.             out2.close();
  60.         }
  61.     }
  62. }

第三步：编写Mapper类

  1. import org.apache.hadoop.io.LongWritable;
  2. import org.apache.hadoop.io.NullWritable;
  3. import org.apache.hadoop.io.Text;
  4. import org.apache.hadoop.mapreduce.Mapper;
  5.  
  6. import java.io.IOException;
  7.  
  8. public class Custom_Mapper extends Mapper<LongWritable,Text,Text,NullWritable> {
  9.     @Override
  10.     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  11.         context.write(value,NullWritable.get());
  12.     }
  13. }

第四步：编写Driver主类

  1. import org.apache.hadoop.conf.Configured;
  2. import org.apache.hadoop.fs.Path;
  3. import org.apache.hadoop.io.NullWritable;
  4. import org.apache.hadoop.io.Text;
  5. import org.apache.hadoop.mapreduce.Job;
  6. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  7. import org.apache.hadoop.util.Tool;
  8. import org.apache.hadoop.util.ToolRunner;
  9.  
  10. public class Custom_OutputFormatDriver extends Configured implements Tool {
  11.  
  12.     public static void main(String[] args) throws Exception {
  13.         System.exit(ToolRunner.run(new Custom_OutputFormatDriver(),args));
  14.     }
  15.  
  16.     @Override
  17.     public int run(String[] args) throws Exception {
  18.  
  19.         Job job = Job.getInstance(super.getConf());
  20.         //设置mapper加载类
  21.         job.setMapperClass(Custom_Mapper.class);
  22.         //设置map输出类型
  23.         job.setMapOutputKeyClass(Text.class);
  24.         job.setMapOutputValueClass(NullWritable.class);
  25.         //设置InputFormat加载类
  26.         job.setInputFormatClass(TextInputFormat.class);
  27.         //设置自定义OutputFormat加载类
  28.         job.setOutputFormatClass(Custom_OutputFormat.class);
  29.         //设置读取和写出路径
  30.         TextInputFormat.addInputPath(job,new Path("M:\\自定义outputformat\\input\\ordercomment.csv"));
  31.         Custom_OutputFormat.setOutputPath(job,new Path("M:\\自定义outputformat\\output\\tmp"));
  32.         return job.waitForCompletion(true)?0:1;
  33.     }
  34. }

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/空白诗007/article/detail/777839
推荐阅读

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。