LuceneSegmentInputFormat extends InputFormat { @Override public List<LuceneSegmentInputSplit> getSplits(JobContext context) throws IOException, InterruptedException { Configuration configuration = context.getConfiguration(); LuceneStorageConfiguration lucene2SeqConfiguration = new LuceneStorageConfiguration(configuration); List<LuceneSegmentInputSplit> inputSplits = Lists.newArrayList(); List<Path> indexPaths = lucene2SeqConfiguration.getIndexPaths(); for (Path indexPath : indexPaths) { ReadOnlyFileSystemDirectory directory = new ReadOnlyFileSystemDirectory(FileSystem.get(configuration), indexPath, false, configuration); SegmentInfos segmentInfos = new SegmentInfos(); segmentInfos.read(directory); for (SegmentCommitInfo segmentInfo : segmentInfos) { LuceneSegmentInputSplit inputSplit = new LuceneSegmentInputSplit(indexPath, segmentInfo.info.name, segmentInfo.sizeInBytes()); inputSplits.add(inputSplit); LOG.info("Created {} byte input split for index '{}' segment {}", segmentInfo.sizeInBytes(), indexPath.toUri(), segmentInfo.info.name); } } return inputSplits; }  @Override List<LuceneSegmentInputSplit> getSplits(JobContext context); @Override RecordReader<Text, NullWritable> createRecordReader(InputSplit inputSplit, TaskAttemptContext context);  }
@Test public void testGetSplits() throws IOException, InterruptedException { SingleFieldDocument doc1 = new SingleFieldDocument("1", "This is simple document 1"); SingleFieldDocument doc2 = new SingleFieldDocument("2", "This is simple document 2"); SingleFieldDocument doc3 = new SingleFieldDocument("3", "This is simple document 3"); commitDocuments(getDirectory(getIndexPath1AsFile()), doc1); commitDocuments(getDirectory(getIndexPath1AsFile()), doc2); commitDocuments(getDirectory(getIndexPath1AsFile()), doc3); List<LuceneSegmentInputSplit> splits = inputFormat.getSplits(jobContext); Assert.assertEquals(3, splits.size()); }