### Question:
LouvainTableSynthesizer extends Configured implements Tool { @Override public int run(String[] args) throws Exception { Job job = null; try { int iteration = 0; if (!basePath.endsWith("/")) basePath = basePath + "/"; String inputPath = basePath + GIRAPH_FOLDER_BASE_NAME + FILE_NAME_SEPARATOR + iteration; String joinPath = basePath + GIRAPH_FOLDER_BASE_NAME + FILE_NAME_SEPARATOR + (iteration + 1); String outputPath = basePath + TABLE_BASE_NAME + FILE_NAME_SEPARATOR + iteration; Configuration mrConf = this.getConf(); job = Job.getInstance(mrConf); for(Map.Entry<String,String> entry : dgaConfiguration.getSystemProperties().entrySet()){ mrConf.set(entry.getKey(), entry.getValue()); } FileSystem fs = FileSystem.get(job.getConfiguration()); boolean nextFileExists = fs.exists(new Path(joinPath)); while (nextFileExists) { System.out.println("Processing " + inputPath + " and " + joinPath); job = Job.getInstance(mrConf); job.setJobName("Louvain Table Synthesizer " + iteration); job.setJarByClass(LouvainTableSynthesizer.class); job.setMapperClass(LouvainTableSynthesizerMapper.class); job.setReducerClass(LouvainTableSynthesizerReducer.class); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class); Path in = new Path(inputPath); Path joinIn = new Path(joinPath); Path out = new Path(outputPath); FileInputFormat.addInputPath(job, in); FileInputFormat.addInputPath(job, joinIn); FileOutputFormat.setOutputPath(job, out); job.waitForCompletion(true); inputPath = outputPath; iteration++; outputPath = basePath + TABLE_BASE_NAME + FILE_NAME_SEPARATOR + iteration; joinPath = basePath + GIRAPH_FOLDER_BASE_NAME + FILE_NAME_SEPARATOR + (iteration + 1); nextFileExists = fs.exists(new Path(joinPath)); } } catch (IOException e) { e.printStackTrace(); return -1; } catch (InterruptedException e) { e.printStackTrace(); return -1; } catch (ClassNotFoundException e) { e.printStackTrace(); return -1; } return 0; } LouvainTableSynthesizer(String basePath, DGAConfiguration dgaConfiguration); @Override int run(String[] args); static final String TABLE_BASE_NAME; static final String GIRAPH_FOLDER_BASE_NAME; static final String FILE_NAME_SEPARATOR; }### Answer:
@Test public void testReducer() throws Exception { List<Text> list = new LinkedList<Text>(); list.add(new Text("1\t2:0")); list.add(new Text("2\t2:0")); list.add(new Text("3\t2:0")); list.add(new Text("6:1")); list.add(new Text("5\t2:0")); reduceDriver.addInput(new Text("2"), list); reduceDriver.addOutput(new Text("1\t2\t6"), NullWritable.get()); reduceDriver.addOutput(new Text("2\t2\t6"), NullWritable.get()); reduceDriver.addOutput(new Text("3\t2\t6"), NullWritable.get()); reduceDriver.addOutput(new Text("5\t2\t6"), NullWritable.get()); List<Pair<Text, NullWritable>> results = reduceDriver.run(); assertTrue(results.size() == 4); }