3. 图算法

第 1 次迭代: 0.125, 0.333, 0.083, 0.458 
第 2 次迭代: 0.042, 0.500, 0.042, 0.417 
第 3 次迭代: 0.021, 0.431, 0.014, 0.535 
第 4 次迭代: 0.007, 0.542, 0.007, 0.444 
第 5 次迭代: 0.003, 0.447, 0.002, 0.547 
第 6 次迭代: 0.001, 0.549, 0.001, 0.449 
第 7 次迭代: 0.001, 0.449, 0.000, 0.550
第 8 次迭代: 0.000, 0.550, 0.000, 0.450 
第 9 次迭代: 0.000, 0.450, 0.000, 0.550
第 10次迭代: 0.000, 0.550, 0.000, 0.450

第 1 次迭代: 0.125, 0.583, 0.083, 0.208
第 2 次迭代: 0.042, 0.833, 0.042, 0.083 
第 3 次迭代: 0.021, 0.931, 0.014, 0.035 
第 4 次迭代: 0.007, 0.972, 0.007, 0.014 
第 5 次迭代: 0.003, 0.988, 0.002, 0.006 
第 6 次迭代: 0.001, 0.995, 0.001, 0.002 
第 7 次迭代: 0.001, 0.998, 0.000, 0.001 
第 8 次迭代: 0.000, 0.999, 0.000, 0.000 
第 9 次迭代: 0.000, 1.000, 0.000, 0.000 
第10次迭代: 0.000, 1.000, 0.000, 0.000 
... ...

第 1 次迭代: 0.125, 0.000, 0.125, 0.250 
第 2 次迭代: 0.063, 0.000, 0.063, 0.125 
第 3 次迭代: 0.031, 0.000, 0.031, 0.063 
第 4 次迭代: 0.016, 0.000, 0.016, 0.031 
第 5 次迭代: 0.008, 0.000, 0.008, 0.016 
第 6 次迭代: 0.004, 0.000, 0.004, 0.008 
第 7 次迭代: 0.002, 0.000, 0.002, 0.004
第 8 次迭代: 0.001, 0.000, 0.001, 0.002 
第 9 次迭代: 0.000, 0.000, 0.000, 0.001 
第10次迭代: 0.000, 0.000, 0.000, 0.000 
... ...

第 1 次迭代: 0.150, 0.317, 0.117, 0.417 
第 2 次迭代: 0.097, 0.423, 0.090, 0.390 
第 3 次迭代: 0.086, 0.388, 0.076, 0.450
第 4 次迭代: 0.080, 0.433, 0.073, 0.413 
第 5 次迭代: 0.079, 0.402, 0.071, 0.447 
第 6 次迭代: 0.079, 0.429, 0.071, 0.421 
第 7 次迭代: 0.078, 0.408, 0.071, 0.443 
第 8 次迭代: 0.078, 0.425, 0.071, 0.426 
第 9 次迭代: 0.078, 0.412, 0.071, 0.439 
第10次迭代: 0.078, 0.422, 0.071, 0.428 
第11次迭代: 0.078, 0.414, 0.071, 0.437 
第12次迭代: 0.078, 0.421, 0.071, 0.430 
第13次迭代: 0.078, 0.415, 0.071, 0.436 
第14次迭代: 0.078, 0.419, 0.071, 0.431 
第15次迭代: 0.078, 0.416, 0.071, 0.435 
第16次迭代: 0.078, 0.419, 0.071, 0.432 
第17次迭代: 0.078, 0.416, 0.071, 0.434 
第18次迭代: 0.078, 0.418, 0.071, 0.432 
第19次迭代: 0.078, 0.417, 0.071, 0.434
第20次迭代: 0.078, 0.418, 0.071, 0.433 
... ...

import org.apache.spark.graphx.GraphLoader

// Load the edges as a graph
val graph = GraphLoader.edgeListFile(sc, "data/graphx/followers.txt")
// Run PageRank
val ranks = graph.pageRank(0.0001).vertices
// Join the ranks with the usernames
val users = sc.textFile("data/graphx/users.txt").map { line =>
  val fields = line.split(",")
  (fields(0).toLong, fields(1))
}
val ranksByUsername = users.join(ranks).map {
  case (id, (username, rank)) => (username, rank)
}
// Print the result
println(ranksByUsername.collect().mkString("\n"))

val graph = GraphLoader.edgeListFile(sc, "graphx/data/test_graph.txt")
val root: VertexId = 1
val initialGraph = graph.mapVertices((id, _) => if (id == root) 0.0 else Double.PositiveInfinity)
val vprog = { (id: VertexId, attr: Double, msg: Double) => math.min(attr,msg) }
val sendMessage = { (triplet: EdgeTriplet[Double, Int]) =>
  var iter:Iterator[(VertexId, Double)] = Iterator.empty
  val isSrcMarked = triplet.srcAttr != Double.PositiveInfinity
  val isDstMarked = triplet.dstAttr != Double.PositiveInfinity
  if(!(isSrcMarked && isDstMarked)){
    if(isSrcMarked){
      iter = Iterator((triplet.dstId,triplet.srcAttr+1))
    }else{
      iter = Iterator((triplet.srcId,triplet.dstAttr+1))
    }
  }
  iter
}
val reduceMessage = { (a: Double, b: Double) => math.min(a,b) }
val bfs = initialGraph.pregel(Double.PositiveInfinity, 20)(vprog, sendMessage, reduceMessage) 
println(bfs.vertices.collect.mkString("\n"))

import scala.reflect.ClassTag
import org.apache.spark.graphx._
/**
  * Computes shortest paths to the given set of landmark vertices, returning a
graph where each
  * vertex attribute is a map containing the shortest-path distance to each
reachable landmark. */
object ShortestPaths {
  /** Stores a map from the vertex id of a landmark to the distance to that
landmark. */
  type SPMap = Map[VertexId, Int]
  private def makeMap(x: (VertexId, Int)*) = Map(x: _*)
  private def incrementMap(spmap: SPMap): SPMap = spmap.map { case (v, d) => v -> (d + 1) }
  private def addMaps(spmap1: SPMap, spmap2: SPMap): SPMap = (spmap1.keySet ++ spmap2.keySet).map {
    k => k -> math.min(spmap1.getOrElse(k, Int.MaxValue), spmap2.getOrElse(k, Int.MaxValue))
  }.toMap
/**
  * Computes shortest paths to the given set of landmark vertices. *
  * @tparam ED the edge attribute type (not used in the computation) *
  * @param graph the graph for which to compute the shortest paths
  * @param landmarks the list of landmark vertex ids. Shortest paths will be
computed to each
  * landmark.
  *
  * @return a graph where each vertex attribute is a map containing the
shortest-path distance to
  * each reachable landmark vertex.
  */
  def run[VD, ED: ClassTag](graph: Graph[VD, ED], landmarks: Seq[VertexId]):
  Graph[SPMap, ED] = {
    val spGraph = graph.mapVertices { (vid, attr) =>
      if (landmarks.contains(vid)) makeMap(vid -> 0) else makeMap() 
    }
    val initialMessage = makeMap()
    def vertexProgram(id: VertexId, attr: SPMap, msg: SPMap): SPMap = { addMaps(attr, msg)
    }
    def sendMessage(edge: EdgeTriplet[SPMap, _]): Iterator[(VertexId, SPMap)] = {
      val newAttr = incrementMap(edge.dstAttr)
      if (edge.srcAttr != addMaps(newAttr, edge.srcAttr)) Iterator((edge.srcId, newAttr))
      else Iterator.empty
    }
    Pregel(spGraph, initialMessage)(vertexProgram, sendMessage, addMaps)
  }
}

import scala.reflect.ClassTag
import org.apache.spark.graphx._
/** Connected components algorithm. */
object ConnectedComponents {
  /**
  * Compute the connected component membership of each vertex and return a graph with the vertex
  * value containing the lowest vertex id in the connected component containing that vertex.
  * @tparam VD the vertex attribute type (discarded in the computation)
  * @tparam ED the edge attribute type (preserved in the computation)
  * @param graph the graph for which to compute the connected components
  * @param maxIterations the maximum number of iterations to run for
  * @return a graph with vertex attributes containing the smallest vertex in each connected component
  */

  def run[VD: ClassTag, ED: ClassTag](graph: Graph[VD, ED],
                                      maxIterations: Int): Graph[VertexId, ED] = {
    require(maxIterations > 0, s"Maximum of iterations must be greater than 0," +
      s" but got ${maxIterations}")
    val ccGraph = graph.mapVertices { case (vid, _) => vid }
    def sendMessage(edge: EdgeTriplet[VertexId, ED]): Iterator[(VertexId, VertexId)] = {
      if (edge.srcAttr < edge.dstAttr) { Iterator((edge.dstId, edge.srcAttr))
      } else if (edge.srcAttr > edge.dstAttr) { Iterator((edge.srcId, edge.dstAttr))
      } else { Iterator.empty
      }
    }
    val initialMessage = Long.MaxValue
    val pregelGraph = Pregel(ccGraph, initialMessage,
      maxIterations, EdgeDirection.Either)(
      vprog = (id, attr, msg) => math.min(attr, msg),
      sendMsg = sendMessage,
      mergeMsg = (a, b) => math.min(a, b))
    ccGraph.unpersist()
    pregelGraph
  } // end of connectedComponents
/**
  * Compute the connected component membership of each vertex and return a
graph with the vertex
  * value containing the lowest vertex id in the connected component
containing that vertex.
  *
  * @tparam VD the vertex attribute type (discarded in the computation) 
  * @tparam ED the edge attribute type (preserved in the computation)
  * @param graph the graph for which to compute the connected components
  * @return a graph with vertex attributes containing the smallest vertex in eac connected component
  */
def run[VD: ClassTag, ED: ClassTag](graph: Graph[VD, ED]): Graph[VertexId, ED] = {
  run(graph, Int.MaxValue) }
}

import scala.reflect.ClassTag
import org.apache.spark.graphx._
/**
  * Compute the number of triangles passing through each vertex.
  *
  * The algorithm is relatively straightforward and can be computed in three
steps:
  *
  * <ul>
  * <li> Compute the set of neighbors for each vertex.</li>
  * <li> For each edge compute the intersection of the sets and send the count to both vertices.</li>
  * <li> Compute the sum at each vertex and divide by two since each triangle is counted twice.</li>
  * </ul>
  *
  * There are two implementations. The default `TriangleCount.run`
implementation first removes
  * self cycles and canonicalizes the graph to ensure that the following
conditions hold:
  *<ul>
  * <li> There are no self edges </li>
  * <li> All edges are oriented src > dst </li>
  * <li> There are no duplicate edges </li>
  * </ul>
  * However, the canonicalization procedure is costly as it requires repartitioning the graph.
  * If the input data is already in "canonical form" with self cycles removed then the
  * `TriangleCount.runPreCanonicalized` should be used instead.
  * {{{
  * val canonicalGraph = graph.mapEdges(e =>
1).removeSelfEdges().canonicalizeEdges()
  * val counts = TriangleCount.runPreCanonicalized(canonicalGraph).vertices
  * }}}
  *
  */
object TriangleCount {
  def run[VD: ClassTag, ED: ClassTag](graph: Graph[VD, ED]): Graph[Int, ED] = { // Transform the edge data something cheap to shuffle and then canonicalize
    val canonicalGraph = graph.mapEdges(e =>
    true).removeSelfEdges().convertToCanonicalEdges()
    // Get the triangle counts
    val counters = runPreCanonicalized(canonicalGraph).vertices
    // Join them bath with the original graph
    graph.outerJoinVertices(counters) { (vid, _, optCounter: Option[Int]) =>
    optCounter.getOrElse(0)
    }
  }

  def runPreCanonicalized[VD: ClassTag, ED: ClassTag](graph: Graph[VD, ED]): Graph[Int, ED] = {
    // Construct set representations of the neighborhoods
    val nbrSets: VertexRDD[VertexSet] =
      graph.collectNeighborIds(EdgeDirection.Either).mapValues { (vid, nbrs) =>
        val set = new VertexSet(nbrs.length)
        var i = 0
        while (i < nbrs.length) {
          // prevent self cycle
          if (nbrs(i) != vid) {
            set.add(nbrs(i))
          }
          i += 1
        }
        set
      }
    // join the sets with the graph
    val setGraph: Graph[VertexSet, ED] = graph.outerJoinVertices(nbrSets) {
      (vid, _, optSet) => optSet.getOrElse(null)
    }

    // Edge function computes intersection of smaller vertex with larger vertex
    def edgeFunc(ctx: EdgeContext[VertexSet, ED, Int]) {
      val (smallSet, largeSet) =
        if (ctx.srcAttr.size < ctx.dstAttr.size) {
          (ctx.srcAttr, ctx.dstAttr)
        } else {
          (ctx.dstAttr, ctx.srcAttr)
        }
      val iter = smallSet.iterator
      var counter: Int = 0
      while (iter.hasNext) {
        val vid = iter.next()
        if (vid != ctx.srcId && vid != ctx.dstId && largeSet.contains(vid)) {
          counter += 1
        }
      }
      ctx.sendToSrc(counter)
      ctx.sendToDst(counter)
    }

    // compute the intersection along edges
    val counters: VertexRDD[Int] = setGraph.aggregateMessages(edgeFunc, _ + _)
    // Merge counters with the graph and divide by two since each triangle is counted twice
    graph.outerJoinVertices(counters) { (_, _, optCounter: Option[Int]) =>
      val dblCount = optCounter.getOrElse(0)
      // This algorithm double counts each triangle so the final count should be even
      require(dblCount % 2 == 0, "Triangle count resulted in an invalid number of triangles.")
      dblCount / 2
    } 
  }
}