import java.io.IOException;import java.util.ArrayList;import java.util.List;import org.apache.hadoop.fs.BlockLocation;import org.apache.hadoop.fs.FileStatus;import org.apache.hadoop.fs.FileSystem;import org.apache.hadoop.fs.Path;import org.apache.hadoop.fs.PathFilter;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.InputFormat;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 org.apache.hadoop.mapreduce.lib.input.InvalidInputException;import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;//import org.apache.hadoop.mapreduce.lib.input.FileInputFormat.MultiPathFilter;import org.apache.hadoop.mapreduce.security.TokenCache;import com.google.common.base.Charsets;public class MyFileinput extends FileInputFormat<LongWritable, Text> {	private static final PathFilter hiddenFileFilter = new PathFilter() {		public boolean accept(Path p) {			String name = p.getName();			return ((!(name.startsWith("_"))) && (!(name.startsWith("."))));		}	};		// 遍历文件列表， 过滤掉_ .开头的文件（可以自定义过滤）	protected List<FileStatus> listStatus(JobContext job) throws IOException {		System.out.println("*********************");		List result = new ArrayList();		Path[] dirs = getInputPaths(job);		System.out.println("dirs" + dirs);		System.out.println("dirs length = " + dirs.length);		for(Path p: dirs){			System.out.println("Path loop " + p);		}		if (dirs.length == 0) {			throw new IOException("No input paths specified in job");		}		TokenCache.obtainTokensForNamenodes(job.getCredentials(), dirs,				job.getConfiguration());		List errors = new ArrayList();		List filters = new ArrayList();		filters.add(hiddenFileFilter);		PathFilter jobFilter = getInputPathFilter(job);		if (jobFilter != null) {			filters.add(jobFilter);		}		// 过滤函数，可以拓展		PathFilter inputFilter = new MultiPathFilter(filters);		for (int i = 0; i < dirs.length; ++i) {			Path p = dirs[i];			FileSystem fs = p.getFileSystem(job.getConfiguration());			FileStatus[] matches = fs.globStatus(p, inputFilter);			System.out.println("matches=" + matches);			for(FileStatus match: matches){				System.out.println("loop matches" + match.getPath());			}						if (matches == null)				errors.add(new IOException("Input path does not exist: " + p));			else if (matches.length == 0)				errors.add(new IOException("Input Pattern " + p						+ " matches 0 files"));			else {				for (FileStatus globStat : matches) {					System.out.println("globStat " + globStat);					if (globStat.isDirectory())						for (FileStatus stat : fs.listStatus(								globStat.getPath(), inputFilter)) {							result.add(stat);						}					else {						result.add(globStat);					}				}			}		}		if (!(errors.isEmpty())) {			throw new InvalidInputException(errors);		}		// LOG.info("Total input paths to process : " + result.size());		return result;	}		// 计算分片大小，返回一个分片列表	public List<InputSplit> getSplits(JobContext job) throws IOException {		long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));		long maxSize = getMaxSplitSize(job);				System.out.print("minSize " + minSize);		System.out.print("maxSize " + maxSize);				List splits = new ArrayList();		// 获取输入目录下的文件列表(过滤文件)		List<FileStatus> files = listStatus(job);		for (FileStatus file : files) {			Path path = file.getPath();			long length = file.getLen();			System.out.println("path: " + path+ " file len = " + length);			if (length != 0L) {				// 通过路径找到块列表 				FileSystem fs = path.getFileSystem(job.getConfiguration());				BlockLocation[] blkLocations = fs.getFileBlockLocations(file,						0L, length);								if (isSplitable(job, path)) {					long blockSize = file.getBlockSize();					System.out.println("blockSize:" + blockSize);					long splitSize = computeSplitSize(blockSize, minSize,							maxSize);					System.out.println("splitSize :" + splitSize);					long bytesRemaining = length;					System.out.println("bytesRemaining :" + bytesRemaining);										System.out.println(bytesRemaining / splitSize);					// 定义为1.1D， 为避免一个分片过小， 也需要启动一个MAP来运行					// 最后剩余的文件大小只要不超过分片大小的1.1倍都会放入一个分片					while (bytesRemaining / splitSize > 1.1D) {						int blkIndex = getBlockIndex(blkLocations, length								- bytesRemaining);						System.out.println("blkIndex :" + blkIndex);												// 添加到分片分片列表						splits.add(makeSplit(path, length - bytesRemaining,								splitSize, blkLocations[blkIndex].getHosts()));						bytesRemaining -= splitSize;					}										// 文件尾 					if (bytesRemaining != 0L) {						Long remain = length - bytesRemaining;						System.out.println("文件尾大小" + bytesRemaining);						int blkIndex = getBlockIndex(blkLocations, length								- bytesRemaining);						splits.add(makeSplit(path, length - bytesRemaining,								bytesRemaining,								blkLocations[blkIndex].getHosts()));					}				} else {					splits.add(makeSplit(path, 0L, length,							blkLocations[0].getHosts()));				}			} else {				// 测试文件大小为0， 也会启动一个map				splits.add(makeSplit(path, 0L, length, new String[0]));			}		}					job.getConfiguration().setLong(				"mapreduce.input.fileinputformat.numinputfiles", files.size());		// LOG.debug("Total # of splits: " + splits.size());		return splits;	}	private static class MultiPathFilter implements PathFilter {		private List<PathFilter> filters;		public MultiPathFilter(List<PathFilter> filters) {			this.filters = filters;		}		public boolean accept(Path path) {			for (PathFilter filter : this.filters) {				if (!(filter.accept(path))) {					return false;				}			}			return true;		}	}		// 文件内容读取， 默认按行读取 	@Override	public RecordReader<LongWritable, Text> createRecordReader(			InputSplit split, TaskAttemptContext context) {		String delimiter = context.getConfiguration().get(				"textinputformat.record.delimiter");				System.out.println("delimiter ==" + delimiter);		// 默认为空		byte[] recordDelimiterBytes = null;		if (null != delimiter)			recordDelimiterBytes = delimiter.getBytes(Charsets.UTF_8);				return new LineRecordReader(recordDelimiterBytes);	}}