UserActivityNormalizer extends BaseExampleJob { @Override public int run(String[] args) throws Exception { if(args.length != 2) { failArguments("Wrong number of arguments"); return -1; } String input = args[0]; String output = args[1]; delete(output); List<Field> fields = new ArrayList<Field>(); fields.add(Field.create("user", Type.STRING)); fields.add(Field.create("feature", Type.STRING)); fields.add(Field.create("all", Type.BOOLEAN)); fields.add(Field.create("clicks", Type.INT)); Schema schema = new Schema("my_schema", fields); TupleMRBuilder mr = new TupleMRBuilder(conf); mr.addIntermediateSchema(schema); mr.setGroupByFields("user", "all", "feature"); mr.setOrderBy(new OrderBy().add("user", Order.ASC).add("all", Order.DESC).add("feature", Order.ASC)); mr.setRollupFrom("user"); mr.setTupleCombiner(new CountCombinerHandler()); mr.setTupleReducer(new NormalizingHandler()); mr.setOutput(new Path(output), new HadoopOutputFormat(TextOutputFormat.class), Text.class, NullWritable.class); mr.addInput(new Path(input), new HadoopInputFormat(TextInputFormat.class), new UserActivityProcessor()); try { mr.createJob().waitForCompletion(true); } finally { mr.cleanUpInstanceFiles(); } return 1; } UserActivityNormalizer(); @Override int run(String[] args); static void main(String args[]);  }
@Test public void test() throws Exception { Configuration conf = new Configuration(); FileSystem fS = FileSystem.get(conf); HadoopUtils.deleteIfExists(fS, new Path(OUTPUT)); Files.write( "user1" + "\t" + "feat1" + "\t" + "10" + "\n" + "user1" + "\t" + "feat1" + "\t" + "20" + "\n" + "user1" + "\t" + "feat2" + "\t" + "30" + "\n" + "user2" + "\t" + "feat1" + "\t" + "10" + "\n" + "user2" + "\t" + "feat2" + "\t" + "10" + "\n" + "user2" + "\t" + "feat3" + "\t" + "10" + "\n" , new File(INPUT), Charset.forName("UTF-8")); ToolRunner.run(getConf(),new UserActivityNormalizer(), new String[] { INPUT, OUTPUT }); 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("user1")) { if(fields[1].equals("feat1")) { assertEquals(0.5, Double.parseDouble(fields[2]), 0.00001); validatedOutputLines++; } else if(fields[1].equals("feat2")) { assertEquals(0.5, Double.parseDouble(fields[2]), 0.00001); validatedOutputLines++; } } else if(fields[0].equals("user2")) { if(fields[1].equals("feat1")) { assertEquals(0.33, Double.parseDouble(fields[2]), 0.01); validatedOutputLines++; } else if(fields[1].equals("feat2")) { assertEquals(0.33, Double.parseDouble(fields[2]), 0.01); validatedOutputLines++; } else if(fields[1].equals("feat3")) { assertEquals(0.33, Double.parseDouble(fields[2]), 0.01); validatedOutputLines++; } } } assertEquals(5, validatedOutputLines); HadoopUtils.deleteIfExists(fS, new Path(INPUT)); HadoopUtils.deleteIfExists(fS, new Path(OUTPUT)); }