// 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.math.BigDecimal;
import java.math.RoundingMode;
import org.apache.commons.lang.time.StopWatch;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.trend.hgraph.HBaseGraphConstants;
    
    sw.reset(); sw.start();
    double oldPageRank = Utils.getPageRank(vertexTable, rowkey, Constants.PAGE_RANK_CQ_TMP_NAME);
    if (!pageRankEquals(oldPageRank, newPageRank, pageRankCompareScale)) {
      // collect pageRank changing count with counter
      context.getCounter(Counters.CHANGED_PAGE_RANK_COUNT).increment(1);
    }
    sw.stop();
    context.getCounter(Counters.CMP_OLD_NEW_PR_TIME_CONSUMED).increment(sw.getTime());

    context.write(key, new DoubleWritable(newPageRank));
  }

  static boolean pageRankEquals(double src, double dest, int scale) {
    BigDecimal a = new BigDecimal(src);
    BigDecimal b = new BigDecimal(dest);
    a = a.setScale(scale, RoundingMode.DOWN);
    b = b.setScale(scale, RoundingMode.DOWN);
    return a.compareTo(b) == 0 ? true : false;
  }

  /*
   * (non-Javadoc)
   * @see org.apache.hadoop.mapreduce.Reducer#setup(org.apache.hadoop.mapreduce.Reducer.Context)
   */
  @Override
  protected void setup(Context context) throws IOException, InterruptedException {
    Configuration conf = context.getConfiguration();

    vertexTable =
// 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/CalculatePageRankReducer.java
import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import org.apache.commons.lang.time.StopWatch;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.trend.hgraph.HBaseGraphConstants;

    
    sw.reset(); sw.start();
    double oldPageRank = Utils.getPageRank(vertexTable, rowkey, Constants.PAGE_RANK_CQ_TMP_NAME);
    if (!pageRankEquals(oldPageRank, newPageRank, pageRankCompareScale)) {
      // collect pageRank changing count with counter
      context.getCounter(Counters.CHANGED_PAGE_RANK_COUNT).increment(1);
    }
    sw.stop();
    context.getCounter(Counters.CMP_OLD_NEW_PR_TIME_CONSUMED).increment(sw.getTime());

    context.write(key, new DoubleWritable(newPageRank));
  }

  static boolean pageRankEquals(double src, double dest, int scale) {
    BigDecimal a = new BigDecimal(src);
    BigDecimal b = new BigDecimal(dest);
    a = a.setScale(scale, RoundingMode.DOWN);
    b = b.setScale(scale, RoundingMode.DOWN);
    return a.compareTo(b) == 0 ? true : false;
  }

  /*
   * (non-Javadoc)
   * @see org.apache.hadoop.mapreduce.Reducer#setup(org.apache.hadoop.mapreduce.Reducer.Context)
   */
  @Override
  protected void setup(Context context) throws IOException, InterruptedException {
    Configuration conf = context.getConfiguration();

    vertexTable =
        Utils.initTable(conf, HBaseGraphConstants.HBASE_GRAPH_TABLE_VERTEX_NAME_KEY,