第1關：GraphX-構建圖及相關基本操作

import org.apache.log4j.{Level, Logger} import org.apache.spark.graphx._ import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object GraphX_Test_stu{def main(args:Array[String]): Unit ={//屏蔽日志Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)//設置運行環境val conf = new SparkConf().setAppName("SimpleGraph").setMaster("local")val sc = new SparkContext(conf)//設置頂點和邊，注意頂點和邊都是用元組定義的Array//頂點的數據類型是VD:(String,Int)val vertexArray = Array((1L,("Bob",89)),(2L,("Sunny",70)),(3L,("Tony",99)),(4L,("Helen",58)),(5L,("John",55)),(6L,("Tom",83)),(7L,("Marry",94)),(8L,("Cook",76)),(9L,("Linda",84)))//邊的數據類型ED:Intval edgeArray = Array(Edge(1L,2L,5),Edge(1L,3L,9),Edge(2L,4L,4),Edge(3L,4L,6),Edge(3L,6L,8),Edge(3L,7L,4),Edge(4L,5L,7),Edge(4L,8L,6),Edge(8L,3L,7),Edge(8L,7L,2),Edge(8L,9L,1))//構造vertexRDD和edgeRDDval vertexRDD:RDD[(Long,(String,Int))] = sc.parallelize(vertexArray)val edgeRDD:RDD[Edge[Int]] = sc.parallelize(edgeArray)//構造Graph[VD,ED]val graph:Graph[(String,Int),Int] = Graph(vertexRDD, edgeRDD)//*********************圖的屬性//找出圖中成績大于60的頂點println("Find the vertices with scores greater than 60 in the graph")graph.vertices.filter{case (id,(name,grade)) => grade > 60}.collect.foreach{case (id,(name,grade)) => println(s"$name $grade")}println//邊操作，找出圖中邊屬性大于5的邊println("Find the edge of the graph whose edge attribute is greater than 5")graph.edges.filter(e => e.attr > 5).collect.foreach(e => println(s"${e.srcId} to ${e.dstId} att ${e.attr}"))println//triplets操作.((srcId,srcAttr),(dstID,dstAttr),attr)//列出邊屬性>5的tripltesprintln("Find the tripltes with edge attributes greater than 5")for (triplet <- graph.triplets.filter(t => t.attr > 5).collect){println(s"${triplet.srcAttr._1} ${triplet.dstAttr._1}")}println//Degrees操作//找出圖中最大的出度、入度、度數println("Find the maximum outDegrees, inDegrees, and Degrees in the graph")def max(a:(VertexId,Int),b:(VertexId,Int)):(VertexId,Int) = {if(a._2 > b._2) a else b}println("max of outDegrees" + graph.outDegrees.reduce(max) + " max of inDegrees" + graph.inDegrees.reduce(max) + " max of Degrees" + graph.degrees.reduce(max))//********************轉換操作//頂點的轉換操作，頂點成績+10println("Vertex conversion operation vertex scores added 10")graph.mapVertices{ case (id, (name, age)) => (id, (name,age+10))}.vertices.collect.foreach(v => println(s"${v._2._1} is ${v._2._1}"))println//邊的轉換操作，邊的屬性println("Edge conversion operation multiplying the attribute of the edge by 2")graph.mapEdges(e => e.attr*2).edges.collect.foreach(e => println(s"${e.srcId} to ${e.dstId} att ${e.attr}"))println//********************結構操作//找出頂點成績>60的子圖println("Find subgraphs with vertex scores greater than 60")val subGraph = graph.subgraph(vpred = (id, vd) => vd._2 >= 60)//找出子圖所有頂點println("Find all the vertices of the subgraph:")subGraph.vertices.collect.foreach(v => println(s"${v._2._1} is ${v._2._2}"))println//找出子圖所有邊println("Find all sides of the subgraph:")subGraph.edges.collect.foreach(e => println(s"${e.srcId} to ${e.dstId} att ${e.attr}"))println//********************結構操作//連接操作val inDegrees:VertexRDD[Int] = graph.inDegreescase class User(name:String,grade:Int,inDeg:Int,outDeg:Int)//創建一個新圖，頂點VD的數據類型為User，并從graph做類型轉換val initialUserGraph:Graph[User,Int] = graph.mapVertices{case (id,(name,grade)) => User(name,grade,0,0)}//initialUserGraph與inDegrees、outDegrees(RDD)進行連接//并修改initialUserGraph中inDeg值、outDeg值val userGraph = initialUserGraph.outerJoinVertices(initialUserGraph.inDegrees){case(id, u, inDegOpt) => User(u.name, u.grade, inDegOpt.getOrElse(0), u.outDeg)}.outerJoinVertices(initialUserGraph.outDegrees){case(id, u, outDegOpt) => User(u.name, u.grade, u.inDeg, outDegOpt.getOrElse(0))}//連接圖的屬性userGraph.vertices.collect.foreach(v => println(s"${v._2.name} inDeg: ${v._2.inDeg} outDeg:${v._2.outDeg}"))println//找出出度和入度相同的頂點println("Find the same vertex with the same degree of penetration")userGraph.vertices.filter{case (id,u) => u.inDeg == u.outDeg}.collect.foreach{case (id,property) => println(property.name)}printlnsc.stop()} }

第2關：GraphX-求最短路徑、二跳鄰居操作

import org.apache.log4j.{Level,Logger} import org.apache.spark.{SparkContext,SparkConf} import org.apache.spark.graphx._ import org.apache.spark.rdd.RDD object GraphX_Test_2_stu{def main(args:Array[String]): Unit ={//屏蔽日志Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)//設置運行環境val conf = new SparkConf().setAppName("SimpleGraph").setMaster("local")val sc = new SparkContext(conf)//設置頂點和邊，注意頂點和邊都是用元組定義的Array//頂點的數據類型是VD:(String,Int)val vertexArray = Array((1L,("Bob",89)),(2L,("Sunny",70)),(3L,("Tony",99)),(4L,("Helen",58)),(5L,("John",55)),(6L,("Tom",83)),(7L,("Marry",94)),(8L,("Cook",76)),(9L,("Linda",84)))//邊的數據類型ED:Intval edgeArray = Array(Edge(1L,2L,5),Edge(1L,3L,9),Edge(2L,4L,4),Edge(3L,4L,6),Edge(3L,6L,8),Edge(3L,7L,4),Edge(4L,5L,7),Edge(4L,8L,6),Edge(8L,3L,7),Edge(8L,7L,2),Edge(8L,9L,1))//構造vertexRDD和edgeRDDval vertexRDD:RDD[(Long,(String,Int))] = sc.parallelize(vertexArray)val edgeRDD:RDD[Edge[Int]] = sc.parallelize(edgeArray)//構造Graph[VD,ED]val graph:Graph[(String,Int),Int] = Graph(vertexRDD, edgeRDD)//********************實用操作//找出頂點1到各頂點的最短距離println("Find the shortest distance from vertex 1 to each vertex")val sourceId:VertexId = 1L //定義遠點val initialGraph = graph.mapVertices((id,_) => if (id == sourceId) 0.0 else Double.PositiveInfinity)val sssp = initialGraph.pregel(Double.PositiveInfinity)((id,dist,newDist) => math.min(dist,newDist),triplet => {//計算權重if(triplet.srcAttr + triplet.attr < triplet.dstAttr){Iterator((triplet.dstId,triplet.srcAttr + triplet.attr))}else{Iterator.empty}},(a,b) => math.min(a,b))println(sssp.vertices.collect.mkString("\n"))printlndef sendMsgFunc(edge:EdgeTriplet[Int, Int]) = {if(edge.srcAttr <= 0){if(edge.dstAttr <= 0){// 如果雙方都小于0，則不發送信息Iterator.empty}else{// srcAttr小于0，dstAttr大于零，則將dstAttr-1后發送Iterator((edge.srcId, edge.dstAttr - 1))}}else{if(edge.dstAttr <= 0){// srcAttr大于0，dstAttr<0,則將srcAttr-1后發送Iterator((edge.dstId, edge.srcAttr - 1))}else{// 雙方都大于零，則將屬性-1后發送val toSrc = Iterator((edge.srcId, edge.dstAttr - 1))val toDst = Iterator((edge.dstId, edge.srcAttr - 1))toDst ++ toSrc}}}val friends = Pregel(graph.mapVertices((vid, value)=> if(vid == 1) 2 else -1),// 發送初始值-1,// 指定階數2,// 雙方向發送EdgeDirection.Either)(// 將值設為大的一方vprog = (vid, attr, msg) => math.max(attr, msg),//sendMsgFunc,//(a, b) => math.max(a, b)).subgraph(vpred = (vid, v) => v >= 0)println("Confirm Vertices of friends ")friends.vertices.collect.foreach(println(_))sc.stop()} }

總結

以上是生活随笔為你收集整理的educoder中Spark GraphX—构建图及相关操作的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。