连通图

  2. import scala.reflect.ClassTag
  4. import org.apache.spark.graphx._
  6. /** Connected components algorithm. */
  7. object ConnectedComponents {
  8.   /**
  9.    * Compute the connected component membership of each vertex and return a graph with the vertex
  10.    * value containing the lowest vertex id in the connected component containing that vertex.
  11.    *
  12.    * @tparam VD the vertex attribute type (discarded in the computation)
  13.    * @tparam ED the edge attribute type (preserved in the computation)
  14.    * @param graph the graph for which to compute the connected components
  15.    * @param maxIterations the maximum number of iterations to run for
  16.    * @return a graph with vertex attributes containing the smallest vertex in each
  17.    *         connected component
  18.    */
  19.   def run[VD: ClassTag, ED: ClassTag](graph: Graph[VD, ED],
  20.                                       maxIterations: Int): Graph[VertexId, ED] = {
  21.     require(maxIterations > 0, s"Maximum of iterations must be greater than 0," +
  22.       s" but got ${maxIterations}")
  24.     val ccGraph = graph.mapVertices { case (vid, _) => vid }
  25.     def sendMessage(edge: EdgeTriplet[VertexId, ED]): Iterator[(VertexId, VertexId)] = {
  26.       if (edge.srcAttr < edge.dstAttr) {
  27.         Iterator((edge.dstId, edge.srcAttr))
  28.       } else if (edge.srcAttr > edge.dstAttr) {
  29.         Iterator((edge.srcId, edge.dstAttr))
  30.       } else {
  31.         Iterator.empty
  32.       }
  33.     }
  34.     val initialMessage = Long.MaxValue
  35.     val pregelGraph = Pregel(ccGraph, initialMessage,
  36.       maxIterations, EdgeDirection.Either)(
  37.       vprog = (id, attr, msg) => math.min(attr, msg),
  38.       sendMsg = sendMessage,
  39.       mergeMsg = (a, b) => math.min(a, b))
  40.     ccGraph.unpersist()
  41.     pregelGraph
  42.   } // end of connectedComponents
  44.   /**
  45.    * Compute the connected component membership of each vertex and return a graph with the vertex
  46.    * value containing the lowest vertex id in the connected component containing that vertex.
  47.    *
  48.    * @tparam VD the vertex attribute type (discarded in the computation)
  49.    * @tparam ED the edge attribute type (preserved in the computation)
  50.    * @param graph the graph for which to compute the connected components
  51.    * @return a graph with vertex attributes containing the smallest vertex in each
  52.    *         connected component
  53.    */
  54.   def run[VD: ClassTag, ED: ClassTag](graph: Graph[VD, ED]): Graph[VertexId, ED] = {
  55.     run(graph, Int.MaxValue)
  56.   }
  57. }