// Path: src/main/java/org/trend/hgraph/mapreduce/pagerank/CalculateInitPageRankMapper.java
// static void dispatchPageRank(List<String> outgoingRowKeys, double pageRank, Configuration conf,
//     HTable edgeTable, Counter dispatchPrTimeConsumeCounter, Counter outgoingEdgeCounter,
//     ContextWriterStrategy strategy)
//     throws IOException,
//     InterruptedException {
//   int outgoingEdgeCount = outgoingRowKeys.size();
//   outgoingEdgeCounter.increment(outgoingEdgeCount);
//   double pageRankForEachOutgoing = pageRank / (double) outgoingEdgeCount;
// 
//   StopWatch sw = null;
//   String outgoingRowKey = null;
//   try {
//     sw = new StopWatch();
//     sw.start();
//     for (int a = 0; a < outgoingRowKeys.size(); a++) {
//       outgoingRowKey = outgoingRowKeys.get(a);
//       strategy.write(outgoingRowKey, pageRankForEachOutgoing);
//     }
//     sw.stop();
//     dispatchPrTimeConsumeCounter.increment(sw.getTime());
//   } catch (IOException e) {
//     System.err.println("failed while writing outgoingRowKey:" + outgoingRowKey);
//     e.printStackTrace(System.err);
//     throw e;
//   } catch (InterruptedException e) {
//     System.err.println("failed while writing outgoingRowKey:" + outgoingRowKey);
//     e.printStackTrace(System.err);
//     throw e;
//   }
// }
// 
// Path: src/main/java/org/trend/hgraph/mapreduce/pagerank/CalculateInitPageRankMapper.java
// static List<String> getOutgoingRowKeys(Configuration conf, HTable vertexTable, HTable edgeTable,
//     String rowKey, Counter counter) throws IOException {
//   ResultScanner rs = null;
//   String key = null;
//   LinkedList<String> rowKeys = new LinkedList<String>();
//   StopWatch sw = null;
//   // Put put = null;
//   try {
//     Scan scan = getRowKeyOnlyScan(rowKey);
//     sw = new StopWatch();
//     sw.start();
//     rs = edgeTable.getScanner(scan);
//     for (Result r : rs) {
//       key = getOutgoingRowKey(r);
//       // collect outgoing rowkeys
//       rowKeys.add(key);
//     }
//     sw.stop();
//     counter.increment(sw.getTime());
//   } catch (IOException e) {
//     System.err.println("access htable:" + Bytes.toString(edgeTable.getTableName()) + " failed");
//     e.printStackTrace(System.err);
//     throw e;
//   } finally {
//     rs.close();
//   }
//   return rowKeys;
// }
// 
// 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/CalculateInitPageRankMapper.java
// interface ContextWriterStrategy {
//   void write(String key, double value) throws IOException, InterruptedException;
// }


import org.trend.hgraph.mapreduce.pagerank.CalculateInitPageRankMapper.ContextWriterStrategy;
import static org.trend.hgraph.mapreduce.pagerank.CalculateInitPageRankMapper.dispatchPageRank;
import static org.trend.hgraph.mapreduce.pagerank.CalculateInitPageRankMapper.getOutgoingRowKeys;
import java.io.IOException;
import java.util.List;
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.Mapper;
import org.trend.hgraph.HBaseGraphConstants;
/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.trend.hgraph.mapreduce.pagerank;




/**
 * A <code>Mapper</code> for calculating intermediate pagerank value from HDFS.
 * @author scott_miao
 */
public class CalculateIntermediatePageRankMapper extends
    Mapper<Text, DoubleWritable, Text, DoubleWritable> {

  private HTable edgeTable = null;
  private HTable vertexTable = null;
  
  private String tmpPageRankCq = Constants.PAGE_RANK_CQ_TMP_NAME;

  enum Counters {
    VERTEX_COUNT, OUTGOING_EDGE_COUNT, GET_OUTGOING_VERTICES_TIME_CONSUMED,
    DISPATCH_PR_TIME_CONSUMED
  }
  /*
   * (non-Javadoc)
   * @see org.apache.hadoop.mapreduce.Mapper#map(java.lang.Object, java.lang.Object, Context)
   */
  @Override
  protected void map(final Text key, final DoubleWritable value, final Context context)
      throws IOException, InterruptedException {
    String rowKey = Bytes.toString(key.getBytes()).trim();
    double pageRank = value.get();
    // write current pageRank to tmp
    Utils.writePageRank(vertexTable, rowKey, tmpPageRankCq, pageRank);
    
    Configuration conf = context.getConfiguration();
    List<String> outgoingRowKeys = null;

    context.getCounter(Counters.VERTEX_COUNT).increment(1);
    outgoingRowKeys =
// Path: src/main/java/org/trend/hgraph/mapreduce/pagerank/CalculateInitPageRankMapper.java
// static void dispatchPageRank(List<String> outgoingRowKeys, double pageRank, Configuration conf,
//     HTable edgeTable, Counter dispatchPrTimeConsumeCounter, Counter outgoingEdgeCounter,
//     ContextWriterStrategy strategy)
//     throws IOException,
//     InterruptedException {
//   int outgoingEdgeCount = outgoingRowKeys.size();
//   outgoingEdgeCounter.increment(outgoingEdgeCount);
//   double pageRankForEachOutgoing = pageRank / (double) outgoingEdgeCount;
// 
//   StopWatch sw = null;
//   String outgoingRowKey = null;
//   try {
//     sw = new StopWatch();
//     sw.start();
//     for (int a = 0; a < outgoingRowKeys.size(); a++) {
//       outgoingRowKey = outgoingRowKeys.get(a);
//       strategy.write(outgoingRowKey, pageRankForEachOutgoing);
//     }
//     sw.stop();
//     dispatchPrTimeConsumeCounter.increment(sw.getTime());
//   } catch (IOException e) {
//     System.err.println("failed while writing outgoingRowKey:" + outgoingRowKey);
//     e.printStackTrace(System.err);
//     throw e;
//   } catch (InterruptedException e) {
//     System.err.println("failed while writing outgoingRowKey:" + outgoingRowKey);
//     e.printStackTrace(System.err);
//     throw e;
//   }
// }
// 
// Path: src/main/java/org/trend/hgraph/mapreduce/pagerank/CalculateInitPageRankMapper.java
// static List<String> getOutgoingRowKeys(Configuration conf, HTable vertexTable, HTable edgeTable,
//     String rowKey, Counter counter) throws IOException {
//   ResultScanner rs = null;
//   String key = null;
//   LinkedList<String> rowKeys = new LinkedList<String>();
//   StopWatch sw = null;
//   // Put put = null;
//   try {
//     Scan scan = getRowKeyOnlyScan(rowKey);
//     sw = new StopWatch();
//     sw.start();
//     rs = edgeTable.getScanner(scan);
//     for (Result r : rs) {
//       key = getOutgoingRowKey(r);
//       // collect outgoing rowkeys
//       rowKeys.add(key);
//     }
//     sw.stop();
//     counter.increment(sw.getTime());
//   } catch (IOException e) {
//     System.err.println("access htable:" + Bytes.toString(edgeTable.getTableName()) + " failed");
//     e.printStackTrace(System.err);
//     throw e;
//   } finally {
//     rs.close();
//   }
//   return rowKeys;
// }
// 
// 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/CalculateInitPageRankMapper.java
// interface ContextWriterStrategy {
//   void write(String key, double value) throws IOException, InterruptedException;
// }


// Path: src/main/java/org/trend/hgraph/mapreduce/pagerank/CalculateIntermediatePageRankMapper.java
import org.trend.hgraph.mapreduce.pagerank.CalculateInitPageRankMapper.ContextWriterStrategy;
import static org.trend.hgraph.mapreduce.pagerank.CalculateInitPageRankMapper.dispatchPageRank;
import static org.trend.hgraph.mapreduce.pagerank.CalculateInitPageRankMapper.getOutgoingRowKeys;
import java.io.IOException;
import java.util.List;
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.Mapper;
import org.trend.hgraph.HBaseGraphConstants;

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.trend.hgraph.mapreduce.pagerank;




/**
 * A <code>Mapper</code> for calculating intermediate pagerank value from HDFS.
 * @author scott_miao
 */
public class CalculateIntermediatePageRankMapper extends
    Mapper<Text, DoubleWritable, Text, DoubleWritable> {

  private HTable edgeTable = null;
  private HTable vertexTable = null;
  
  private String tmpPageRankCq = Constants.PAGE_RANK_CQ_TMP_NAME;

  enum Counters {
    VERTEX_COUNT, OUTGOING_EDGE_COUNT, GET_OUTGOING_VERTICES_TIME_CONSUMED,
    DISPATCH_PR_TIME_CONSUMED
  }
  /*
   * (non-Javadoc)
   * @see org.apache.hadoop.mapreduce.Mapper#map(java.lang.Object, java.lang.Object, Context)
   */
  @Override
  protected void map(final Text key, final DoubleWritable value, final Context context)
      throws IOException, InterruptedException {
    String rowKey = Bytes.toString(key.getBytes()).trim();
    double pageRank = value.get();
    // write current pageRank to tmp
    Utils.writePageRank(vertexTable, rowKey, tmpPageRankCq, pageRank);
    
    Configuration conf = context.getConfiguration();
    List<String> outgoingRowKeys = null;

    context.getCounter(Counters.VERTEX_COUNT).increment(1);
    outgoingRowKeys =
        getOutgoingRowKeys(conf, vertexTable, edgeTable, rowKey,