Step0Job { protected static Step0Output[] processOutput(List<Integer> keys, List<Step0Output> values) { int numMaps = values.size(); Step0Output[] sorted = new Step0Output[numMaps]; values.toArray(sorted); Arrays.sort(sorted); int[] orderedIds = new int[numMaps]; orderedIds[0] = 0; for (int p = 1; p < numMaps; p++) { orderedIds[p] = orderedIds[p - 1] + sorted[p - 1].size; } for (Step0Output value : values) { int order = ArrayUtils.indexOf(sorted, value); value.firstId = orderedIds[order]; } Step0Output[] reordered = new Step0Output[numMaps]; for (int p = 0; p < numMaps; p++) { reordered[keys.get(p)] = values.get(p); } return reordered; } Step0Job(Path base, Path dataPath, Path datasetPath); Step0Output[] run(Configuration conf);  }
@Test public void testProcessOutput() throws Exception { Random rng = RandomUtils.getRandom(); String descriptor = Utils.randomDescriptor(rng, NUM_ATTRIBUTES); double[][] source = Utils.randomDoubles(rng, descriptor, NUM_INSTANCES); int labelId = Utils.findLabel(descriptor); for (int index = 0; index < NUM_INSTANCES; index++) { source[index][labelId] = index; } String[] sData = Utils.double2String(source); Path dataPath = Utils.writeDataToTestFile(sData); Dataset dataset = DataLoader.generateDataset(descriptor, sData); DataConverter converter = new DataConverter(dataset); Job job = new Job(); job.setInputFormatClass(TextInputFormat.class); FileInputFormat.setInputPaths(job, dataPath); setMaxSplitSize(job.getConfiguration(), dataPath, NUM_MAPS); TextInputFormat input = new TextInputFormat(); List<InputSplit> splits = input.getSplits(job); assertEquals(NUM_MAPS, splits.size()); InputSplit[] sorted = new InputSplit[NUM_MAPS]; splits.toArray(sorted); Builder.sortSplits(sorted); List<Integer> keys = new ArrayList<Integer>(); List<Step0Output> values = new ArrayList<Step0Output>(); int[] expectedIds = new int[NUM_MAPS]; TaskAttemptContext context = new TaskAttemptContext(job.getConfiguration(), new TaskAttemptID()); for (int p = 0; p < NUM_MAPS; p++) { InputSplit split = sorted[p]; RecordReader<LongWritable, Text> reader = input.createRecordReader(split, context); reader.initialize(split, context); Long firstKey = null; int size = 0; while (reader.nextKeyValue()) { LongWritable key = reader.getCurrentKey(); Text value = reader.getCurrentValue(); if (firstKey == null) { firstKey = key.get(); expectedIds[p] = converter.convert(0, value.toString()).getLabel(); } size++; } keys.add(p); values.add(new Step0Output(firstKey, size)); } Step0Output[] partitions = Step0Job.processOutput(keys, values); int[] actualIds = Step0Output.extractFirstIds(partitions); assertTrue("Expected: " + Arrays.toString(expectedIds) + " But was: " + Arrays.toString(actualIds), Arrays.equals(expectedIds, actualIds)); }