LuceneIndexInputFormat extends InputFormat<IntWritable, T> { protected List<InputSplit> combineSplits(PriorityQueue<LuceneIndexInputSplit> splits, long maxCombinedIndexSizePerSplit, long maxNumIndexesPerSplit) { List<InputSplit> combinedSplits = Lists.newLinkedList(); LuceneIndexInputSplit currentSplit = splits.poll(); while (currentSplit != null) { while (currentSplit.getLength() < maxCombinedIndexSizePerSplit) { LuceneIndexInputSplit nextSplit = splits.peek(); if (nextSplit == null) { break; } if (currentSplit.getLength() + nextSplit.getLength() > maxCombinedIndexSizePerSplit) { break; } if (currentSplit.getIndexDirs().size() >= maxNumIndexesPerSplit) { break; } currentSplit.combine(nextSplit); splits.poll(); } combinedSplits.add(currentSplit); currentSplit = splits.poll(); } return combinedSplits; }  PathFilter getIndexDirPathFilter(Configuration conf); @Override List<InputSplit> getSplits(JobContext job); static void setQueries(List<String> queries, Configuration conf); static List<String> getQueries(Configuration conf); static boolean queriesSet(Configuration conf); static void setInputPaths(List<Path> paths, Configuration conf); static Path[] getInputPaths(Configuration conf); static void setMaxCombinedIndexSizePerSplitBytes(long size, Configuration conf); static long getMaxCombinedIndexSizePerSplit(Configuration conf); static void setMaxNumIndexesPerSplit(long num, Configuration conf); static long getMaxNumIndexesPerSplit(Configuration conf); static final String QUERIES_KEY; static final String INPUT_PATHS_KEY; static final String MAX_NUM_INDEXES_PER_SPLIT_KEY; static final String MAX_COMBINED_INDEX_SIZE_PER_SPLIT_KEY; }
@Test public void testCombineSplits() throws Exception { DummyLuceneInputFormat lif = new DummyLuceneInputFormat(); PriorityQueue<LuceneIndexInputSplit> splits = new PriorityQueue<LuceneIndexInputSplit>(); String[] paths = new String[] { "/index/1", "/index/2", "/index/3", "/index/4", "/index/5", "/index/6"}; Long[] sizes = new Long[]{500L, 300L, 100L, 150L, 1200L, 500L}; for (int i = 0; i < paths.length; i++) { splits.add(new LuceneIndexInputSplit(Lists.newArrayList(new Path(paths[i])), sizes[i])); } List<InputSplit> combined = lif.combineSplits(splits, 1000L, 10000L); assertEquals(3, combined.size()); List<Path> dirs = ((LuceneIndexInputSplit) combined.get(0)).getIndexDirs(); Set<String> dirsStrings = Sets.newHashSet(Iterables.transform(dirs, Functions.toStringFunction())); assertEquals(3, dirsStrings.size()); assertTrue(dirsStrings.contains("/index/2")); assertTrue(dirsStrings.contains("/index/3")); assertTrue(dirsStrings.contains("/index/4")); dirs = ((LuceneIndexInputSplit) combined.get(1)).getIndexDirs(); dirsStrings = Sets.newHashSet(Iterables.transform(dirs, Functions.toStringFunction())); assertEquals(2, dirsStrings.size()); assertTrue(dirsStrings.contains("/index/1")); assertTrue(dirsStrings.contains("/index/6")); dirs = ((LuceneIndexInputSplit) combined.get(2)).getIndexDirs(); dirsStrings = Sets.newHashSet(Iterables.transform(dirs, Functions.toStringFunction())); assertEquals(1, dirsStrings.size()); assertTrue(dirsStrings.contains("/index/5")); }
@Test public void testCombineSplitsOneSplit() throws Exception { DummyLuceneInputFormat lif = new DummyLuceneInputFormat(); PriorityQueue<LuceneIndexInputSplit> splits = new PriorityQueue<LuceneIndexInputSplit>(); splits.add(new LuceneIndexInputSplit(Lists.newArrayList(new Path("/index/1")), 1500L)); List<InputSplit> combined = lif.combineSplits(splits, 1000L, 10000L); assertEquals(1, combined.size()); List<Path> dirs = ((LuceneIndexInputSplit) combined.get(0)).getIndexDirs(); Set<String> dirsStrings = Sets.newHashSet(Iterables.transform(dirs, Functions.toStringFunction())); assertEquals(1, dirsStrings.size()); assertTrue(dirsStrings.contains("/index/1")); }
@Test public void testCombineSplitsAllTooBig() throws Exception { DummyLuceneInputFormat lif = new DummyLuceneInputFormat(); PriorityQueue<LuceneIndexInputSplit> splits = new PriorityQueue<LuceneIndexInputSplit>(); String[] paths = new String[]{"/index/1", "/index/2", "/index/3"}; Long[] sizes = new Long[]{1500L, 1501L, 1502L}; for (int i = 0; i < paths.length; i++) { splits.add(new LuceneIndexInputSplit(Lists.newArrayList(new Path(paths[i])), sizes[i])); } List<InputSplit> combined = lif.combineSplits(splits, 1000L, 10000L); assertEquals(3, combined.size()); for (int i=0; i < paths.length; i++) { List<Path> dirs = ((LuceneIndexInputSplit) combined.get(i)).getIndexDirs(); List<String> dirsStrings = Lists.newLinkedList(Iterables.transform(dirs, Functions.toStringFunction())); assertEquals(1, dirsStrings.size()); assertEquals("/index/" + String.valueOf(i + 1), dirsStrings.get(0)); } }
@Test public void testCombineSplitsWithMaxNumberIndexesPerMapper() throws Exception { DummyLuceneInputFormat lif = new DummyLuceneInputFormat(); PriorityQueue<LuceneIndexInputSplit> splits = new PriorityQueue<LuceneIndexInputSplit>(); String[] paths = new String[1000]; long[] sizes = new long[1000]; for (int i = 0; i< 100; i++) { switch (i) { case 0: sizes[i] = 500L; paths[i] = "/index/500"; break; case 1: sizes[i] = 300L; paths[i] = "/index/300"; break; case 2: sizes[i] = 100L; paths[i] = "/index/100"; break; default: sizes[i] = 1L; paths[i] = "/index/small-" + i; } splits.add(new LuceneIndexInputSplit(Lists.newArrayList(new Path(paths[i])), sizes[i])); } List<InputSplit> combined = lif.combineSplits(splits, 150L, 10L); assertEquals(12, combined.size()); for (int i = 0; i < 9; i++) { LuceneIndexInputSplit split = (LuceneIndexInputSplit) combined.get(i); assertEquals(10L, split.getIndexDirs().size()); assertEquals(10L, split.getLength()); for (Path p : split.getIndexDirs()) { assertTrue(p.toString().startsWith("/index/small-")); } } LuceneIndexInputSplit split = (LuceneIndexInputSplit) combined.get(9); assertEquals(8, split.getIndexDirs().size()); assertEquals(107, split.getLength()); for (int i = 0; i < 7; i++) { assertTrue(split.getIndexDirs().get(i).toString().startsWith("/index/small-")); } assertEquals("/index/100", split.getIndexDirs().get(7).toString()); split = (LuceneIndexInputSplit) combined.get(10); assertEquals(1, split.getIndexDirs().size()); assertEquals(300, split.getLength()); assertEquals("/index/300", split.getIndexDirs().get(0).toString()); split = (LuceneIndexInputSplit) combined.get(11); assertEquals(1, split.getIndexDirs().size()); assertEquals(500, split.getLength()); assertEquals("/index/500", split.getIndexDirs().get(0).toString()); }