您的位置:首页 > 产品设计 > UI/UE

输入的InputFormat----SequenceFileInputFormat

2012-04-27 14:07 441 查看
继承关系:SequenceFileInputFormat extends FileInputFormat implements InputFormat 。

SequenceFileInputFormat 代码如下(其实很简单):

/**
* 覆盖了FileInputFormat的这个方法,FileInputFormat通过这个方法得到的FileStatus[]
* 长度就是将要运行的map的长度,每个FileStatus对应一个文件
*/
@Override
protected FileStatus[] listStatus(JobConf job) throws IOException {
FileStatus[] files = super.listStatus(job);
/*调用父类的listStatus方法,然后进行了自己的处理,将得到的FileStatus[]遍历一遍,
遇到文件夹时候,看其是否为MapFile,是的话去除其中也是SequenceFile的data文件,否则将文件夹过滤掉。
*/
for (int i = 0; i < files.length; i++) {
FileStatus file = files[i];
if (file.isDir()) {     // it's a MapFile
Path dataFile = new Path(file.getPath(), MapFile.DATA_FILE_NAME);
FileSystem fs = file.getPath().getFileSystem(job);
// use the data file
files[i] = fs.getFileStatus(dataFile);
}
}
return files;
}


下面看看FileInputFormat的listStatus(JobConf job)方法:

protected FileStatus[] listStatus(JobConf job) throws IOException {
//得到job配置中的所有输入路径,路径中是以 ,号隔开的 。
Path[] dirs = getInputPaths(job);
if (dirs.length == 0) {
throw new IOException("No input paths specified in job");
}

// get tokens for all the required FileSystems..
TokenCache.obtainTokensForNamenodes(job.getCredentials(), dirs, job);

List<FileStatus> result = new ArrayList<FileStatus>();
List<IOException> errors = new ArrayList<IOException>();

// creates a MultiPathFilter with the hiddenFileFilter and the
// user provided one (if any).
//处理一个过路文件的filter,可以将输入文件夹中的一些文件过滤掉
List<PathFilter> filters = new ArrayList<PathFilter>();
filters.add(hiddenFileFilter);
PathFilter jobFilter = getInputPathFilter(job);
if (jobFilter != null) {
filters.add(jobFilter);
}
PathFilter inputFilter = new MultiPathFilter(filters);
//对于每个输入文件夹进行遍历
for (Path p: dirs) {
FileSystem fs = p.getFileSystem(job);
//得到这个输入文件下下的所有文件(夹)
FileStatus[] matches = fs.globStatus(p, inputFilter);
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) {
if (globStat.isDir()) {
//文件夹的话,将该文件夹下的所有文件和文件夹添加到结果中
//****注意此处没有再往下层遍历,而是将文件和文件夹都返回到结果中 。
for(FileStatus stat: fs.listStatus(globStat.getPath(),
inputFilter)) {
result.add(stat);
}
} else {
//文件的话直接添加到result中,其实没有任何判断该文件是否是输入需要的格式等等
result.add(globStat);
}
}
}
}

if (!errors.isEmpty()) {
throw new InvalidInputException(errors);
}
LOG.info("Total input paths to process : " + result.size());
return result.toArray(new FileStatus[result.size()]);
}


是以总结SequenceFileInputFormat中输出文件的规律(假设输入文件夹是/input ):

1、输入文件夹中的文件 ,即满足:/input/***文件 。

2、输入文件夹中子文件夹中的文件 ,/input/***/***文件 。

3、输入文件夹中的子文件夹的子文件夹中的data文件 ,/input/***/***/data文件 ,该主要是针对MapFile的 。

得到一个个文件后,怎么将文件映射到InputSplit(有可能一个file映射1个InputSplit,也可能映射几个InputSplit),代码见下:

/** Splits files returned by {@link #listStatus(JobConf)} when
* they're too big.*/
@SuppressWarnings("deprecation")
public InputSplit[] getSplits(JobConf job, int numSplits)
throws IOException {
FileStatus[] files = listStatus(job);

// Save the number of input files in the job-conf
job.setLong(NUM_INPUT_FILES, files.length);
long totalSize = 0;                           // compute total size
for (FileStatus file: files) {                // check we have valid files
if (file.isDir()) {
throw new IOException("Not a file: "+ file.getPath());
}
totalSize += file.getLen();
}

long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
long minSize = Math.max(job.getLong("mapred.min.split.size", 1),
minSplitSize);

// generate splits
ArrayList<FileSplit> splits = new ArrayList<FileSplit>(numSplits);
NetworkTopology clusterMap = new NetworkTopology();
//对于每个文件该分割成多少InputSplit的处理
for (FileStatus file: files) {
Path path = file.getPath();
FileSystem fs = path.getFileSystem(job);
long length = file.getLen();
BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
//允许切割文件时候,即允许将一个file切割成多个InputSplit 。
if ((length != 0) && isSplitable(fs, path)) {
long blockSize = file.getBlockSize();
//分块的大小,一般是以快为单位,即会选择blockSize ,其实会将按照block来分块,这样比较合适
long splitSize = computeSplitSize(goalSize, minSize, blockSize);

long bytesRemaining = length;
//按照分割设置(每块大小),将文件从从offset=0 到offset=length 分割成 (length/splitSize+1) 个(其实也并不是这些个啦 ~ ~ ,最后一块的大小可以是splitSize*SPLIT_SLOP)
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
String[] splitHosts = getSplitHosts(blkLocations,
length-bytesRemaining, splitSize, clusterMap);
splits.add(new FileSplit(path, length-bytesRemaining, splitSize,
splitHosts));
bytesRemaining -= splitSize;
}
//返回splits
if (bytesRemaining != 0) {
splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining,
blkLocations[blkLocations.length-1].getHosts()));
}
} else if (length != 0) {
String[] splitHosts = getSplitHosts(blkLocations,0,length,clusterMap);
splits.add(new FileSplit(path, 0, length, splitHosts));
} else {
//Create empty hosts array for zero length files
splits.add(new FileSplit(path, 0, length, new String[0]));
}
}
LOG.debug("Total # of splits: " + splits.size());
return splits.toArray(new FileSplit[splits.size()]);
}


ok ,一切都完了 ,但是你看上面的代码可能会产生一个疑问:SequenceFile是存储的一个个的key-value值,这样分割文件的话,会不会破坏原有的数据结构,即 要是某一个key-value被分到了两个FileSplit肿么办 ?

见文章:http://www.cnblogs.com/serendipity/articles/2112613.html

个人感觉mapreduce的这个工具InputFormat比较乱,往往不看源代码,你永远也无法知道到底哪些文件被选上了 。
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签: