Robust/src/Benchmarks/MapReduce/Java/JobClient.java

   1 //package mapreduce;
   2
   3 public class JobClient{
   4
   5     public JobClient() {}
   6
   7     public static void runJob(Configuration conf) {
   8         Splitter splitter = new Splitter(conf.getInputfile(), conf.getM(), conf.getSeperator());
   9         Master master = new Master(conf.getM(), conf.getR(), splitter);
  10
  11         // split input file
  12         //System.printString("Split\n");
  13         master.split();
  14
  15         // do 'map'
  16         //System.printString("Map\n");
  17         MapWorker[] mworkers = master.assignMap();
  18         for(int i = 0; i < mworkers.length; ++i) {
  19             MapWorker mworker = mworkers[i];
  20             mworker.setMapreducer(conf.getMapReduce());
  21             mworker.map();
  22             mworker.partition();
  23         }
  24
  25         // register intermediate output from map workers to master
  26         //System.printString("Mapoutput\n");
  27         for(int i = 0; i < mworkers.length; ++i) {
  28             for(int j = 0; j < conf.getR(); ++j) {
  29                 String temp = mworkers[i].outputFile(j);
  30                 if(temp != null) {
  31                     master.addInterOutput(temp);
  32                 }
  33             }
  34             master.setMapFinish(mworkers[i].getID());
  35         }
  36         //assert(master.isMapFinish());
  37
  38         // do 'reduce'
  39         //System.printString("Reduce\n");
  40         ReduceWorker[] rworkers = master.assignReduce();
  41         for(int i = 0; i < rworkers.length; ++i) {
  42             ReduceWorker rworker = rworkers[i];
  43             rworker.setMapreducer(conf.getMapReduce());
  44             rworker.sortgroup();
  45             rworker.reduce();
  46         }
  47
  48         // merge all the intermediate output from reduce workers to master
  49         //System.printString("Merge\n");
  50         for(int i = 0; i < rworkers.length; ++i) {
  51             master.collectROutput(rworkers[i].getOutputFile());
  52             master.setReduceFinish(rworkers[i].getID());
  53         }
  54         //assert(master.isReduceFinish());
  55
  56         System./*out.println*/printString("Finish! Results are in the output file: " + master.getOutputFile() + "\n");
  57     }
  58
  59 }