SimpleSecondarySort extends BaseExampleJob { @Override public int run(String[] args) throws Exception { if(args.length != 2) { failArguments("Invalid number of arguments"); return -1; } String input = args[0]; String output = args[1]; delete(output); TupleMRBuilder builder = new TupleMRBuilder(conf); builder.addIntermediateSchema(getSchema()); builder.setGroupByFields("first"); builder.setOrderBy(new OrderBy().add("first", Order.ASC).add("second", Order.ASC)); builder.setTupleReducer(new Handler()); builder.setOutput(new Path(output), new HadoopOutputFormat(TextOutputFormat.class), Text.class, NullWritable.class); builder.addInput(new Path(input), new HadoopInputFormat(TextInputFormat.class), new IProcessor()); try { builder.createJob().waitForCompletion(true); } finally { builder.cleanUpInstanceFiles(); } return 1; } SimpleSecondarySort(); @Override int run(String[] args); static void main(String args[]);  }
@Test public void test() throws Exception { trash(OUTPUT); Files.write("10 3 \n 5 3 \n 5 30 \n 10 10", new File(INPUT), Charset.forName("UTF-8")); ToolRunner.run(getConf(),new SimpleSecondarySort(), new String[] { INPUT, OUTPUT }); String[][] expectedOutput = new String[][] { new String[] { "5", "3" }, new String[] { "5", "30" }, new String[] { "10", "3" }, new String[] { "10", "10" } }; int count = 0; for(String line : Files.readLines(new File(OUTPUT + "/part-r-00000"), Charset.forName("UTF-8"))) { String[] fields = line.split("\t"); assertEquals(fields[0], expectedOutput[count][0]); assertEquals(fields[1], expectedOutput[count][1]); count++; } trash(INPUT, OUTPUT); }