maperduce自定义inputformat-白红宇

maperduce自定义inputformat

阅读量：3968 次

发布时间：2019-05-24

本文共 6954 字，大约阅读时间需要 23 分钟。

MapReduce提供的inputformat输入类型并不能满足我们的使用需求，因此，mapreduce可以由用户自定义inputformat逻辑来处理各类数据。

步骤：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-naJgpogf-1603113808497)(https://s1.ax1x.com/2020/10/19/0xiCKH.png)]

自定义InputFormat方案解决处理大量小文件的问题

需求：

将多个小文件合并成一个SequenceFile文件（SequenceFile文件是Hadoop用来存储二进制形式的key-value对的文件格式），

SequenceFile里面存储着多个文件，存储的形式为文件路径+名称为key，文件内容为value。

步骤：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-T9ytTm6V-1603113808511)(https://s1.ax1x.com/2020/10/19/0xkQNq.png)]

代码实例

可以看到在下面的代码实例中我们自定义了一个继承FileInputFormat类的WholeFileInputFormat类用来实现以文件整体作为输入的input format，并在其中重写了RecordReader对象将文件内容转换为ByteWritable来封装到Value中去，最终通过reducer自带的功能将这些小文件全部整合写入到一个文件中去。在Driver驱动类中设置自定义的InputFormat类：

job.setInputFormatClass(WholeFileInputformat.class);

注意这样合并小文件的方式实际上就是每个小文件mapreduce都会开启一个maptask来处理一个小文件。

自定义Inputformat

package com.IN.mat;import java.io.IOException;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.BytesWritable;import org.apache.hadoop.io.NullWritable;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;// 定义类继承FileInputFormatpublic class WholeFileInputformat extends FileInputFormat
   
    {
       @Override    protected boolean isSplitable(JobContext context, Path filename) {
           return false;    }    @Override    public RecordReader
    
      createRecordReader(InputSplit split, TaskAttemptContext context)	throws IOException, InterruptedException {
           WholeRecordReader recordReader = new WholeRecordReader();        recordReader.initialize(split, context);        return recordReader;    }}

重写Reader

package com.IN.mat;import java.io.IOException;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.NullWritable;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;public class WholeRecordReader extends RecordReader
   
    {
       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，只需要将kv写入Context中即可

package com.IN.mat;import java.io.IOException;import org.apache.hadoop.io.BytesWritable;import org.apache.hadoop.io.NullWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.lib.input.FileSplit;public class SequenceFileMapper extends Mapper
   
    {
       @Override    protected void map(Text key, BytesWritable value,Context context)		throws IOException, InterruptedException {
           context.write(key, value);    }}

Reducer，同样只需要将kv写入即可，最后会自动合并为一个文件

package com.IN.mat;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
   
     {
       @Override    protected void reduce(Text key, Iterable
    
      values, Context context)		throws IOException, InterruptedException {
           context.write(key, values.iterator().next());    }}

Driver驱动类

package com.IN.mat;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[] {
    "d:/mapreduceinput/input1", "d:/mapreduceoutput/output1" };        // 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);    }}

转载地址：http://jicki.baihongyu.com/

你可能感兴趣的文章