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

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