SecondarySort 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); List<Field> fields = new ArrayList<Field>(); fields.add(Field.create("intField", Type.INT)); fields.add(Field.create("strField", Type.STRING)); fields.add(Field.create("longField", Type.LONG)); fields.add(Field.create("doubleField", Type.DOUBLE)); Schema schema = new Schema("schema", fields); TupleMRBuilder mr = new TupleMRBuilder(conf, "Pangool Secondary Sort"); mr.addIntermediateSchema(schema); mr.setGroupByFields("intField", "strField"); mr.setOrderBy(new OrderBy().add("intField", Order.ASC).add("strField", Order.ASC) .add("longField", Order.ASC)); mr.setTupleReducer(new Handler()); mr.addInput(new Path(input), new HadoopInputFormat(TextInputFormat.class), new IProcessor()); mr.setOutput(new Path(output), new HadoopOutputFormat(TextOutputFormat.class), Text.class, DoubleWritable.class); try { mr.createJob().waitForCompletion(true); } finally { mr.cleanUpInstanceFiles(); } return 1; } SecondarySort(); @Override int run(String[] args); static void main(String[] args);  }
@Test public void testPangool() throws Exception { ToolRunner.run(getConf(),new SecondarySort(),new String[] { TEST_FILE, OUT_PANGOOL }); String outPangool = getReducerOutputAsText(OUT_PANGOOL); String expectedOutput = getOutputAsText(EXPECTED_OUTPUT); assertEquals(expectedOutput, outPangool); trash(OUT_PANGOOL); }