SplitUtil { public static List<CompositeInputSplit> getCombinedCompositeSplits( List<InputSplit> oneInputSplits, long maxCombinedSplitSize, Configuration conf) throws IOException, InterruptedException { List<CompositeInputSplit> compositeInputSplits = new ArrayList<CompositeInputSplit>(oneInputSplits.size()); for (List<InputSplit> inputSplits : getCombinedSplits(oneInputSplits, maxCombinedSplitSize, conf)) { compositeInputSplits.add(new CompositeInputSplit(inputSplits)); } return compositeInputSplits; }  static List<List<InputSplit>> getCombinedSplits(
          List<InputSplit> oneInputSplits, long maxCombinedSplitSize, Configuration conf); static List<CompositeInputSplit> getCombinedCompositeSplits(
          List<InputSplit> oneInputSplits, long maxCombinedSplitSize, Configuration conf); static List<CompositeInputSplit> getCombinedCompositeSplits(
          List<InputSplit> oneInputSplits, Configuration conf); static void serializeInputSplit(Configuration conf, DataOutputStream out, InputSplit split); static InputSplit deserializeInputSplit(Configuration conf, DataInputStream in); static final String COMBINE_SPLIT_SIZE; }
@Test public void test1() throws IOException, InterruptedException { List<InputSplit> rawSplits = new ArrayList<InputSplit>(); rawSplits.add(new DummyInputSplit(500, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(400, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(400, new String[] { "l1", "l4", "l5" })); List<CompositeInputSplit> result = SplitUtil.getCombinedCompositeSplits(rawSplits, conf); Assert.assertEquals(result.size(), 2); int index = 0; for (CompositeInputSplit split : result) { int len = split.getSplits().size(); if (index == 0) { Assert.assertEquals(2, len); checkLocations(split.getLocations(), new String[] { "l1", "l2", "l3" }); Assert.assertEquals(500, split.getLength(0)); Assert.assertEquals(400, split.getLength(1)); } else { Assert.assertEquals(1, len); checkLocations(split.getLocations(), new String[] { "l1", "l4", "l5" }); Assert.assertEquals(400, split.getLength(0)); } index++; } }
@Test public void test2() throws IOException, InterruptedException { ArrayList<InputSplit> rawSplits = new ArrayList<InputSplit>(); rawSplits.add(new DummyInputSplit(600, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(700, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(800, new String[] { "l1", "l4", "l5" })); List<CompositeInputSplit> result = SplitUtil.getCombinedCompositeSplits(rawSplits, conf); Assert.assertEquals(result.size(), 3); int index = 0; for (CompositeInputSplit split : result) { int len = split.getSplits().size(); if (index == 0) { checkLocations(split.getLocations(), new String[] { "l1", "l4", "l5" }); Assert.assertEquals(1, len); Assert.assertEquals(800, split.getLength(0)); } else if (index == 1) { checkLocations(split.getLocations(), new String[] { "l1", "l2", "l3" }); Assert.assertEquals(1, len); Assert.assertEquals(700, split.getLength(0)); } else { checkLocations(split.getLocations(), new String[] { "l1", "l2", "l3" }); Assert.assertEquals(1, len); Assert.assertEquals(600, split.getLength(0)); } index++; } }
@Test public void test3() throws IOException, InterruptedException { ArrayList<InputSplit> rawSplits = new ArrayList<InputSplit>(); rawSplits.add(new DummyInputSplit(500, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(200, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(100, new String[] { "l1", "l4", "l5" })); List<CompositeInputSplit> result = SplitUtil.getCombinedCompositeSplits(rawSplits, conf); Assert.assertEquals(1, result.size()); for (CompositeInputSplit split : result) { int len = split.getSplits().size(); Assert.assertEquals(3, len); checkLocations(split.getLocations(), new String[] { "l1", "l2", "l3", "l4", "l5" }); Assert.assertEquals(500, split.getLength(0)); Assert.assertEquals(200, split.getLength(1)); Assert.assertEquals(100, split.getLength(2)); } }
@Test public void test4() throws IOException, InterruptedException { ArrayList<InputSplit> rawSplits = new ArrayList<InputSplit>(); rawSplits.add(new DummyInputSplit(500, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(200, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(100, new String[] { "l1", "l4", "l5" })); rawSplits.add(new DummyInputSplit(100, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(200, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(500, new String[] { "l1", "l4", "l5" })); List<CompositeInputSplit> result = SplitUtil.getCombinedCompositeSplits(rawSplits, conf); Assert.assertEquals(2, result.size()); int idx = 0; for (CompositeInputSplit split : result) { int len = split.getSplits().size(); if (idx == 0) { Assert.assertEquals(2, len); checkLocations(split.getLocations(), new String[] { "l1", "l4", "l5" }); Assert.assertEquals(500, split.getLength(0)); Assert.assertEquals(100, split.getLength(1)); } else { Assert.assertEquals(4, len); Assert.assertEquals(500, split.getLength(0)); checkLocations(split.getLocations(), new String[] { "l1", "l2", "l3" }); Assert.assertEquals(200, split.getLength(1)); Assert.assertEquals(200, split.getLength(2)); Assert.assertEquals(100, split.getLength(3)); } idx++; } }
@Test public void test5() throws IOException, InterruptedException { ArrayList<InputSplit> rawSplits = new ArrayList<InputSplit>(); rawSplits.add(new DummyInputSplit(600, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(500, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(400, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(300, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(200, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(100, new String[] { "l1", "l2", "l3" })); List<CompositeInputSplit> result = SplitUtil.getCombinedCompositeSplits(rawSplits, conf); Assert.assertEquals(3, result.size()); int idx = 0; for (CompositeInputSplit split : result) { int len = split.getSplits().size(); if (idx == 0) { Assert.assertEquals(2, len); checkLocations(split.getLocations(), new String[] { "l1", "l2", "l3" }); Assert.assertEquals(600, split.getLength(0)); Assert.assertEquals(400, split.getLength(1)); } else if (idx == 1) { Assert.assertEquals(3, len); checkLocations(split.getLocations(), new String[] { "l1", "l2", "l3" }); Assert.assertEquals(500, split.getLength(0)); Assert.assertEquals(300, split.getLength(1)); Assert.assertEquals(200, split.getLength(2)); } else { Assert.assertEquals(1, len); checkLocations(split.getLocations(), new String[] { "l1", "l2", "l3" }); Assert.assertEquals(100, split.getLength(0)); } idx++; } }
@Test public void test6() throws IOException, InterruptedException { ArrayList<InputSplit> rawSplits = new ArrayList<InputSplit>(); rawSplits.add(new DummyInputSplit(100, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(200, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(300, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(400, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(500, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(600, new String[] { "l1", "l2", "l3" })); List<CompositeInputSplit> result = SplitUtil.getCombinedCompositeSplits(rawSplits, conf); Assert.assertEquals(3, result.size()); int idx = 0; for (CompositeInputSplit split : result) { int len = split.getSplits().size(); if (idx == 0) { Assert.assertEquals(2, len); checkLocations(split.getLocations(), new String[] { "l1", "l2", "l3" }); Assert.assertEquals(600, split.getLength(0)); Assert.assertEquals(400, split.getLength(1)); } else if (idx == 1) { Assert.assertEquals(3, len); checkLocations(split.getLocations(), new String[] { "l1", "l2", "l3" }); Assert.assertEquals(500, split.getLength(0)); Assert.assertEquals(300, split.getLength(1)); Assert.assertEquals(200, split.getLength(2)); } else { Assert.assertEquals(1, len); checkLocations(split.getLocations(), new String[] { "l1", "l2", "l3" }); Assert.assertEquals(100, split.getLength(0)); } idx++; } }
@Test public void test7() throws IOException, InterruptedException { ArrayList<InputSplit> rawSplits = new ArrayList<InputSplit>(); rawSplits.add(new DummyInputSplit(100, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(100, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(200, new String[] { "l1", "l4", "l5" })); List<CompositeInputSplit> result = SplitUtil.getCombinedCompositeSplits(rawSplits, conf); Assert.assertEquals(result.size(), 1); for (CompositeInputSplit split : result) { int len = split.getSplits().size(); Assert.assertEquals(3, len); checkLocations(split.getLocations(), new String[] { "l1", "l2", "l3", "l4", "l5" }); Assert.assertEquals(200, split.getLength(0)); Assert.assertEquals(100, split.getLength(1)); Assert.assertEquals(100, split.getLength(2)); } }
@Test public void test8() throws IOException, InterruptedException { ArrayList<InputSplit> rawSplits = new ArrayList<InputSplit>(); rawSplits.add(new DummyInputSplit(100, new String[] { "l1", "l2", "l3" })); rawSplits.add(new DummyInputSplit(200, new String[] { "l3", "l4", "l5" })); rawSplits.add(new DummyInputSplit(400, new String[] { "l5", "l6", "l1" })); List<CompositeInputSplit> result = SplitUtil.getCombinedCompositeSplits(rawSplits, conf); Assert.assertEquals(result.size(), 1); for (CompositeInputSplit split : result) { int len = split.getSplits().size(); Assert.assertEquals(3, len); checkLocationOrdering(split.getLocations(), new String[] { "l5", "l3", "l1", "l2", "l4" }); Assert.assertEquals(400, split.getLength(0)); Assert.assertEquals(200, split.getLength(1)); Assert.assertEquals(100, split.getLength(2)); } }