MovingAverage 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]; Integer nDaysAverage = Integer.parseInt(args[2]); delete(output); List<Field> fields = new ArrayList<Field>(); fields.add(Field.create("url", Type.STRING)); fields.add(Field.create("date", Type.STRING)); fields.add(Field.create("visits", Type.INT)); Schema schema = new Schema("my_schema", fields); TupleMRBuilder mr = new TupleMRBuilder(conf); mr.addIntermediateSchema(schema); mr.setGroupByFields("url"); mr.setOrderBy(new OrderBy().add("url", Order.ASC).add("date", Order.ASC)); mr.setTupleReducer(new MovingAverageHandler(nDaysAverage)); mr.setOutput(new Path(output), new HadoopOutputFormat(TextOutputFormat.class), Text.class, NullWritable.class); mr.addInput(new Path(input), new HadoopInputFormat(TextInputFormat.class), new URLVisitsProcessor()); try { mr.createJob().waitForCompletion(true); } finally { mr.cleanUpInstanceFiles(); } return 1; } MovingAverage(); @Override int run(String[] args); static void main(String args[]);  }
@Test public void test() throws Exception { trash(OUTPUT); Files.write( "url1" + "\t" + "2011-02-28" + "\t" + "100" + "\n" + "url1" + "\t" + "2011-02-27" + "\t" + "50" + "\n" + "url1" + "\t" + "2011-02-24" + "\t" + "25" + "\n" + "url2" + "\t" + "2011-02-28" + "\t" + "100" + "\n" + "url2" + "\t" + "2011-02-26" + "\t" + "50" + "\n" + "url2" + "\t" + "2011-02-25" + "\t" + "25" + "\n" , new File(INPUT), Charset.forName("UTF-8")); ToolRunner.run(getConf(),new MovingAverage(), new String[] { INPUT, OUTPUT, "3" }); 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("url1")) { if(fields[1].equals("2011-02-28")) { assertEquals("75.0", fields[2]); validatedOutputLines++; } else if(fields[1].equals("2011-02-27")) { assertEquals("50.0", fields[2]); validatedOutputLines++; } else if(fields[1].equals("2011-02-24")) { assertEquals("25.0", fields[2]); validatedOutputLines++; } } else if(fields[0].equals("url2")) { if(fields[1].equals("2011-02-28")) { assertEquals("75.0", fields[2]); validatedOutputLines++; } else if(fields[1].equals("2011-02-26")) { assertEquals("37.5", fields[2]); validatedOutputLines++; } else if(fields[1].equals("2011-02-25")) { assertEquals("25.0", fields[2]); validatedOutputLines++; } } } assertEquals(6, validatedOutputLines); trash(INPUT, OUTPUT); }