@Test public void ga_many() throws Exception { String[][] locations = { { }, { "a", "b" }, { "b", "c" }, { "a" }, { "c" }, { "b", "c" }, { "d" }, { "e", "f" }, { "a", "g" }, }; List<StageInputSplit> splits = new ArrayList<>(); long total = 0; for (int i = 0; i < 1000; i++) { long size = i * 10 + 100; splits.add(split(i, size, locations[i % locations.length])); total += size; } DefaultSplitCombiner combiner = new DefaultSplitCombiner(); for (int i = 1; i < 12; i += 2) { int slots = i * 5; int prefSlots = i * 10 / 9; List<StageInputSplit> combined = combine(combiner, slots, splits); assertThat(combined.size(), is(greaterThan(prefSlots))); assertSan(combined); long prefMaxSize = total * 2 / prefSlots; for (StageInputSplit split : combined) { assertThat(split.getLength(), is(lessThan(prefMaxSize))); } } }
@Override protected List<StageInputSplit> combine( JobContext context, List<StageInputSplit> splits) throws IOException, InterruptedException { Configuration conf = new Configuration(); conf.withSlotsPerInput(getMaxSplitsPerMapper(context)); conf.withPopulations(context.getConfiguration().getInt(KEY_POPULATIONS, DEFAULT_POPULATIONS)); conf.withGenerations(context.getConfiguration().getInt(KEY_GENERATIONS, DEFAULT_GENERATIONS)); conf.withMutations(context.getConfiguration().getFloat(KEY_MUTATION_RATIO, DEFAULT_MUTATION_RATIO)); conf.withTinyLimit(context.getConfiguration().getLong(KEY_TINY_LIMIT, DEFAULT_TINY_LIMIT)); conf.withAverageTimeWeight(context.getConfiguration().getFloat( KEY_AVERAGE_TIME_WEIGHT, DEFAULT_AVERAGE_TIME_WEIGHT)); conf.withNonLocalPenaltyRatio(context.getConfiguration().getFloat( KEY_NON_LOCAL_PENALTY_RATIO, DEFAULT_NON_LOCAL_PENALTY_RATIO)); conf.withInitialLocalityRatio(context.getConfiguration().getFloat( KEY_INITIAL_LOCALITY_RATIO, DEFAULT_INITIAL_LOCALITY_RATIO)); return combine(conf, splits); }
DefaultSplitCombiner extends SplitCombiner { @Override protected List<StageInputSplit> combine( JobContext context, List<StageInputSplit> splits) throws IOException, InterruptedException { Configuration conf = new Configuration(); conf.withSlotsPerInput(getMaxSplitsPerMapper(context)); conf.withPopulations(context.getConfiguration().getInt(KEY_POPULATIONS, DEFAULT_POPULATIONS)); conf.withGenerations(context.getConfiguration().getInt(KEY_GENERATIONS, DEFAULT_GENERATIONS)); conf.withMutations(context.getConfiguration().getFloat(KEY_MUTATION_RATIO, DEFAULT_MUTATION_RATIO)); conf.withTinyLimit(context.getConfiguration().getLong(KEY_TINY_LIMIT, DEFAULT_TINY_LIMIT)); conf.withAverageTimeWeight(context.getConfiguration().getFloat( KEY_AVERAGE_TIME_WEIGHT, DEFAULT_AVERAGE_TIME_WEIGHT)); conf.withNonLocalPenaltyRatio(context.getConfiguration().getFloat( KEY_NON_LOCAL_PENALTY_RATIO, DEFAULT_NON_LOCAL_PENALTY_RATIO)); conf.withInitialLocalityRatio(context.getConfiguration().getFloat( KEY_INITIAL_LOCALITY_RATIO, DEFAULT_INITIAL_LOCALITY_RATIO)); return combine(conf, splits); } }
DefaultSplitCombiner extends SplitCombiner { @Override protected List<StageInputSplit> combine( JobContext context, List<StageInputSplit> splits) throws IOException, InterruptedException { Configuration conf = new Configuration(); conf.withSlotsPerInput(getMaxSplitsPerMapper(context)); conf.withPopulations(context.getConfiguration().getInt(KEY_POPULATIONS, DEFAULT_POPULATIONS)); conf.withGenerations(context.getConfiguration().getInt(KEY_GENERATIONS, DEFAULT_GENERATIONS)); conf.withMutations(context.getConfiguration().getFloat(KEY_MUTATION_RATIO, DEFAULT_MUTATION_RATIO)); conf.withTinyLimit(context.getConfiguration().getLong(KEY_TINY_LIMIT, DEFAULT_TINY_LIMIT)); conf.withAverageTimeWeight(context.getConfiguration().getFloat( KEY_AVERAGE_TIME_WEIGHT, DEFAULT_AVERAGE_TIME_WEIGHT)); conf.withNonLocalPenaltyRatio(context.getConfiguration().getFloat( KEY_NON_LOCAL_PENALTY_RATIO, DEFAULT_NON_LOCAL_PENALTY_RATIO)); conf.withInitialLocalityRatio(context.getConfiguration().getFloat( KEY_INITIAL_LOCALITY_RATIO, DEFAULT_INITIAL_LOCALITY_RATIO)); return combine(conf, splits); }  }
DefaultSplitCombiner extends SplitCombiner { @Override protected List<StageInputSplit> combine( JobContext context, List<StageInputSplit> splits) throws IOException, InterruptedException { Configuration conf = new Configuration(); conf.withSlotsPerInput(getMaxSplitsPerMapper(context)); conf.withPopulations(context.getConfiguration().getInt(KEY_POPULATIONS, DEFAULT_POPULATIONS)); conf.withGenerations(context.getConfiguration().getInt(KEY_GENERATIONS, DEFAULT_GENERATIONS)); conf.withMutations(context.getConfiguration().getFloat(KEY_MUTATION_RATIO, DEFAULT_MUTATION_RATIO)); conf.withTinyLimit(context.getConfiguration().getLong(KEY_TINY_LIMIT, DEFAULT_TINY_LIMIT)); conf.withAverageTimeWeight(context.getConfiguration().getFloat( KEY_AVERAGE_TIME_WEIGHT, DEFAULT_AVERAGE_TIME_WEIGHT)); conf.withNonLocalPenaltyRatio(context.getConfiguration().getFloat( KEY_NON_LOCAL_PENALTY_RATIO, DEFAULT_NON_LOCAL_PENALTY_RATIO)); conf.withInitialLocalityRatio(context.getConfiguration().getFloat( KEY_INITIAL_LOCALITY_RATIO, DEFAULT_INITIAL_LOCALITY_RATIO)); return combine(conf, splits); }   }
DefaultSplitCombiner extends SplitCombiner { @Override protected List<StageInputSplit> combine( JobContext context, List<StageInputSplit> splits) throws IOException, InterruptedException { Configuration conf = new Configuration(); conf.withSlotsPerInput(getMaxSplitsPerMapper(context)); conf.withPopulations(context.getConfiguration().getInt(KEY_POPULATIONS, DEFAULT_POPULATIONS)); conf.withGenerations(context.getConfiguration().getInt(KEY_GENERATIONS, DEFAULT_GENERATIONS)); conf.withMutations(context.getConfiguration().getFloat(KEY_MUTATION_RATIO, DEFAULT_MUTATION_RATIO)); conf.withTinyLimit(context.getConfiguration().getLong(KEY_TINY_LIMIT, DEFAULT_TINY_LIMIT)); conf.withAverageTimeWeight(context.getConfiguration().getFloat( KEY_AVERAGE_TIME_WEIGHT, DEFAULT_AVERAGE_TIME_WEIGHT)); conf.withNonLocalPenaltyRatio(context.getConfiguration().getFloat( KEY_NON_LOCAL_PENALTY_RATIO, DEFAULT_NON_LOCAL_PENALTY_RATIO)); conf.withInitialLocalityRatio(context.getConfiguration().getFloat( KEY_INITIAL_LOCALITY_RATIO, DEFAULT_INITIAL_LOCALITY_RATIO)); return combine(conf, splits); }    }