GraphX图计算之aggregateMessages算子案例详解

叁歲伎倆 2022-12-03 13:10 95阅读 0赞

## 一.简介 ##

GraphX中的核心聚合操作为aggregateMessages。该运算符将用户定义的sendMsg函数应用于图形中的每个边三元组，然后使用该mergeMsg函数在其目标顶点处聚合这些消息。

class Graph[VD, ED] { 
      def aggregateMessages[Msg: ClassTag](
          sendMsg: EdgeContext[VD, ED, Msg] => Unit,
          mergeMsg: (Msg, Msg) => Msg,
          tripletFields: TripletFields = TripletFields.All)
        : VertexRDD[Msg]
    }

用户定义的sendMsg函数采用EdgeContext，将公开源和目标属性以及边属性和函数（sendToSrc和sendToDst），以将消息发送到源和目标节点。sendMsg可以认为是 map-reduce中的map函数。用户定义的mergeMsg函数接受两条发往同一顶点的消息，并产生一条消息。可以认为是map-reduce中的reduce函数。Graph的 aggregateMessages操作返回一个VertexRDD\[Msg\] ，包含发往每个顶点的聚合消息（类型的Msg）。未收到消息的顶点不包含在返回的VertexRDD中。

另外，aggregateMessages采用一个可选参数 tripletsFields，该参数指示访问哪些数据EdgeContext （即，源顶点属性，而不是目标顶点属性）。Graph的可能选项在tripletsFields中定义，TripletFields默认值为TripletFields.All，指示用户定义的sendMsg函数可以访问任何顶点。该tripletFields参数可用于限制GraphX仅访问部分顶点， EdgeContext允许GraphX选择优化的联接策略。例如，如果我们正在计算每个用户的关注者的平均年龄，则仅需要源字段，因此我们可以TripletFields.Src用来表明我们仅需要源字段。

## 二.代码实战 ##

package spark2.graphx
    
    import org.apache.log4j.{ Level, Logger}
    import org.apache.spark.graphx._
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.SparkSession
    
    object AggregateMessagesExample { 
      Logger.getLogger("org").setLevel(Level.WARN)
      def main(args: Array[String]): Unit = { 
        val spark = SparkSession
          .builder
          .appName(s"${this.getClass.getSimpleName}")
          .master("local[2]")
          .getOrCreate()
        val sc = spark.sparkContext
    
        // 加载顶点数据
        val vertexLines: RDD[String] = sc.textFile("D:\\software\\spark-2.4.4\\data\\graphx\\vertices.txt")
        val vertices: RDD[(VertexId, Double)] = vertexLines.map(line => { 
          val cols = line.split(",")
          (cols(0).toLong, cols(1).toDouble)
        })
    
        // 加载边数据
        val edgeLines: RDD[String] = sc.textFile("D:\\software\\spark-2.4.4\\data\\graphx\\edges.txt")
        val edges:RDD[Edge[Int]] = edgeLines.map(line => { 
          val cols = line.split(",")
          Edge(cols(0).toInt, cols(1).toInt, cols(2).toInt)
        })
    
        // 使用vertices和edges生成图
        val graph:Graph[Double, Int] = Graph(vertices, edges)
        graph.triplets.collect.foreach(println)
    
        // 统计相同目标顶点的个数和属性之和
        val olderFollowers: VertexRDD[(Int, Double)] = graph.aggregateMessages[(Int, Double)](
          triplet => {  // 类似旧算子mapReduceTriplets中的map
            //if (triplet.srcAttr > triplet.dstAttr) { 
              // 项目标顶点发送源顶点信息
              triplet.sendToDst((1, triplet.srcAttr)) // 发往目标顶点的源顶点个数，msg
           // }
          },
          // 计数，求和
          // 类似旧算子mapReduceTriplets中的reduce
          (a, b) => (a._1 + b._1, a._2 + b._2) // 相同目标顶点的源顶点个数之和及属性和
        )
        olderFollowers.collect.foreach(println)
    
        /** * 根据个数和属性之和求属性的平均值 */
        val avgAgeOfOlderFollowers: VertexRDD[Double] =
          olderFollowers.mapValues( (id, value) =>
            value match { case (count, totalAge) => totalAge / count})
        // 打印结果信息
        avgAgeOfOlderFollowers.collect.foreach(println)
    
        spark.stop()
      }
    }

**备注：详细注释参考代码中的备注！**

## 三.执行结果 ##

节点和边的属性信息：  
![在这里插入图片描述][20200903173633377.png_pic_center]  
相同源顶点的计算和属性累计值：  
![在这里插入图片描述][20200903173646956.png_pic_center]  
属性均值：  
![在这里插入图片描述][20200903173706867.png_pic_center]

[20200903173633377.png_pic_center]: /images/20221123/eca841f3f48940b780ba4770c8eac84f.png
[20200903173646956.png_pic_center]: /images/20221123/0e693b4049d54d309c24b082489f3471.png
[20200903173706867.png_pic_center]: /images/20221123/69f802669db54f62a12ca2558aa7f463.png