SortJob implements Tool, Configurable { @Override public int run(String[] args) throws Exception { if(args.length != 2) { System.out.println("Invalid number of arguments\n\n" + "Usage: IdentityJob <input_path> <output_path>\n\n"); return -1; } String input = args[0]; String output = args[1]; FileSystem.get(conf).delete(new Path(output), true); TupleMRBuilder mr = new TupleMRBuilder(conf); mr.addIntermediateSchema(getSchema()); mr.setGroupByFields("line"); mr.setTupleReducer(new IdentityReducer()); mr.addInput(new Path(input), new HadoopInputFormat(TextInputFormat.class), new IdentityMap()); mr.setOutput(new Path(output), new HadoopOutputFormat(TextOutputFormat.class), Text.class, NullWritable.class); mr.createJob().waitForCompletion(true); return 0; }  @Override Configuration getConf(); @Override void setConf(Configuration conf); static Schema getSchema(); @Override int run(String[] args); static void main(String args[]);  }
@Test public void test() throws Exception { trash(OUTPUT); File iFile = new File(INPUT); Charset utf8 = Charset.forName("UTF-8"); String inputStr = "A-This is\nB-the file\nC-content\n"; Files.write(inputStr, iFile, utf8); ToolRunner.run(new SortJob(), new String[]{INPUT, OUTPUT}); String outputStr = Files.toString(new File(OUTPUT, "part-r-00000"), utf8); assertEquals(inputStr, outputStr); trash(INPUT, OUTPUT); }