package com.yasikstudio.devrank.rank;

import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.PosixParser;
import org.apache.giraph.graph.GiraphJob;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;

public class DeveloperRankJob implements Tool {

  private static final String ISO8601 = "yyyy-MM-dd'T'HH:mm:ssZ";
  private Configuration configuration;

  @Override
  public int run(String[] args) throws Exception {
    Options options = new Options();
    options.addOption("h", "help", false, "Help");
    options.addOption("w", "workers", true, "Number of workers");
    options.addOption("i", "inputPath", true, "Input Path");
    options.addOption("o", "outputPath", true, "Output Path");

    HelpFormatter formatter = new HelpFormatter();
    if (args.length == 0) {
      formatter.printHelp(getClass().getName(), options, true);
      return 0;
    }
    CommandLineParser parser = new PosixParser();
    CommandLine cmd = parser.parse(options, args);
    if (cmd.hasOption('h')) {
      formatter.printHelp(getClass().getName(), options, true);
      return 0;
    }

    int workers = Integer.parseInt(cmd.getOptionValue('w', "10"));
    String inputPath = cmd.getOptionValue("inputPath");
    String outputPath = cmd.getOptionValue("outputPath");

    String s = new SimpleDateFormat(ISO8601).format(new Date());
    configuration.setLong("superstep", 10L);

    GiraphJob job = new GiraphJob(getConf(), "devrank-job-rank");
    job.setVertexClass(DeveloperRankVertex.class);
    job.setVertexInputFormatClass(DeveloperRankVertexInputFormat.class);
    job.setVertexOutputFormatClass(DeveloperRankVertexOutputFormat.class);
    FileInputFormat.addInputPath(job, new Path(inputPath));
    FileOutputFormat.setOutputPath(job, new Path(outputPath));

    job.setWorkerConfiguration(workers, workers, 100.0f);
    return job.run(true) ? 0 : -1;
  }

  @Override
  public Configuration getConf() {
    return configuration;
  }

  @Override
  public void setConf(Configuration configuration) {
    this.configuration = configuration;
  }
}

@@ -904,74 +904,8 @@
 %22);%0A
-    options.addOption(%22w%22, %22workers%22, true, %22Number of workers%22);%0A
     
@@ -1019,24 +1019,162 @@
 tput Path%22);
+%0A    options.addOption(%22w%22, %22workers%22, true, %22Number of workers%22);%0A    options.addOption(%22s%22, %22supersteps%22, true, %22Number of supersteps%22);
 %0A%0A    HelpFo
@@ -1609,24 +1609,93 @@
 w', %2210%22));%0A
+    long supersteps = Long.parseLong(cmd.getOptionValue('s', %2210%22));%0A
     String i
@@ -1905,11 +1905,18 @@
 p%22, 
-10L
+supersteps
 );%0A%0A

Add supersteps option for devrank-job-rank