MultiShakespeareIndexer extends BaseExampleJob implements Serializable { @Override public int run(String[] args) throws Exception { if(args.length != 2) { failArguments("Invalid number of arguments"); return -1; } delete(args[1]); TupleMRBuilder job = new TupleMRBuilder(conf); job.addIntermediateSchema(SCHEMA); job.setGroupByFields("line"); String input = args[0], output = args[1]; FileSystem fileSystem = FileSystem.get(conf); for(Category category : Category.values()) { String categoryString = category.toString().toLowerCase(); for(FileStatus fileStatus : fileSystem.listStatus(new Path(input + "/" + categoryString))) { job.addInput(fileStatus.getPath(), new HadoopInputFormat(TextInputFormat.class), new CategoryMapper(category, fileStatus.getPath().getName())); } job.addNamedOutput(categoryString, new TupleSolrOutputFormat(new File( "src/test/resources/shakespeare-solr"), job.getConf()), ITuple.class, NullWritable.class); } job.setOutput(new Path(output), new HadoopOutputFormat(NullOutputFormat.class), ITuple.class, NullWritable.class); job.setTupleReducer(new TupleReducer<ITuple, NullWritable>() { ITuple outTuple = new Tuple(OUT_SCHEMA); public void reduce(ITuple group, Iterable<ITuple> tuples, TupleMRContext context, Collector collector) throws IOException, InterruptedException, TupleMRException { for(ITuple tuple : tuples) { Category category = (Category) tuple.get("category"); outTuple.set("line", tuple.get("line")); outTuple.set("text", tuple.get("text")); outTuple.set("title", tuple.get("title")); collector.getNamedOutput(category.toString().toLowerCase()) .write(outTuple, NullWritable.get()); } } }); try { Job hadoopJob = job.createJob(); hadoopJob.waitForCompletion(true); } finally { job.cleanUpInstanceFiles(); } return 0; } MultiShakespeareIndexer(); @Override int run(String[] args); static void main(String args[]);  }
@Test public void test() throws Exception { trash(OUTPUT); ToolRunner.run(getConf(),new MultiShakespeareIndexer(), new String[] { INPUT, OUTPUT }); for(String category: new String[] { "comedies", "histories", "tragedies", "poetry" }) { assertTrue(new File(OUTPUT + "/" + category + "/part-00000/data/index").exists()); assertTrue(new File(OUTPUT + "/" + category + "/part-00000/conf/schema.xml").exists()); assertTrue(new File(OUTPUT + "/" + category + "/part-00000/conf/solrconfig.xml").exists()); assertTrue(new File(OUTPUT + "/" + category + "/part-00000/conf/stopwords.txt").exists()); } IndexReader r = IndexReader.open(FSDirectory.open(new File(OUTPUT + "/comedies/part-00000/data/index"))); assertEquals(1, r.maxDoc()); String document = r.document(0).toString(); assertTrue(document.contains("comedy")); r = IndexReader.open(FSDirectory.open(new File(OUTPUT + "/histories/part-00000/data/index"))); assertEquals(1, r.maxDoc()); document = r.document(0).toString(); assertTrue(document.contains("history")); r = IndexReader.open(FSDirectory.open(new File(OUTPUT + "/tragedies/part-00000/data/index"))); assertEquals(1, r.maxDoc()); document = r.document(0).toString(); assertTrue(document.contains("tragedy")); r = IndexReader.open(FSDirectory.open(new File(OUTPUT + "/poetry/part-00000/data/index"))); assertEquals(1, r.maxDoc()); document = r.document(0).toString(); assertTrue(document.contains("poetry")); trash(OUTPUT); }