GameOfLifeJob extends BaseExampleJob implements Serializable { @Override public int run(String[] args) throws Exception { if(args.length != 3) { failArguments("Wrong number of arguments"); return -1; } String output = args[0]; String input = GameOfLifeJob.class.getName() + "-prepared-input"; delete(output); delete(input); final int gridSize = Integer.parseInt(args[1]); int parallelism = Integer.parseInt(args[2]); int maxCombinations = (int) Math.pow(2, gridSize * gridSize); int splitSize = maxCombinations / parallelism; FileSystem fS = FileSystem.get(conf); BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(fS.create(new Path(input)))); for(int i = 0; i < parallelism; i++) { writer.write(((i * splitSize) + 1) + "\t" + ((i + 1) * splitSize) + "\n"); } writer.close(); final int maxX = conf.getInt("gol.max_x", 32); final int maxY = conf.getInt("gol.max_y", 32); final int iterations = conf.getInt("gol.iterations", 1000); Log.info("using parameters: maxX grid: " + maxX + " maxY grid: " + maxY + " max #iterations: " + iterations); final Schema schema = new Schema("minMax", Fields.parse("min:int, max:int")); TupleMRBuilder job = new TupleMRBuilder(conf); job.addIntermediateSchema(schema); job.setGroupByFields("min", "max"); job.setCustomPartitionFields("min"); job.addInput(new Path(input), new HadoopInputFormat(TextInputFormat.class), new TupleMapper<LongWritable, Text>() { Tuple tuple = new Tuple(schema); @Override public void map(LongWritable key, Text value, TupleMRContext context, Collector collector) throws IOException, InterruptedException { String[] fields = value.toString().split("\t"); tuple.set("min", Integer.parseInt(fields[0])); tuple.set("max", Integer.parseInt(fields[1])); collector.write(tuple); } }); job.setTupleReducer(new TupleReducer<Text, NullWritable>() { public void reduce(ITuple group, Iterable<ITuple> tuples, TupleMRContext context, Collector collector) throws IOException, InterruptedException, TupleMRException { int min = (Integer) group.get("min"), max = (Integer) group.get("max"); for(int i = min; i < max; i++) { try { GameOfLife gameOfLife = new GameOfLife(gridSize, GameOfLife.longToBytes((long) i), maxX, maxY, iterations); while(true) { gameOfLife.nextCycle(); } } catch(GameOfLifeException e) { context.getHadoopContext().progress(); context.getHadoopContext().getCounter("stats", e.getCauseMessage() + "").increment(1); if(e.getCauseMessage().equals(CauseMessage.CONVERGENCE_REACHED)) { collector.write( new Text(Arrays.toString(GameOfLife.longToBytes((long) i)) + "\t" + e.getIterations()), NullWritable.get()); } } } }; }); job.setOutput(new Path(output), new HadoopOutputFormat(TextOutputFormat.class), Text.class, NullWritable.class); try { job.createJob().waitForCompletion(true); } finally { job.cleanUpInstanceFiles(); } delete(input); return 0; } GameOfLifeJob(); @Override int run(String[] args); static void main(String args[]);  }
@Test public void test() throws Exception { trash(OUTPUT); ToolRunner.run(getConf(),new GameOfLifeJob(), new String[] { OUTPUT, "2", "1" }); List<String> lines = Files.readLines(new File(OUTPUT + "/part-r-00000"), Charset.forName("UTF-8")); String[] spl = lines.get(0).split("\t"); assertEquals("[0, 0, 0, 0, 0, 0, 0, 7]", spl[0]); assertEquals("2", spl[1]); spl = lines.get(1).split("\t"); assertEquals("[0, 0, 0, 0, 0, 0, 0, 11]", spl[0]); assertEquals("2", spl[1]); spl = lines.get(2).split("\t"); assertEquals("[0, 0, 0, 0, 0, 0, 0, 13]", spl[0]); assertEquals("2", spl[1]); spl = lines.get(3).split("\t"); assertEquals("[0, 0, 0, 0, 0, 0, 0, 14]", spl[0]); assertEquals("2", spl[1]); spl = lines.get(4).split("\t"); assertEquals("[0, 0, 0, 0, 0, 0, 0, 15]", spl[0]); assertEquals("1", spl[1]); trash(OUTPUT); }