PageRank

  1. import scala.language.postfixOps
  2. import scala.reflect.ClassTag
  4. import org.apache.spark.graphx._
  5. import org.apache.spark.internal.Logging
  7. /**
  8.  * PageRank algorithm implementation. There are two implementations of PageRank implemented.
  9.  *
  10.  * The first implementation uses the standalone [[Graph]] interface and runs PageRank
  11.  * for a fixed number of iterations:
  12.  * {{{
  13.  * var PR = Array.fill(n)( 1.0 )
  14.  * val oldPR = Array.fill(n)( 1.0 )
  15.  * for( iter <- 0 until numIter ) {
  16.  *   swap(oldPR, PR)
  17.  *   for( i <- 0 until n ) {
  18.  *     PR[i] = alpha + (1 - alpha) * inNbrs[i].map(j => oldPR[j] / outDeg[j]).sum
  19.  *   }
  20.  * }
  21.  * }}}
  22.  *
  23.  * The second implementation uses the [[Pregel]] interface and runs PageRank until
  24.  * convergence:
  25.  *
  26.  * {{{
  27.  * var PR = Array.fill(n)( 1.0 )
  28.  * val oldPR = Array.fill(n)( 0.0 )
  29.  * while( max(abs(PR - oldPr)) > tol ) {
  30.  *   swap(oldPR, PR)
  31.  *   for( i <- 0 until n if abs(PR[i] - oldPR[i]) > tol ) {
  32.  *     PR[i] = alpha + (1 - \alpha) * inNbrs[i].map(j => oldPR[j] / outDeg[j]).sum
  33.  *   }
  34.  * }
  35.  * }}}
  36.  *
  37.  * `alpha` is the random reset probability (typically 0.15), `inNbrs[i]` is the set of
  38.  * neighbors which link to `i` and `outDeg[j]` is the out degree of vertex `j`.
  39.  *
  40.  * Note that this is not the "normalized" PageRank and as a consequence pages that have no
  41.  * inlinks will have a PageRank of alpha.
  42.  */
  43. object PageRank extends Logging {
  46.   /**
  47.    * Run PageRank for a fixed number of iterations returning a graph
  48.    * with vertex attributes containing the PageRank and edge
  49.    * attributes the normalized edge weight.
  50.    *
  51.    * @tparam VD the original vertex attribute (not used)
  52.    * @tparam ED the original edge attribute (not used)
  53.    *
  54.    * @param graph the graph on which to compute PageRank
  55.    * @param numIter the number of iterations of PageRank to run
  56.    * @param resetProb the random reset probability (alpha)
  57.    *
  58.    * @return the graph containing with each vertex containing the PageRank and each edge
  59.    *         containing the normalized weight.
  60.    */
  61.   def run[VD: ClassTag, ED: ClassTag](graph: Graph[VD, ED], numIter: Int,
  62.     resetProb: Double = 0.15): Graph[Double, Double] =
  63.   {
  64.     runWithOptions(graph, numIter, resetProb)
  65.   }
  67.   /**
  68.    * Run PageRank for a fixed number of iterations returning a graph
  69.    * with vertex attributes containing the PageRank and edge
  70.    * attributes the normalized edge weight.
  71.    *
  72.    * @tparam VD the original vertex attribute (not used)
  73.    * @tparam ED the original edge attribute (not used)
  74.    *
  75.    * @param graph the graph on which to compute PageRank
  76.    * @param numIter the number of iterations of PageRank to run
  77.    * @param resetProb the random reset probability (alpha)
  78.    * @param srcId the source vertex for a Personalized Page Rank (optional)
  79.    *
  80.    * @return the graph containing with each vertex containing the PageRank and each edge
  81.    *         containing the normalized weight.
  82.    *
  83.    */
  84.   def runWithOptions[VD: ClassTag, ED: ClassTag](
  85.       graph: Graph[VD, ED], numIter: Int, resetProb: Double = 0.15,
  86.       srcId: Option[VertexId] = None): Graph[Double, Double] =
  87.   {
  88.     require(numIter > 0, s"Number of iterations must be greater than 0," +
  89.       s" but got ${numIter}")
  90.     require(resetProb >= 0 && resetProb <= 1, s"Random reset probability must belong" +
  91.       s" to [0, 1], but got ${resetProb}")
  93.     val personalized = srcId isDefined
  94.     val src: VertexId = srcId.getOrElse(-1L)
  96.     // Initialize the PageRank graph with each edge attribute having
  97.     // weight 1/outDegree and each vertex with attribute resetProb.
  98.     // When running personalized pagerank, only the source vertex
  99.     // has an attribute resetProb. All others are set to 0.
  100.     var rankGraph: Graph[Double, Double] = graph
  101.       // Associate the degree with each vertex
  102.       .outerJoinVertices(graph.outDegrees) { (vid, vdata, deg) => deg.getOrElse(0) }
  103.       // Set the weight on the edges based on the degree
  104.       .mapTriplets( e => 1.0 / e.srcAttr, TripletFields.Src )
  105.       // Set the vertex attributes to the initial pagerank values
  106.       .mapVertices { (id, attr) =>
  107.         if (!(id != src && personalized)) resetProb else 0.0
  108.       }
  110.     def delta(u: VertexId, v: VertexId): Double = { if (u == v) 1.0 else 0.0 }
  112.     var iteration = 0
  113.     var prevRankGraph: Graph[Double, Double] = null
  114.     while (iteration < numIter) {
  115.       rankGraph.cache()
  117.       // Compute the outgoing rank contributions of each vertex, perform local preaggregation, and
  118.       // do the final aggregation at the receiving vertices. Requires a shuffle for aggregation.
  119.       val rankUpdates = rankGraph.aggregateMessages[Double](
  120.         ctx => ctx.sendToDst(ctx.srcAttr * ctx.attr), _ + _, TripletFields.Src)
  122.       // Apply the final rank updates to get the new ranks, using join to preserve ranks of vertices
  123.       // that didn't receive a message. Requires a shuffle for broadcasting updated ranks to the
  124.       // edge partitions.
  125.       prevRankGraph = rankGraph
  126.       val rPrb = if (personalized) {
  127.         (src: VertexId, id: VertexId) => resetProb * delta(src, id)
  128.       } else {
  129.         (src: VertexId, id: VertexId) => resetProb
  130.       }
  132.       rankGraph = rankGraph.joinVertices(rankUpdates) {
  133.         (id, oldRank, msgSum) => rPrb(src, id) + (1.0 - resetProb) * msgSum
  134.       }.cache()
  136.       rankGraph.edges.foreachPartition(x => {}) // also materializes rankGraph.vertices
  137.       logInfo(s"PageRank finished iteration $iteration.")
  138.       prevRankGraph.vertices.unpersist(false)
  139.       prevRankGraph.edges.unpersist(false)
  141.       iteration += 1
  142.     }
  144.     rankGraph
  145.   }
  147.   /**
  148.    * Run a dynamic version of PageRank returning a graph with vertex attributes containing the
  149.    * PageRank and edge attributes containing the normalized edge weight.
  150.    *
  151.    * @tparam VD the original vertex attribute (not used)
  152.    * @tparam ED the original edge attribute (not used)
  153.    *
  154.    * @param graph the graph on which to compute PageRank
  155.    * @param tol the tolerance allowed at convergence (smaller => more accurate).
  156.    * @param resetProb the random reset probability (alpha)
  157.    *
  158.    * @return the graph containing with each vertex containing the PageRank and each edge
  159.    *         containing the normalized weight.
  160.    */
  161.   def runUntilConvergence[VD: ClassTag, ED: ClassTag](
  162.     graph: Graph[VD, ED], tol: Double, resetProb: Double = 0.15): Graph[Double, Double] =
  163.   {
  164.       runUntilConvergenceWithOptions(graph, tol, resetProb)
  165.   }
  167.   /**
  168.    * Run a dynamic version of PageRank returning a graph with vertex attributes containing the
  169.    * PageRank and edge attributes containing the normalized edge weight.
  170.    *
  171.    * @tparam VD the original vertex attribute (not used)
  172.    * @tparam ED the original edge attribute (not used)
  173.    *
  174.    * @param graph the graph on which to compute PageRank
  175.    * @param tol the tolerance allowed at convergence (smaller => more accurate).
  176.    * @param resetProb the random reset probability (alpha)
  177.    * @param srcId the source vertex for a Personalized Page Rank (optional)
  178.    *
  179.    * @return the graph containing with each vertex containing the PageRank and each edge
  180.    *         containing the normalized weight.
  181.    */
  182.   def runUntilConvergenceWithOptions[VD: ClassTag, ED: ClassTag](
  183.       graph: Graph[VD, ED], tol: Double, resetProb: Double = 0.15,
  184.       srcId: Option[VertexId] = None): Graph[Double, Double] =
  185.   {
  186.     require(tol >= 0, s"Tolerance must be no less than 0, but got ${tol}")
  187.     require(resetProb >= 0 && resetProb <= 1, s"Random reset probability must belong" +
  188.       s" to [0, 1], but got ${resetProb}")
  190.     val personalized = srcId.isDefined
  191.     val src: VertexId = srcId.getOrElse(-1L)
  193.     // Initialize the pagerankGraph with each edge attribute
  194.     // having weight 1/outDegree and each vertex with attribute 1.0.
  195.     val pagerankGraph: Graph[(Double, Double), Double] = graph
  196.       // Associate the degree with each vertex
  197.       .outerJoinVertices(graph.outDegrees) {
  198.         (vid, vdata, deg) => deg.getOrElse(0)
  199.       }
  200.       // Set the weight on the edges based on the degree
  201.       .mapTriplets( e => 1.0 / e.srcAttr )
  202.       // Set the vertex attributes to (initialPR, delta = 0)
  203.       .mapVertices { (id, attr) =>
  204.         if (id == src) (resetProb, Double.NegativeInfinity) else (0.0, 0.0)
  205.       }
  206.       .cache()
  208.     // Define the three functions needed to implement PageRank in the GraphX
  209.     // version of Pregel
  210.     def vertexProgram(id: VertexId, attr: (Double, Double), msgSum: Double): (Double, Double) = {
  211.       val (oldPR, lastDelta) = attr
  212.       val newPR = oldPR + (1.0 - resetProb) * msgSum
  213.       (newPR, newPR - oldPR)
  214.     }
  216.     def personalizedVertexProgram(id: VertexId, attr: (Double, Double),
  217.       msgSum: Double): (Double, Double) = {
  218.       val (oldPR, lastDelta) = attr
  219.       var teleport = oldPR
  220.       val delta = if (src==id) 1.0 else 0.0
  221.       teleport = oldPR*delta
  223.       val newPR = teleport + (1.0 - resetProb) * msgSum
  224.       val newDelta = if (lastDelta == Double.NegativeInfinity) newPR else newPR - oldPR
  225.       (newPR, newDelta)
  226.     }
  228.     def sendMessage(edge: EdgeTriplet[(Double, Double), Double]) = {
  229.       if (edge.srcAttr._2 > tol) {
  230.         Iterator((edge.dstId, edge.srcAttr._2 * edge.attr))
  231.       } else {
  232.         Iterator.empty
  233.       }
  234.     }
  236.     def messageCombiner(a: Double, b: Double): Double = a + b
  238.     // The initial message received by all vertices in PageRank
  239.     val initialMessage = if (personalized) 0.0 else resetProb / (1.0 - resetProb)
  241.     // Execute a dynamic version of Pregel.
  242.     val vp = if (personalized) {
  243.       (id: VertexId, attr: (Double, Double), msgSum: Double) =>
  244.         personalizedVertexProgram(id, attr, msgSum)
  245.     } else {
  246.       (id: VertexId, attr: (Double, Double), msgSum: Double) =>
  247.         vertexProgram(id, attr, msgSum)
  248.     }
  250.     Pregel(pagerankGraph, initialMessage, activeDirection = EdgeDirection.Out)(
  251.       vp, sendMessage, messageCombiner)
  252.       .mapVertices((vid, attr) => attr._1)
  253.   } // end of deltaPageRank
  255. }