Here's how I solved the problem using a custom InputFormat... the key
part is in listStatus(), where we traverse the directory tree. Since
HDFS doesn't have links this code is probably safe, but if you have a
filesystem with cycles you will get trapped.
Ian
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.List;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.ArrayDeque;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.InvalidInputException;
import org.apache.hadoop.mapred.LineRecordReader;
public class TrecWebInputFormat extends FileInputFormat<DocLocation, Text> {
@Override
public boolean isSplitable(FileSystem fs, Path filename) {
return false;
}
@Override
public RecordReader<DocLocation, Text>
getRecordReader(InputSplit split, JobConf job, Reporter reporter)
throws IOException {
return new TrecWebRecordReader(job, (FileSplit)split);
}
// The following are incomprehensibly private in FileInputFormat...
private static final PathFilter hiddenFileFilter = new PathFilter(){
public boolean accept(Path p){
String name = p.getName();
return !name.startsWith("_") && !name.startsWith(".");
}
};
/**
* Proxy PathFilter that accepts a path only if all filters given in the
* constructor do. Used by the listPaths() to apply the built-in
* hiddenFileFilter together with a user provided one (if any).
*/
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 : filters) {
if (!filter.accept(path)) {
return false;
}
}
return true;
}
}
@Override
protected FileStatus[] listStatus(JobConf job)
throws IOException {
Path[] dirs = getInputPaths(job);
if (dirs.length == 0) {
throw new IOException("No input paths specified in job");
}
List<FileStatus> result = new ArrayList<FileStatus>();
List<IOException> errors = new ArrayList<IOException>();
ArrayDeque<FileStatus> stats = new ArrayDeque<FileStatus>(dirs.length);
// creates a MultiPathFilter with the hiddenFileFilter and the
// user provided one (if any).
List<PathFilter> filters = new ArrayList<PathFilter>();
filters.add(hiddenFileFilter);
PathFilter jobFilter = getInputPathFilter(job);
if (jobFilter != null) {
filters.add(jobFilter);
}
PathFilter inputFilter = new MultiPathFilter(filters);
// Set up traversal from input paths, which may be globs
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) {
stats.add(globStat);
}
}
}
while (!stats.isEmpty()) {
FileStatus stat = stats.pop();
if (stat.isDir()) {
FileSystem fs = stat.getPath().getFileSystem(job);
for (FileStatus sub: fs.listStatus(stat.getPath(),
inputFilter)) {
stats.push(sub);
}
} else {
result.add(stat);
}
}
if (!errors.isEmpty()) {
throw new InvalidInputException(errors);
}
LOG.info("Total input paths to process : " + result.size());
return result.toArray(new FileStatus[result.size()]);
}
public static class TrecWebRecordReader
implements RecordReader<DocLocation, Text> {
private CompressionCodecFactory compressionCodecs = null;
private long start;
private long end;
private long pos;
private Path file;
private LineRecordReader.LineReader in;
public TrecWebRecordReader(JobConf job, FileSplit split)
throws IOException {
file = split.getPath();
start = 0;
end = split.getLength();
compressionCodecs = new CompressionCodecFactory(job);
CompressionCodec codec = compressionCodecs.getCodec(file);
FileSystem fs = file.getFileSystem(job);
FSDataInputStream fileIn = fs.open(file);
if (codec != null) {
in = new
LineRecordReader.LineReader(codec.createInputStream(fileIn), job);
} else {
in = new LineRecordReader.LineReader(fileIn, job);
}
pos = 0;
}
public DocLocation createKey() {
return new DocLocation();
}
public Text createValue() {
return new Text();
}
public synchronized boolean next(DocLocation key, Text value)
throws IOException {
Text line = new Text();
StringBuilder buf = new StringBuilder();
boolean in_doc = false;
try {
while (true) {
int size = in.readLine(line);
if (size <= 0)
break;
pos += size;
if (!in_doc && line.find("<DOC>") >= 0) {
in_doc = true;
key.offset = pos;
key.filename = file.toString();
}
if (in_doc) {
buf.append(line.toString()).append("\n");
if (line.find("</DOC>") >= 0) {
in_doc = false;
break;
}
}
}
} catch (java.io.EOFException e) {
}
if (buf.length() > 0) {
value.set(buf.toString());
key.length = value.getLength();
return true;
} else {
return false;
}
}
public synchronized long getPos() throws IOException {
return pos;
}
public float getProgress() {
return Math.min(1.0f, (pos) / (float)(end));
}
public synchronized void close() throws IOException {
if (in != null) {
in.close();
}
}
}
}
