Spark GraphX图计算核心算子实战AggreagteMessage

Posted yszd

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark GraphX图计算核心算子实战AggreagteMessage相关的知识,希望对你有一定的参考价值。

一.简介

  参考博客:https://www.cnblogs.com/yszd/p/10186556.html

二.代码实现

 1 package graphx
 2 
 3 import org.apache.log4j.{Level, Logger}
 4 import org.apache.spark.graphx.util.GraphGenerators
 5 import org.apache.spark.sql.SparkSession
 6 
 7 /**
 8   * Created by Administrator on 2019/10/22.
 9   */
10 object AggregateMessage {
11   /**
12     * 设置日志级别为WARN
13     */
14   Logger.getLogger("org").setLevel(Level.WARN)
15   def main(args: Array[String]) {
16     /**
17       * 创建spark入口
18       */
19     val spark = SparkSession.builder().appName("AggregateMessage").master("local[2]").getOrCreate()
20     val sc = spark.sparkContext
21 
22     /**
23       * 随机生成图,默认出度为4,标准偏差为1.3,并行生成numVertices,partition默认为sc的默认partition
24       */
25     val graph = GraphGenerators.logNormalGraph(sc, numVertices = 100).mapVertices((id, _) => id.toDouble)
26     graph.vertices.take(5).foreach(println)
27     /**
28       * 将用户定义的sendMsg函数应用于图形中的每个边三元组,然后使用mergeMsg函数汇聚信息到目标顶点
29       */
30     val olderFollowers = graph.aggregateMessages[(Int, Double)](triplet =>{
31       if(triplet.srcAttr > triplet.dstAttr){
32         triplet.sendToDst(1, triplet.srcAttr)
33       }
34     },
35       (a, b) => (a._1 + b._1, a._2 + b._2)
36     )
37 
38     /**
39       * 求平均值
40       */
41     val avgAgeOfOlderFollowers = olderFollowers.mapValues((id, value) => value match {case (count, totalAge) => totalAge / count})
42 
43     /**
44       * 输出结果
45       */
46     avgAgeOfOlderFollowers.collect().take(5).foreach(println)
47   }
48 }

三.结果

  随机生成的顶点数据:

    技术图片

  聚合结果:

    技术图片

以上是关于Spark GraphX图计算核心算子实战AggreagteMessage的主要内容,如果未能解决你的问题,请参考以下文章

GraphX学习笔记——Programming Guide

Spark GraphX企业运用

Spark GraphX企业运用

大数据技术之_19_Spark学习_05_Spark GraphX 应用解析小结

spark graphx图计算

Spark GraphX图计算代码实现,源码分析