TopicalWordCountWithStopWords extends BaseExampleJob { @Override public int run(String[] args) throws Exception { if(args.length != 3) { failArguments("Wrong number of arguments"); return -1; } delete(args[1]); List<String> stopWords = Files.readLines(new File(args[2]), Charset.forName("UTF-8")); TupleMRBuilder cg = new TupleMRBuilder(conf, "Pangool Topical Word Count With Stop Words"); cg.addIntermediateSchema(TopicalWordCount.getSchema()); cg.setGroupByFields("topic", "word"); StopWordMapper mapper = new StopWordMapper(stopWords); cg.addInput(new Path(args[0]), new HadoopInputFormat(TextInputFormat.class), mapper); cg.setTupleOutput(new Path(args[1]), TopicalWordCount.getSchema()); cg.setTupleReducer(new CountReducer()); cg.setTupleCombiner(new CountReducer()); try { cg.createJob().waitForCompletion(true); } finally { cg.cleanUpInstanceFiles(); } return 1; } TopicalWordCountWithStopWords(); @Override int run(String[] args); static void main(String[] args);  }
@Test public void test() throws Exception { trash(INPUT, STOP_WORDS, OUTPUT); Configuration conf = new Configuration(); TestTopicalWordCount.createInput(INPUT); Files.touch(new File(STOP_WORDS)); ToolRunner.run(getConf(), new TopicalWordCountWithStopWords(), new String[] { INPUT, OUTPUT, STOP_WORDS }); assertEquals(6, TestTopicalWordCount.assertOutput(OUTPUT + "/part-r-00000", conf)); Files.write(("bar" + "\n" + "bloh").getBytes("UTF-8"), new File(STOP_WORDS)); trash(OUTPUT); ToolRunner.run(getConf(), new TopicalWordCountWithStopWords(), new String[] { INPUT, OUTPUT, STOP_WORDS }); assertEquals(3, TestTopicalWordCount.assertOutput(OUTPUT + "/part-r-00000", conf)); trash(INPUT, STOP_WORDS, OUTPUT); }