hadoop如何通过CombineFileInputFormat实现小文件合并减少map的个数

//map读入的键
package hgs.combinefileinputformat.test;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
public class CombineFileKey implements  WritableComparable<CombineFileKey> {
	private String fileName;
	private long offset;
	
	
	public String getFileName() {
		return fileName;
	}
	public void setFileName(String fileName) {
		this.fileName = fileName;
	}
	public long getOffset() {
		return offset;
	}
	public void setOffset(long offset) {
		this.offset = offset;
	}
	@Override
	public void readFields(DataInput input) throws IOException {
		this.fileName = Text.readString(input);
		this.offset = input.readLong();
		
	}
	@Override
	public void write(DataOutput output) throws IOException {
		Text.writeString(output, fileName);
		output.writeLong(offset);
		
	}
	@Override
	public int compareTo(CombineFileKey obj) {
		int f = this.fileName.compareTo(obj.fileName);
		if(f==0)
			return (int)Math.signum((double)(this.offset-obj.offset));
		return f;
	}
	@Override
	public int hashCode() {
		//摘自于 http://www.idryman.org/blog/2013/09/22/process-small-files-on-hadoop-using-combinefileinputformat-1/
		final int prime = 31;
	    int result = 1;
	    result = prime * result + ((fileName == null) ? 0 : fileName.hashCode());
	    result = prime * result + (int) (offset ^ (offset >>> 32));
	    return result;
	}
	
	@Override
	public boolean equals(Object o) {
		if(o instanceof CombineFileKey)
			return this.compareTo((CombineFileKey)o)==0;
		return false;
	}
}

package hgs.combinefileinputformat.test;
import java.io.IOException;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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.CombineFileSplit;
import org.apache.hadoop.util.LineReader;
public class CombineFileReader extends RecordReader<CombineFileKey, Text>{
	private long startOffset; //offset of the chunk;
	private long end; //end of the chunk;
	private long position; // current pos
	private FileSystem fs;
	private Path path; 
	private CombineFileKey key;
	private Text value;
	private FSDataInputStream input;
	private LineReader reader;
	public CombineFileReader(CombineFileSplit split,TaskAttemptContext context ,
			Integer index) throws IOException {
		//初始化path fs startOffset end
		this.path = split.getPath(index);
		this.fs = this.path.getFileSystem(context.getConfiguration());
		this.startOffset = split.getOffset(index);
		this.end = split.getLength()+this.startOffset;
		//判断现在开始的位置是否在一行的内部
		boolean skipFirstLine = false;
		
		//open the file
		this.input = fs.open(this.path);
		//不等于0说明读取位置在一行的内部
		if(this.startOffset !=0 ){
			skipFirstLine = true;
			--(this.startOffset);
			//定位到开始读取的位置
			this.input.seek(this.startOffset);
		}
		//初始化reader
		this.reader = new LineReader(input);
		if(skipFirstLine){ // skip first line and re-establish "startOffset".
			//这里着这样做的原因是 一行可能包含了这个文件的所有的数据，猜测如果遇到一行的话，还是会读取一行
			//将其实位置调整到一行的开始，这样的话会舍弃部分数据
			this.startOffset += this.reader.readLine(new Text(), 0, (int)Math.min
					             ((long)Integer.MAX_VALUE, this.end - this.startOffset));
		}
		this.position = this.startOffset;
	}
	
	@Override
	public void close() throws IOException {}
	@Override
	public void initialize(InputSplit splite, TaskAttemptContext context) throws IOException, InterruptedException {}
	//返回当前的key
	@Override
	public CombineFileKey getCurrentKey() throws IOException, InterruptedException {
		return key;
	}
	//返回当前的value
	@Override
	public Text getCurrentValue() throws IOException, InterruptedException {
		return value;
	}
	//执行的进度
	@Override
	public float getProgress() throws IOException, InterruptedException {
		//返回的类型为float
		if(this.startOffset==this.end){
			return 0.0f;
		}else{
			return Math.min(1.0f, (this.position - this.startOffset)/(float)(this.end - this.startOffset));
		}
	}
	//该方法判断是否有下一个key value
	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		//对key和value初始化
		if(this.key == null){
			this.key = new CombineFileKey();
			this.key.setFileName(this.path.getName());
		}
		this.key.setOffset(this.position);
		if(this.value == null){
			this.value = new Text();
		}
		//读取一行数据，如果读取的newSieze=0说明split的数据已经处理完成
		int newSize = 0;
		if(this.position<this.end){
			newSize = reader.readLine(this.value);
			position += newSize;
		}
		//没有数据，将key value置位空
		if(newSize == 0){
			this.key = null;
			this.value = null;
			return false;
		}else{
			return true;
		}
	}
	
}

package hgs.combinefileinputformat.test;
import java.io.IOException;
import org.apache.hadoop.fs.Path;
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.CombineFileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.CombineFileRecordReader;
import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;
public class CustCombineInputFormat extends CombineFileInputFormat<CombineFileKey, Text> {
	public CustCombineInputFormat(){
		super();
		//最大切片大小
		this.setMaxSplitSize(67108864);//64 MB
	}
	@Override
	public RecordReader<CombineFileKey, Text> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException {		
		return new CombineFileRecordReader<CombineFileKey, Text>((CombineFileSplit)split,context,CombineFileReader.class);
	}
	@Override
	protected boolean isSplitable(JobContext context, Path file) {
		return false;
	}
}
//驱动类
package hgs.test;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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 hgs.combinefileinputformat.test.CustCombineInputFormat;
public class LetterCountDriver {
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		//conf.set("mapreduce.map.log.level", "INFO");
		///conf.set("mapreduce.reduce.log.level", "INFO");
		Job job = Job.getInstance(conf, "LetterCount");
		job.setJarByClass(hgs.test.LetterCountDriver.class);
		// TODO: specify a mapper
		job.setMapperClass(LetterCountMapper.class);
		// TODO: specify a reducer
		job.setReducerClass(LetterReducer.class);
		// TODO: specify output types
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		if(args[0].equals("1"))
			job.setInputFormatClass(CustCombineInputFormat.class);
		else{}
		// TODO: specify input and output DIRECTORIES (not files)
		FileInputFormat.setInputPaths(job, new Path("/words"));
		FileOutputFormat.setOutputPath(job, new Path("/result"));
		if (!job.waitForCompletion(true))
			return;
	}
}