overload PageRank.runWithOptions and runWithOptionsWithPreviousPageRank with a 'normalized' parameter to trigger or not the normalization

Author: ebonnal

graphx/src/main/scala/org/apache/spark/graphx/lib/PageRank.scala

@@ -140,8 +140,31 @@ object PageRank extends Logging {
    */
   def runWithOptions[VD: ClassTag, ED: ClassTag](
       graph: Graph[VD, ED], numIter: Int, resetProb: Double = 0.15,
-      srcId: Option[VertexId] = None): Graph[Double, Double] =
-  {
+      srcId: Option[VertexId] = None): Graph[Double, Double] = {
+    runWithOptions(graph, numIter, resetProb, srcId, normalized = true)
+  }
+
+  /**
+   * Run PageRank for a fixed number of iterations returning a graph
+   * with vertex attributes containing the PageRank and edge
+   * attributes the normalized edge weight.
+   *
+   * @tparam VD the original vertex attribute (not used)
+   * @tparam ED the original edge attribute (not used)
+   *
+   * @param graph the graph on which to compute PageRank
+   * @param numIter the number of iterations of PageRank to run
+   * @param resetProb the random reset probability (alpha)
+   * @param srcId the source vertex for a Personalized Page Rank (optional)
+   * @param normalized whether or not to normalize rank sum
+   *
+   * @return the graph containing with each vertex containing the PageRank and each edge
+   *         containing the normalized weight.
+   *
+   */
+  def runWithOptions[VD: ClassTag, ED: ClassTag](
+      graph: Graph[VD, ED], numIter: Int, resetProb: Double,
+      srcId: Option[VertexId], normalized: Boolean): Graph[Double, Double] = {
     require(numIter > 0, s"Number of iterations must be greater than 0," +
       s" but got ${numIter}")
     require(resetProb >= 0 && resetProb <= 1, s"Random reset probability must belong" +
@@ -179,8 +202,13 @@ object PageRank extends Logging {
       iteration += 1
     }
 
-    // SPARK-18847 If the graph has sinks (vertices with no outgoing edges) correct the sum of ranks
-    normalizeRankSum(rankGraph, personalized)
+    if (normalized) {
+      // SPARK-18847 If the graph has sinks (vertices with no outgoing edges),
+      // correct the sum of ranks
+      normalizeRankSum(rankGraph, personalized)
+    } else {
+      rankGraph
+    }
   }
 
   /**
@@ -204,6 +232,32 @@ object PageRank extends Logging {
   def runWithOptionsWithPreviousPageRank[VD: ClassTag, ED: ClassTag](
       graph: Graph[VD, ED], numIter: Int, resetProb: Double, srcId: Option[VertexId],
       preRankGraph: Graph[Double, Double]): Graph[Double, Double] = {
+    runWithOptionsWithPreviousPageRank(
+      graph, numIter, resetProb, srcId, normalized = true, preRankGraph
+    )
+  }
+
+  /**
+   * Run PageRank for a fixed number of iterations returning a graph
+   * with vertex attributes containing the PageRank and edge
+   * attributes the normalized edge weight.
+   *
+   * @tparam VD the original vertex attribute (not used)
+   * @tparam ED the original edge attribute (not used)
+   *
+   * @param graph the graph on which to compute PageRank
+   * @param numIter the number of iterations of PageRank to run
+   * @param resetProb the random reset probability (alpha)
+   * @param srcId the source vertex for a Personalized Page Rank (optional)
+   * @param normalized whether or not to normalize rank sum
+   * @param preRankGraph PageRank graph from which to keep iterating
+   *
+   * @return the graph containing with each vertex containing the PageRank and each edge
+   *         containing the normalized weight.
+   */
+  def runWithOptionsWithPreviousPageRank[VD: ClassTag, ED: ClassTag](
+      graph: Graph[VD, ED], numIter: Int, resetProb: Double, srcId: Option[VertexId],
+      normalized: Boolean, preRankGraph: Graph[Double, Double]): Graph[Double, Double] = {
     require(numIter > 0, s"Number of iterations must be greater than 0," +
       s" but got ${numIter}")
     require(resetProb >= 0 && resetProb <= 1, s"Random reset probability must belong" +
@@ -238,8 +292,13 @@ object PageRank extends Logging {
       iteration += 1
     }
 
-    // SPARK-18847 If the graph has sinks (vertices with no outgoing edges) correct the sum of ranks
-    normalizeRankSum(rankGraph, personalized)
+    if (normalized) {
+      // SPARK-18847 If the graph has sinks (vertices with no outgoing edges),
+      // correct the sum of ranks
+      normalizeRankSum(rankGraph, personalized)
+    } else {
+      rankGraph
+    }
   }
 
   /**