// Path: src/main/java/org/trend/hgraph/HBaseGraphConstants.java
// public class HBaseGraphConstants {
//   
//   public static final String HBASE_GRAPH_TABLE_VERTEX_NAME_KEY = "hbase.graph.table.vertex.name";
//   
//   public static final String HBASE_GRAPH_TABLE_EDGE_NAME_KEY = "hbase.graph.table.edge.name";
//   
//   public static final String HBASE_GRAPH_TABLE_COLFAM_PROPERTY_NAME = "property";
//   
//   public static final String HBASE_GRAPH_TABLE_COLFAM_PROPERTY_NAME_DELIMITER = "@";
// 
//   public static final String HBASE_GRAPH_TABLE_EDGE_DELIMITER_1 = "-->";
// 
//   public static final String HBASE_GRAPH_TABLE_EDGE_DELIMITER_2 = "-->";
// }


import java.io.IOException;
import java.util.Arrays;
import org.apache.commons.lang.Validate;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.mapreduce.RowCounter;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.trend.hgraph.HBaseGraphConstants;
      LOGGER.info("inputPath=" + inputPath);

      // HBaseConfiguration.merge(job.getConfiguration(),
      // HBaseConfiguration.create(job.getConfiguration()));

      FileInputFormat.setInputPaths(job, new Path(inputPath));
      job.setMapperClass(CalculateIntermediatePageRankMapper.class);
      job.setInputFormatClass(SequenceFileInputFormat.class);
      job.setMapOutputKeyClass(Text.class);
      job.setMapOutputValueClass(DoubleWritable.class);

      job.setReducerClass(CalculatePageRankReducer.class);
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(DoubleWritable.class);
      job.setOutputFormatClass(SequenceFileOutputFormat.class);
      String outputPath = outputBasePath + "/" + timestamp;
      LOGGER.info("outputPath=" + outputPath);
      FileOutputFormat.setOutputPath(job, new Path(outputPath));

      Utils.setAuthenticationToken(job, LOGGER);

    } catch (IOException e) {
      LOGGER.error("run " + jobName + " failed", e);
      throw e;
    }
    return job;
  }

  private static Job createInitialPageRankJob(Configuration conf, String outputBasePath,
      Class<? extends TableInputFormat> tableInputFormat)
// Path: src/main/java/org/trend/hgraph/HBaseGraphConstants.java
// public class HBaseGraphConstants {
//   
//   public static final String HBASE_GRAPH_TABLE_VERTEX_NAME_KEY = "hbase.graph.table.vertex.name";
//   
//   public static final String HBASE_GRAPH_TABLE_EDGE_NAME_KEY = "hbase.graph.table.edge.name";
//   
//   public static final String HBASE_GRAPH_TABLE_COLFAM_PROPERTY_NAME = "property";
//   
//   public static final String HBASE_GRAPH_TABLE_COLFAM_PROPERTY_NAME_DELIMITER = "@";
// 
//   public static final String HBASE_GRAPH_TABLE_EDGE_DELIMITER_1 = "-->";
// 
//   public static final String HBASE_GRAPH_TABLE_EDGE_DELIMITER_2 = "-->";
// }


// Path: src/main/java/org/trend/hgraph/mapreduce/pagerank/Driver.java
import java.io.IOException;
import java.util.Arrays;
import org.apache.commons.lang.Validate;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.mapreduce.RowCounter;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.trend.hgraph.HBaseGraphConstants;

      LOGGER.info("inputPath=" + inputPath);

      // HBaseConfiguration.merge(job.getConfiguration(),
      // HBaseConfiguration.create(job.getConfiguration()));

      FileInputFormat.setInputPaths(job, new Path(inputPath));
      job.setMapperClass(CalculateIntermediatePageRankMapper.class);
      job.setInputFormatClass(SequenceFileInputFormat.class);
      job.setMapOutputKeyClass(Text.class);
      job.setMapOutputValueClass(DoubleWritable.class);

      job.setReducerClass(CalculatePageRankReducer.class);
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(DoubleWritable.class);
      job.setOutputFormatClass(SequenceFileOutputFormat.class);
      String outputPath = outputBasePath + "/" + timestamp;
      LOGGER.info("outputPath=" + outputPath);
      FileOutputFormat.setOutputPath(job, new Path(outputPath));

      Utils.setAuthenticationToken(job, LOGGER);

    } catch (IOException e) {
      LOGGER.error("run " + jobName + " failed", e);
      throw e;
    }
    return job;
  }

  private static Job createInitialPageRankJob(Configuration conf, String outputBasePath,
      Class<? extends TableInputFormat> tableInputFormat)
      throws IOException {