AvroTopicalWordCount extends BaseExampleJob { @Override public int run(String[] args) throws Exception { if (args.length != 2) { failArguments("Wrong number of arguments"); return -1; } delete(args[1]); TupleMRBuilder mr = new TupleMRBuilder(conf, "Pangool Topical Word Count"); mr.addIntermediateSchema(getSchema()); mr.setGroupByFields("my_avro"); MyAvroComparator customComp = new MyAvroComparator(getAvroSchema(), "topic", "word"); mr.setOrderBy(new OrderBy().add("my_avro", Order.ASC, Criteria.NullOrder.NULL_SMALLEST, customComp)); mr.addInput(new Path(args[0]), new HadoopInputFormat(TextInputFormat.class), new TokenizeMapper()); mr.setTupleOutput(new Path(args[1]), getSchema()); mr.setTupleReducer(new CountReducer()); mr.setTupleCombiner(new CountReducer()); try { mr.createJob().waitForCompletion(true); } finally { mr.cleanUpInstanceFiles(); } return 1; } AvroTopicalWordCount(); @Override int run(String[] args); static void main(String[] args);  }
@Test public void test() throws Exception { createInput(INPUT); Configuration conf = getConf(); ToolRunner.run(conf,new AvroTopicalWordCount(), new String[] { INPUT, OUTPUT }); assertEquals(6, assertOutput(OUTPUT + "/part-r-00000", new Configuration())); trash(INPUT, OUTPUT); }