@Test public void ga_locality() throws Exception { DefaultSplitCombiner combiner = new DefaultSplitCombiner(); List<StageInputSplit> combined = combine(combiner, 2, list( split(1, 100, "a"), split(2, 100, "b"), split(3, 1, "c"), split(4, 1, "d"))); assertThat(combined.size(), is(2)); assertSan(combined); assertThat(find(combined, 1), is(not(find(combined, 2)))); }
@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); }    }