UrlResolution extends BaseExampleJob { @Override public int run(String[] args) throws Exception { if(args.length != 3) { failArguments("Wrong number of arguments"); return -1; } String input1 = args[0]; String input2 = args[1]; String output = args[2]; delete(output); TupleMRBuilder mr = new TupleMRBuilder(conf, "Pangool Url Resolution"); mr.addIntermediateSchema(getURLMapSchema()); mr.addIntermediateSchema(getURLRegisterSchema()); mr.addInput(new Path(input1), new TupleTextInputFormat(getURLMapSchema(), false, false, '\t', NO_QUOTE_CHARACTER, NO_ESCAPE_CHARACTER, null, null), new IdentityTupleMapper()); mr.addInput(new Path(input2), new TupleTextInputFormat(getURLRegisterSchema(), false, false, '\t', NO_QUOTE_CHARACTER, NO_ESCAPE_CHARACTER, null, null), new IdentityTupleMapper()); mr.setFieldAliases("urlMap", new Aliases().add("url", "nonCanonicalUrl")); mr.setGroupByFields("url"); mr.setOrderBy(new OrderBy().add("url", Order.ASC).addSchemaOrder(Order.ASC)); mr.setSpecificOrderBy("urlRegister", new OrderBy().add("timestamp", Order.ASC)); mr.setTupleReducer(new Handler()); mr.setOutput(new Path(output), new TupleTextOutputFormat(getURLRegisterSchema(), false, '\t', NO_QUOTE_CHARACTER, NO_ESCAPE_CHARACTER), ITuple.class, NullWritable.class); try { mr.createJob().waitForCompletion(true); } finally { mr.cleanUpInstanceFiles(); } return 1; } UrlResolution(); @Override int run(String[] args); static void main(String[] args);  }
@Test public void testPangool() throws Exception { ToolRunner.run(getConf(),new UrlResolution(),new String[] { TEST_FILE_URL_MAP, TEST_FILE_URL_REG, OUT_PANGOOL }); String outPangool = getReducerOutputAsText(OUT_PANGOOL); String expectedOutput = getOutputAsText(EXPECTED_OUTPUT); assertEquals(expectedOutput,outPangool); trash(OUT_PANGOOL); }