MapFile { public static long fix(FileSystem fs, Path dir, Class<? extends Writable> keyClass, Class<? extends Writable> valueClass, boolean dryrun, Configuration conf) throws Exception { String dr = (dryrun ? "[DRY RUN ] " : ""); Path data = new Path(dir, DATA_FILE_NAME); Path index = new Path(dir, INDEX_FILE_NAME); int indexInterval = conf.getInt(Writer.INDEX_INTERVAL, 128); if (!fs.exists(data)) { throw new Exception(dr + "Missing data file in " + dir + ", impossible to fix this."); } if (fs.exists(index)) { return -1; } SequenceFile.Reader dataReader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(data)); if (!dataReader.getKeyClass().equals(keyClass)) { throw new Exception(dr + "Wrong key class in " + dir + ", expected" + keyClass.getName() + ", got " + dataReader.getKeyClass().getName()); } if (!dataReader.getValueClass().equals(valueClass)) { throw new Exception(dr + "Wrong value class in " + dir + ", expected" + valueClass.getName() + ", got " + dataReader.getValueClass().getName()); } long cnt = 0L; Writable key = ReflectionUtils.newInstance(keyClass, conf); Writable value = ReflectionUtils.newInstance(valueClass, conf); SequenceFile.Writer indexWriter = null; if (!dryrun) { indexWriter = SequenceFile.createWriter(conf, SequenceFile.Writer.file(index), SequenceFile.Writer.keyClass(keyClass), SequenceFile.Writer.valueClass (LongWritable.class)); } try { long lastIndexPos = -1; long lastIndexKeyCount = Long.MIN_VALUE; long pos = dataReader.getPosition(); LongWritable position = new LongWritable(); long nextBlock = pos; boolean blockCompressed = dataReader.isBlockCompressed(); while(dataReader.next(key, value)) { if (blockCompressed) { long curPos = dataReader.getPosition(); if (curPos > nextBlock) { pos = nextBlock; nextBlock = curPos; } } if (cnt >= lastIndexKeyCount + indexInterval && pos > lastIndexPos) { position.set(pos); if (!dryrun) { indexWriter.append(key, position); } lastIndexPos = pos; lastIndexKeyCount = cnt; } if (!blockCompressed) { pos = dataReader.getPosition(); } cnt++; } } catch(Throwable t) { } dataReader.close(); if (!dryrun) indexWriter.close(); return cnt; } protected  MapFile(); static void rename(FileSystem fs, String oldName, String newName); static void delete(FileSystem fs, String name); static long fix(FileSystem fs, Path dir,
                         Class<? extends Writable> keyClass,
                         Class<? extends Writable> valueClass, boolean dryrun,
                         Configuration conf); static void main(String[] args); static final String INDEX_FILE_NAME; static final String DATA_FILE_NAME; }
@Test public void testFix() { final String INDEX_LESS_MAP_FILE = "testFix.mapfile"; int PAIR_SIZE = 20; MapFile.Writer writer = null; try { FileSystem fs = FileSystem.getLocal(conf); Path dir = new Path(TEST_DIR, INDEX_LESS_MAP_FILE); writer = createWriter(INDEX_LESS_MAP_FILE, IntWritable.class, Text.class); for (int i = 0; i < PAIR_SIZE; i++) writer.append(new IntWritable(0), new Text("value")); writer.close(); File indexFile = new File(".", "." + INDEX_LESS_MAP_FILE + "/index"); boolean isDeleted = false; if (indexFile.exists()) isDeleted = indexFile.delete(); if (isDeleted) assertTrue("testFix error !!!", MapFile.fix(fs, dir, IntWritable.class, Text.class, true, conf) == PAIR_SIZE); } catch (Exception ex) { fail("testFix error !!!"); } finally { IOUtils.cleanup(null, writer); } }