AvroTweetsJoin extends BaseExampleJob { @Override public int run(String[] args) throws Exception { if(args.length != 3) { failArguments("Wrong number of arguments"); return -1; } Path tweetsPath = new Path(args[0]); Path retweetsPath = new Path(args[1]); Path outputPath = new Path(args[2]); delete(outputPath.toString()); TupleMRBuilder mr = new TupleMRBuilder(conf, "AvroTweetsJoin"); mr.addIntermediateSchema(getPangoolTweetSchema()); mr.addIntermediateSchema(getPangoolRetweetSchema()); mr.setGroupByFields("tweet_id"); mr.setOrderBy(new OrderBy().add("tweet_id", Order.ASC).addSchemaOrder(Order.ASC)); mr.setSpecificOrderBy("retweet", new OrderBy().add("username", Order.ASC)); mr.addInput(tweetsPath, new AvroInputFormat<Record>(getAvroTweetSchema()), new TweetsMapper()); mr.addInput(retweetsPath, new HadoopInputFormat(TextInputFormat.class), new RetweetsMapper()); mr.setOutput(outputPath, new AvroOutputFormat<Record>(getAvroOutputSchema()), AvroWrapper.class, NullWritable.class); mr.setTupleReducer(new Red()); try { Job job = mr.createJob(); job.waitForCompletion(true); } finally { mr.cleanUpInstanceFiles(); } return 0; } AvroTweetsJoin(); static org.apache.avro.Schema getAvroStringArraySchema(); static org.apache.avro.Schema getAvroTweetSchema(); static org.apache.avro.Schema getAvroOutputSchema(); @Override int run(String[] args); static void main(String[] args);  }
@Test public void test() throws Exception { createTweetInput(INPUT_TWEETS); createRetweetsInput(INPUT_RETWEETS); ToolRunner.run(getConf(), new AvroTweetsJoin(), new String[] { INPUT_TWEETS,INPUT_RETWEETS,OUTPUT }); assertOutput(OUTPUT + "/part-r-00000.avro", getConf()); trash(INPUT_TWEETS,INPUT_RETWEETS,OUTPUT); }