@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")); }
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; }
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; } }
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; }  }
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); }
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; }