TopNHashTags extends BaseExampleJob { @Override public int run(String[] args) throws Exception { if(args.length != 3) { failArguments("Invalid number of arguments"); return -1; } String input = args[0]; String output = args[1]; int n = Integer.parseInt(args[2]); delete(output); List<Field> fields = new ArrayList<Field>(); fields.add(Field.create("location", Type.STRING)); fields.add(Field.create("date", Type.STRING)); fields.add(Field.create("hashtag", Type.STRING)); fields.add(Field.create("count", Type.INT)); Schema schema = new Schema("my_schema", fields); TupleMRBuilder mr = new TupleMRBuilder(conf); mr.addIntermediateSchema(schema); mr.setGroupByFields("location", "date", "hashtag"); mr.setOrderBy(new OrderBy().add("location", Order.ASC).add("date", Order.ASC) .add("hashtag", Order.ASC)); mr.setRollupFrom("date"); mr.setTupleReducer(new TweetsHandler(n)); mr.setOutput(new Path(output), new HadoopOutputFormat(TextOutputFormat.class), Text.class, NullWritable.class); mr.addInput(new Path(input), new HadoopInputFormat(TextInputFormat.class), new TweetsProcessor()); try { mr.createJob().waitForCompletion(true); } finally { mr.cleanUpInstanceFiles(); } return 0; } TopNHashTags(); @Override int run(String[] args); static void main(String[] args);  }
@Test public void test() throws Exception { trash(OUTPUT); BufferedWriter writer = new BufferedWriter(new FileWriter(INPUT)); long today = System.currentTimeMillis(); DateTime dateTime = new DateTime(today); String todayDate = dateTime.getYear() + "-" + dateTime.getMonthOfYear() + "-" + dateTime.getDayOfMonth(); long anotherDay = today - 60 * 60 * 1000 * 24 *10; dateTime = new DateTime(anotherDay); String anotherDayDate = dateTime.getYear() + "-" + dateTime.getMonthOfYear() + "-" + dateTime.getDayOfMonth(); writer.write(getTweet(today, "h1", "l1") + "\n"); writer.write(getTweet(today, "h1", "l1") + "\n"); writer.write(getTweet(today, "h2", "l1") + "\n"); writer.write(getTweet(today, "h2", "l1") + "\n"); writer.write(getTweet(today, "h1", "l1") + "\n"); writer.write(getTweet(today, "h3", "l2") + "\n"); writer.write(getTweet(today, "h4", "l2") + "\n"); writer.write(getTweet(today, "h4", "l2") + "\n"); writer.write(getTweet(today, "h4", "l2") + "\n"); writer.write(getTweet(today, "h4", "l2") + "\n"); writer.write(getTweet(today, "h1", "l1") + "\n"); writer.write(getTweet(anotherDay, "h2", "l1") + "\n"); writer.write(getTweet(anotherDay, "h2", "l1") + "\n"); writer.write(getTweet(anotherDay, "h1", "l1") + "\n"); writer.write(getTweet(anotherDay, "h4", "l2") + "\n"); writer.write(getTweet(anotherDay, "h3", "l2") + "\n"); writer.write(getTweet(anotherDay, "h3", "l2") + "\n"); writer.close(); ToolRunner.run(getConf(),new TopNHashTags(), new String[] { INPUT, OUTPUT, "1" }); int validatedOutputLines = 0; for(String line: Files.readLines(new File(OUTPUT + "/part-r-00000"), Charset.forName("UTF-8"))) { String[] fields = line.split("\t"); if(fields[0].equals("l1")) { if(fields[1].equals(todayDate)) { assertEquals("h1", fields[2]); validatedOutputLines++; } else if(fields[1].equals(anotherDayDate)) { assertEquals("h2", fields[2]); validatedOutputLines++; } } else if(fields[0].equals("l2")) { if(fields[1].equals(todayDate)) { assertEquals(fields[2], "h4"); validatedOutputLines++; } else if(fields[1].equals(anotherDayDate)) { assertEquals(fields[2], "h3"); validatedOutputLines++; } } } assertEquals(4, validatedOutputLines); trash(INPUT, OUTPUT); }