}
// open the file and seek to the start of the split
FileSystem fs = file.getFileSystem(ctx.getConfiguration());
FSDataInputStream fileIn = fs.open(split.getPath());
InputStream dis = new BufferedInputStream(fileIn);
scanner = new CommitLogColumnarScanner(new DataInputStream(dis),
Descriptor.fromFilename(split.getPath().getName()).version, cfId);
this.pos = start;
iterator = scanner.observable()
.onErrorFlatMap(new Func1<OnErrorThrowable, Observable<? extends AtomWritable>>() {
@Override