MapReduce在处理小文件时效率很低，但面对大量的小文件又不可避免，这个时候就需要相应的解决方案。
默认的输入格式为TextInputFormat，对于小文件，它是按照它的父类FileInputFormat的切片机制来切片的，也就是不管一个文件多小，独占一片！对于之前的wordcount案例来说，输入目录下一共有4个文件，这将开启4个reduceTask去执行！！！
我们可以自定义InputFormat来实现小文件的合并：
需求
我们需要将三个文本文件合并为一个序列化文件
输入
三个文本文件
输出
一个二进制序列化文件
1.自定义Inputformat类
需要实现两个方法
- isSplitable()：是否可以切片，我们修改返回值为false不可切割。
- createRecordReader：返回我们自定义的RecordReader对象。
public class MyInputFormat extends FileInputFormat<Text, BytesWritable> { //设置文件不可切片,使得一个文件最多作为1片 @Override protected boolean isSplitable(JobContext context, Path filename) { return false; } //设置读取文件的格式为自定义格式 @Override public RecordReader<Text,BytesWritable> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException { return new MyRecordReader(); } }
2. 自定义RecordReader类
我们主要修改两个地方：key和value。
因为我们Mapper类中的map方法需要有四个参数，其中的KEY_IN和VALUE_IN都是由我们的RecordReader类来设置的，这里我们需要设置一下。
- 默认的RecordReader类的key为LongWritable类型，也就是一行数据对应的字节偏移量，这里我们设置key为我们的文件名，也就是Text类型。
- 默认的RecordReader类的value为Text类型，也就是一行文本，这里我们设置value为文件名key对应的文件的二进制序列，也就是BytesWritable类型。
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 MyRecordReader extends RecordReader<Text,BytesWritable> { private Text key; private BytesWritable value; private String filename; private int length; private FileSystem fs; private Path path; private FSDataInputStream is; private boolean flag=true; @Override public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException { FileSplit fileSplit = (FileSplit) inputSplit; filename = fileSplit.getPath().getName(); length = (int) fileSplit.getLength(); path = fileSplit.getPath(); //获取当前Job的配置对象 Configuration conf = taskAttemptContext.getConfiguration(); //获取当前Job使用的文件系统 fs = FileSystem.get(conf); is = fs.open(path); } //文件的名称做为 key - 文件的内容分装为BytesWritable类型的 value, 返回true @Override public boolean nextKeyValue() throws IOException, InterruptedException { //第一次调用nextKeyValue方法 if (flag){ //实例化对象 if (key==null){ key = new Text(); } if (value==null){ value = new BytesWritable(); } //赋值 //将文件名封装到key中 key.set(filename); //将文件的内容读取封装到value中 byte[] content = new byte[ length]; IOUtils.readFully(is,content,0,length); value.set(content,0,length); flag = false; return true; } //第二次调用直接执行 return false return false; } //返回当前读取到的key @Override public Text getCurrentKey() throws IOException, InterruptedException { return key; } //返回当前读取到的value @Override public BytesWritable getCurrentValue() throws IOException, InterruptedException { return value; } //返回读取切片的进度 @Override public float getProgress() throws IOException, InterruptedException { return 0; } //关闭资源 @Override public void close() throws IOException { if (is != null){ IOUtils.closeStream(is); } if (fs != null){ fs.close(); } } }
3. Mapper类
在我们的自定义RecordReader类中，我们已经设置输入的key为文件名，value设置为文件的二进制序列，所以这里直接将key和value写出即可，key的类型为Text，value的类型为BytesWritable。
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); } }
4.Reducer类
public class SequenceFileReducer extends Reducer<Text, BytesWritable,Text,Text> { private Text OUT_VALUE = new Text(); @Override protected void reduce(Text key, Iterable<BytesWritable> values, Context context) throws IOException, InterruptedException { String value = values.toString(); OUT_VALUE.set(value); context.write(key,OUT_VALUE); } }
5.启动类
import com.lyh.mapreduce.MaxTemp.MaxTempMapper; import com.lyh.mapreduce.MaxTemp.MaxTempReducer; import com.lyh.mapreduce.MaxTemp.MaxTempRunner; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.IntWritable; 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.util.Tool; import org.apache.hadoop.util.ToolRunner; public class MySequenceFileRunner extends Configured implements Tool { public static void main(String[] args) throws Exception { ToolRunner.run(new Configuration(),new MySequenceFileRunner(),args); } @Override public int run(String[] args) throws Exception { //1.获取job Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "my sequence file demo"); //2.配置jar包路径 job.setJarByClass(MySequenceFileRunner.class); //3.关联mapper和reducer job.setMapperClass(SequenceFileMapper.class); job.setReducerClass(SequenceFileReducer.class); //4.设置map、reduce输出的k、v类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(BytesWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(BytesWritable.class); //设置切片机制为我们自定义的切片机制 job.setInputFormatClass(MyInputFormat.class); //5.设置统计文件输入的路径,将命令行的第一个参数作为输入文件的路径 FileInputFormat.setInputPaths(job,new Path("D:\\MapReduce_Data_Test\\myinputformat\\input")); //6.设置结果数据存放路径,将命令行的第二个参数作为数据的输出路径 FileOutputFormat.setOutputPath(job,new Path("D:\\MapReduce_Data_Test\\myinputformat\\output1")); return job.waitForCompletion(true) ? 0 : 1;//verbose：是否监控并打印job的信息 } }
执行结果
