TopicFingerprint extends BaseExampleJob { @Override public int run(String[] args) throws Exception { if(args.length != 3) { failArguments("Wrong number of arguments"); return -1; } delete(args[1]); Integer n = Integer.parseInt(args[2]); TupleMRBuilder builder = new TupleMRBuilder(conf, "Pangool Topic Fingerprint From Topical Word Count"); builder.addIntermediateSchema(TopicalWordCount.getSchema()); builder.setGroupByFields("topic"); builder.setOrderBy(new OrderBy().add("topic", Order.ASC).add("count", Order.DESC)); builder.addTupleInput(new Path(args[0]), new IdentityTupleMapper()); builder.setTupleOutput(new Path(args[1]), TopicalWordCount.getSchema()); builder.addNamedTupleOutput(OUTPUT_TOTALCOUNT, getOutputCountSchema()); builder.setTupleCombiner(new TopNWords(n)); builder.setTupleReducer(new TopNWords(n)); try { builder.createJob().waitForCompletion(true); } finally { builder.cleanUpInstanceFiles(); } return 1; } TopicFingerprint(); static Schema getOutputCountSchema(); @Override int run(String[] args); static void main(String[] args); final static String OUTPUT_TOTALCOUNT; }
@Test public void test() throws Exception { trash(OUTPUT); Configuration conf = new Configuration(); createInput(INPUT, conf); ToolRunner.run(getConf(), new TopicFingerprint(), new String[] { INPUT, OUTPUT, 2 + "" } ); Path outPath = new Path(OUTPUT + "/part-r-00000"); FileSystem fs = FileSystem.get(outPath.toUri(), conf); TupleFile.Reader reader = new TupleFile.Reader(fs, conf, outPath); Tuple tuple = new Tuple(reader.getSchema()); reader.next(tuple); assertEquals(1, tuple.get("topic")); assertEquals("a", tuple.get("word").toString()); reader.next(tuple); assertEquals(1, tuple.get("topic")); assertEquals("c", tuple.get("word").toString()); reader.next(tuple); assertEquals(2, tuple.get("topic")); assertEquals("a", tuple.get("word").toString()); reader.next(tuple); assertEquals(2, tuple.get("topic")); assertEquals("b", tuple.get("word").toString()); reader.close(); outPath = new Path(OUTPUT + "/" + TopicFingerprint.OUTPUT_TOTALCOUNT + "/" + "part-r-00000"); reader = new TupleFile.Reader(fs, conf, outPath); tuple = new Tuple(reader.getSchema()); reader.next(tuple); assertEquals(1, tuple.get("topic")); assertEquals(15, tuple.get("totalcount")); reader.next(tuple); assertEquals(2, tuple.get("topic")); assertEquals(19, tuple.get("totalcount")); reader.close(); trash(INPUT, OUTPUT); }