hadoop案例:自定义inputformat

r囧r小猫 2022-11-21 14:59 261阅读 0赞

### 文章目录 ###

*   *  需求
     *  输入数据
     *  期望输出数据
     *  需求分析
     *  编写代码
     *   *  自定义inputformat
         *  自定义WholeRecordReader
         *  Mapper类
         *  Reducer类
         *  Driver类
         *  执行结果

*  无论HDFS还是MapReduce，在处理小文件时效率都非常低，但又难免面临处理大量小文件的场景，此时，就需要有相应解决方案。可以自定义InputFormat实现小文件的合并

## 需求 ##

将多个小文件合并成一个SequenceFile文件（SequenceFile文件是Hadoop用来存储二进制形式的key-value对的文件格式），SequenceFile里面存储着多个文件，存储的形式为文件路径+名称为key，文件内容为value

## 输入数据 ##

one.txt

yongpeng weidong weinan
    sanfeng luozong xiaoming

two.txt

longlong fanfan
    mazong kailun yuhang yixin
    longlong fanfan
    mazong kailun yuhang yixin

three.txt

shuaige changmo zhenqiang 
    dongli lingu xuanxuan

## 期望输出数据 ##

![在这里插入图片描述][2020110214181551.png_pic_center]

## 需求分析 ##

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

## 编写代码 ##

### 自定义inputformat ###

package com.mr.fileinputformat;
    
    
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.BytesWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.InputSplit;
    import org.apache.hadoop.mapreduce.JobContext;
    import org.apache.hadoop.mapreduce.RecordReader;
    import org.apache.hadoop.mapreduce.TaskAttemptContext;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    
    ;import java.io.IOException;
    
    // 定义类继承FileInputFormat
    public class WholeFileInputformat extends FileInputFormat<Text, BytesWritable> { 
    
        @Override
        protected boolean isSplitable(JobContext context, Path filename) { 
            return false;
        }
    
        @Override
        public RecordReader<Text, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context)	throws IOException, InterruptedException { 
    
            WholeRecordReader recordReader = new WholeRecordReader();
            recordReader.initialize(split, context);
    
            return recordReader;
        }
    }

### 自定义WholeRecordReader ###

package com.mr.fileinputformat;
    
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FSDataInputStream;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.BytesWritable;
    import org.apache.hadoop.io.IOUtils;
    import org.apache.hadoop.io.Text;
    
    import org.apache.hadoop.mapreduce.InputSplit;
    import org.apache.hadoop.mapreduce.RecordReader;
    import org.apache.hadoop.mapreduce.TaskAttemptContext;
    import org.apache.hadoop.mapreduce.lib.input.FileSplit;
    
    import java.io.IOException;
    
    
    public class WholeRecordReader extends RecordReader<Text, BytesWritable> { 
    
        private Configuration configuration;
        private FileSplit split;
    
        private boolean isProgress= true;
        private BytesWritable value = new BytesWritable();
        private Text k = new Text();
    
        @Override
        public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException { 
    
            this.split = (FileSplit)split;
            configuration = context.getConfiguration();
        }
    
        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException { 
    
            if (isProgress) { 
    
                // 1 定义缓存区
                byte[] contents = new byte[(int)split.getLength()];
    
                FileSystem fs = null;
                FSDataInputStream fis = null;
    
                try { 
                    // 2 获取文件系统
                    Path path = split.getPath();
                    fs = path.getFileSystem(configuration);
    
                    // 3 读取数据
                    fis = fs.open(path);
    
                    // 4 读取文件内容
                    IOUtils.readFully(fis, contents, 0, contents.length);
    
                    // 5 输出文件内容
                    value.set(contents, 0, contents.length);
    
    // 6 获取文件路径及名称
                    String name = split.getPath().toString();
    
    // 7 设置输出的key值
                    k.set(name);
    
                } catch (Exception e) { 
    
                }finally { 
                    IOUtils.closeStream(fis);
                }
    
                isProgress = false;
    
                return true;
            }
    
            return false;
        }
    
        @Override
        public Text getCurrentKey() throws IOException, InterruptedException { 
            return k;
        }
    
        @Override
        public BytesWritable getCurrentValue() throws IOException, InterruptedException { 
            return value;
        }
    
        @Override
        public float getProgress() throws IOException, InterruptedException { 
            return 0;
        }
    
        @Override
        public void close() throws IOException { 
        }
    }

### Mapper类 ###

package com.mr.fileinputformat;
    
    import java.io.IOException;
    import org.apache.hadoop.io.BytesWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    public class SequenceFileMapper extends Mapper<Text, BytesWritable, Text, BytesWritable>{ 
    
        @Override
        protected void map(Text key, BytesWritable value,			Context context)		throws IOException, InterruptedException { 
    
            context.write(key, value);
        }
    }

### Reducer类 ###

package com.mr.fileinputformat;
    
    import java.io.IOException;
    import org.apache.hadoop.io.BytesWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    public class SequenceFileReducer extends Reducer<Text, BytesWritable, Text, BytesWritable> { 
    
        @Override
        protected void reduce(Text key, Iterable<BytesWritable> values, Context context)		throws IOException, InterruptedException { 
    
            context.write(key, values.iterator().next());
        }
    }

### Driver类 ###

package com.mr.fileinputformat;
    
    import java.io.IOException;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.BytesWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
    
    public class SequenceFileDriver { 
    
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
    
            // 输入输出路径需要根据自己电脑上实际的输入输出路径设置
            args = new String[] {  "E:/Hadoop/src/main/resources/input/iphone", "E:/Hadoop/src/main/resources/ouput/sequencefile" };
    
            // 1 获取job对象
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
    
            // 2 设置jar包存储位置、关联自定义的mapper和reducer
            job.setJarByClass(SequenceFileDriver.class);
            job.setMapperClass(SequenceFileMapper.class);
            job.setReducerClass(SequenceFileReducer.class);
    
            // 7设置输入的inputFormat
            job.setInputFormatClass(WholeFileInputformat.class);
    
            // 8设置输出的outputFormat
            job.setOutputFormatClass(SequenceFileOutputFormat.class);
    
    // 3 设置map输出端的kv类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(BytesWritable.class);
    
            // 4 设置最终输出端的kv类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(BytesWritable.class);
    
            // 5 设置输入输出路径
            FileInputFormat.setInputPaths(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
            // 6 提交job
            boolean result = job.waitForCompletion(true);
            System.exit(result ? 0 : 1);
        }
    }

### 执行结果 ###

![在这里插入图片描述][2020110214253893.png_pic_center]

[2020110214181551.png_pic_center]: /images/20221120/14f88622d2fc459a918b0807cf0fe8b7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/08d53f9ba68a45558536d015055b36e7.png
[2020110214253893.png_pic_center]: /images/20221120/1b1d668adfff4a399a6366b158ff5d64.png