强连通体的GraphX的实现和解析

Posted 2022-11-30 ZL小屁孩

/**连通体*/
object ConnectComponents 

  /**在图中若从某个顶点Vp出发，沿着一些边经过顶点V1，V2，。。。Vm到达Vg则称顶点序列(Vp,V1,V2....Vm，Vg)为从Vp到Vg的路径，其中Vp是路径的起始点，Vg为路径终点。
   * 路径上的边的数目成为该路径的长度
   * 连通：在无向图中，若从顶点Vi到顶点Vj之间有路径称为这两个顶点是连通的。
   * 连通图：若图中任意一对顶点之间都是连通的，则称为此图为连通图。
   * 连通分量：非连通图中的每个连通部分称为连通分量
   * 强连通：对于有向图，若从顶点Vi到顶点Vj到顶点Vi之间都有路径，则称这两顶点是强连通的。
   * 强连通图：若有向图中任何一对顶点都是强连通的，则此图为强连通图。
   *
   * 性质：
   * 无向连通图，则边E的数目大于等于顶点V的数目减1：|E| >= |V| - 1，反之不成立；
   * 有向强连通图的必要条件是边的数目大于等于顶点的数目：|E| >= |V|，反之不成立。
   * 没有回路的无向图是连通的当且它是树，即等价于：|E| = |V| - 1
   * */
  def conCompts(sc: SparkContext): Unit =

    val graph = Graph(sc.makeRDD((1L to 7L).map((_,""))),
      sc.makeRDD(Array(Edge(2L,5L,"配偶"),Edge(5L,3L,"亲戚"),Edge(3L,2L,"子女"),Edge(4L,5L,"父母"),Edge(6L,7L,"亲戚")))
    )

    graph
      .connectedComponents()
      .vertices
      .map(_.swap)
      .groupByKey().map(_._2)
      .foreach(println)
  

  /**强连通：
   * （1）对图中所有节点设定初始连通分支id，用自己的节点id作为所属连通分支的id，并将所有节点打上初始标记false；
   * （2）首先做循环，将只有出边或入边的节点标记为true，将 **只存在单向边的或者孤立的节点** 和 **已经
   * 确认且打好标记的强连通分量中的节点**（即被标记为true的节点）从图中去除；
   * （3）为图中节点正向着色，先用节点id为自身着色，之后沿着出边向邻居节点发送自己的着色id（只有较小的着色id向较大的着色id的节点发送消息）。
   * （4）为着色完成的图中节点反向打标签（是否完成连通分支id标记）。在着色完成的图中，节点id与节点所在连通分支id相同时表明
   * 该节点是着色的root节点，标记为true。若一个节点对应的入边的另外一个节点是true，则该节点也被标记为true。
   * 节点沿着入边由src节点向dst节点发送自身标记情况，只有收到true的消息则节点便标记为true。（只有着色相同，
   * 且一条边上dst节点——发消息者是true但是src节点——收消息者是false时，dst节点才会向src节点发送消息）*/
  def strongConCompt(sc: SparkContext): Unit =
    val graph = Graph(sc.makeRDD((1L to 7L).map((_,""))),
      sc.makeRDD(Array(Edge(2L,5L,"配偶"),Edge(5L,3L,"亲戚"),Edge(3L,2L,"子女"),Edge(4L,5L,"父母"),Edge(6L,7L,"亲戚")))
    )
    val strongConCpt: Graph[VertexId, String] = graph.stronglyConnectedComponents(3)
    strongConCpt.vertices.foreach(println)
  

  /**StronglyConnectedComponents源码*/
  def strongCompt[VD: ClassTag, ED: ClassTag](graph: Graph[VD, ED], numIter: Int): Graph[VertexId, ED] = 

    // 初始化图，将节点id作为节点属性，sccGraph是最后的返回结果图
    var sccGraph = graph.mapVertices  case (vid, _) => vid 
    // 在迭代中使用的图
    var sccWorkGraph = graph.mapVertices  case (vid, _) => (vid, false) .cache()

    // 辅助变量prevSccGraph，用来unpersist缓存图
    var prevSccGraph = sccGraph

    var numVertices = sccWorkGraph.numVertices
    var iter = 0
    while (sccWorkGraph.numVertices > 0 && iter < numIter) 
      iter += 1
      // 此处循环内部工作：
      // 1.第一次循环进入时：将sccWorkGrpah图中只有单向边的节点或者孤立节点去掉； 后面循环进入时：将sccWorkGraph图中已经标识完成的强连通分量去掉。
      // 2.更新图中节点所属的强连通分支id
      // 只有在第一次进入第一层循环时，第一层循环内部的do-while循环才会循环多次，第2次以上只会只运行一次do的内容，因为后面图中不存在单向节点了。
      do 
        numVertices = sccWorkGraph.numVertices
        sccWorkGraph = sccWorkGraph.outerJoinVertices(sccWorkGraph.outDegrees) 
          (vid, data, degreeOpt) => if (degreeOpt.isDefined) data else (vid, true)
        .outerJoinVertices(sccWorkGraph.inDegrees) 
          (vid, data, degreeOpt) => if (degreeOpt.isDefined) data else (vid, true)
        .cache() //得到图中的有双向边的节点（vid，false）， 单向边或者孤立节点（vid，true），并且已经成功标记完连通分支的节点自身属性便是（vid，true）

        // 拿到图中只有单向边的节点或孤立节点
        val finalVertices = sccWorkGraph.vertices
          .filter  case (vid, (scc, isFinal)) => isFinal
          .mapValues  (vid, data) => data._1


        // //外部第一次循环不会变动sccGraph节点的属性，只有在第二次开始才会将顶点所属的强连通分支id更新到图节点属性中。
        sccGraph = sccGraph.outerJoinVertices(finalVertices) 
          (vid, scc, opt) => opt.getOrElse(scc)
        .cache()
        sccGraph.vertices.count()
        sccGraph.edges.count()
        prevSccGraph.unpersist(blocking = false)
        prevSccGraph = sccGraph

        //只保留属性attr._2为false的节点（这些节点是未完成连通分量打标签的节点，后面进入pregel重新着色）
        sccWorkGraph = sccWorkGraph.subgraph(vpred = (vid, data) => !data._2).cache()
       while (sccWorkGraph.numVertices < numVertices)

      // 如果达到迭代次数则返回此时的sccGraph，将不再进入pregel进行下一步的着色和打标签。
      if (iter < numIter) 
        // 初始用vid为自身节点着色，每次重新进入pregel的图将重新着色
        sccWorkGraph = sccWorkGraph.mapVertices  case (vid, (color, isFinal)) => (vid, isFinal) 

        sccWorkGraph = Pregel[(VertexId, Boolean), ED, VertexId](
          sccWorkGraph, Long.MaxValue, activeDirection = EdgeDirection.Out)(
          // vprog： 节点在自己所属连通分支和邻居所属分支中取最小者更新自己。
          (vid, myScc, neighborScc) => (math.min(myScc._1, neighborScc), myScc._2),
          // sendMsg：正向（out）向邻居传播自身所属的连通分支（只有当自己所属连通分支比邻居小才会发送消息）
          e => 
            if (e.srcAttr._1 < e.dstAttr._1) 
              Iterator((e.dstId, e.srcAttr._1))
             else 
              Iterator()
            
          ,
          // mergeMsg： 多条消息（邻居的连通分支）取最小者
          (vid1, vid2) => math.min(vid1, vid2))

        //第二个pregel：为着色后的节点打标签，final表示该节点的连通分支id已经标记完成。
        sccWorkGraph = Pregel[(VertexId, Boolean), ED, Boolean](
          sccWorkGraph, false, activeDirection = EdgeDirection.In)(
          // vprog： 如果节点id和所属连通分支id相同，则该节点是root
          //         root节点是完成连通分支标记的节点，是final （final是被标记为true）
          //         如果节点和final节点是邻居（收到的消息是final），则该节点也是final
          (vid, myScc, existsSameColorFinalNeighbor) => 
            val isColorRoot = vid == myScc._1
            (myScc._1, myScc._2 || isColorRoot || existsSameColorFinalNeighbor)
          ,
          // 从完成着色的分量的root开始，反向（in）遍历节点，当一条边上两个节点的着色不同时则不发送消息。
          e => 
            val sameColor = e.dstAttr._1 == e.srcAttr._1
            val onlyDstIsFinal = e.dstAttr._2 && !e.srcAttr._2
            if (sameColor && onlyDstIsFinal) 
              Iterator((e.srcId, e.dstAttr._2))
             else 
              Iterator()
            
          ,
          (final1, final2) => final1 || final2)
      
    
    sccGraph
  

  def createGraph[VD: ClassTag, ED: ClassTag](sc: SparkContext): Graph[String, String] =
    val graph = Graph(sc.makeRDD((1L to 7L).map((_,""))),
      sc.makeRDD(Array(Edge(2L,5L,"配偶"),Edge(5L,3L,"亲戚"),Edge(3L,2L,"子女"),Edge(4L,5L,"父母"),Edge(6L,7L,"亲戚"))))
    graph
  

  def main(args: Array[String]): Unit = 
    val sparkConf = new SparkConf().setAppName("graphx visual").setMaster("local[2]")
    val sc = SparkContext.getOrCreate(sparkConf)
    conCompts(sc)
    val graph = createGraph(sc)
    strongCompt(graph, 3)