spark20.sparkGraphx_2_图的转换

1.Graph的创建

1.根据边和顶点来创建。

def apply[VD: ClassTag, ED: ClassTag](vertices: RDD[(VertexId, VD)],edges: RDD[Edge[ED]],defaultVertexAttr: VD &＃61; null.asInstanceOf[VD],edgeStorageLevel: StorageLevel &＃61; StorageLevel.MEMORY_ONLY,vertexStorageLevel: StorageLevel &＃61; StorageLevel.MEMORY_ONLY): Graph[VD, ED]


2.根据边来创建。

所有顶点的属性相同&＃xff0c;都是VD类型的defaultValue。

def fromEdges[VD: ClassTag, ED: ClassTag](edges: RDD[Edge[ED]],defaultValue: VD,edgeStorageLevel: StorageLevel &＃61; StorageLevel.MEMORY_ONLY,vertexStorageLevel: StorageLevel &＃61; StorageLevel.MEMORY_ONLY): Graph[VD, ED]


3.根据裸边&＃xff08;只有顶点ID&＃xff09;进行创建。

顶点的属性是defaultValue&＃xff0c;边的属性为相同顶点边的个数&＃xff0c;默认为1。

def fromEdgeTuples[VD: ClassTag](rawEdges: RDD[(VertexId, VertexId)],defaultValue: VD,uniqueEdges: Option[PartitionStrategy] &＃61; None,edgeStorageLevel: StorageLevel &＃61; StorageLevel.MEMORY_ONLY,vertexStorageLevel: StorageLevel &＃61; StorageLevel.MEMORY_ONLY): Graph[VD, Int]


2.Graph的转换

1.基本信息

1. numEdges 返回边的数量
2. numVertices 顶点的个数
3. inDegrees:VertexRDD[Int] 返回顶点的入度&＃xff0c;返回类型为RDD[(VertexId,Int)] Int是入度的具体值。
4. outDegrees:VertexRDD[Int] 返回顶点的出度&＃xff0c;返回类型为RDD[(VertexId,Int)] Int是出度的具体值。
5. degrees:VertexRDD[Int] 返回顶点的入度与出度之和&＃xff0c;返回类型为RDD[(VertexId,Int)] Int是度的具体值。

2.转换操作

def mapVertices[VD2: ClassTag](map: (VertexId, VD) &＃61;> VD2)(implicit eq: VD &＃61;:&＃61; VD2 &＃61; null): Graph[VD2, ED]


对图中的每一个顶点进行map操作&＃xff0c;顶点ID不能变&＃xff0c;可以将顶点的属性改变成另一种类型。
如&＃xff1a;scala> graph.mapVertices((id,attr)&＃61;>attr._1&＃43;":"&＃43;attr._2)

def mapEdges[ED2: ClassTag](map: Edge[ED] &＃61;> ED2): Graph[VD, ED2]


对图中的每个边进行map操作&＃xff0c;边的方向不能改变&＃xff0c;可以将边的属性改为lin一种类型。

def mapTriplets[ED2: ClassTag](map: EdgeTriplet[VD, ED] &＃61;> ED2): Graph[VD, ED2]


对图中的每个三元组进行map操作&＃xff0c;只能修改边的属性。

3.结构操作

• reverse 反转

def reverse: Graph[VD, ED]


反转整个图&＃xff0c;将边的方向调头。
如&＃xff1a;

graph.reverse.triplets.map(x&＃61;>"["&＃43;x.srcId&＃43;":"&＃43;x.srcAttr&＃43;"-->"&＃43;x.attr&＃43;"-->"&＃43;x.dstId&＃43;":"&＃43;x.dstAttr&＃43;"]").collect.foreach(println)


• subgrap 获取子图

def subgraph(epred: EdgeTriplet[VD, ED] &＃61;> Boolean &＃61; (x &＃61;> true),vpred: (VertexId, VD) &＃61;> Boolean &＃61; ((v, d) &＃61;> true)): Graph[VD, ED]


获取子图。可以通过参数名来指定传参&＃xff0c;如果subgraph中有的边没有顶点对应&＃xff0c;那么会自动将该边去除。

graph.subgraph(vpred&＃61;(id,attr)&＃61;>attr._2 &＃61;&＃61; "professor").triplets.map(x&＃61;>"["&＃43;x.srcId&＃43;":"&＃43;x.srcAttr&＃43;"-->"&＃43;x.attr&＃43;"-->"&＃43;x.dstId&＃43;":"&＃43;x.dstAttr&＃43;"]").collect.foreach(println)


没有边的顶点不会自动被删除。

graph.subgraph((x&＃61;>false)).numVertices


• mask 求两个图的交集

def mask[VD2: ClassTag, ED2: ClassTag](other: Graph[VD2, ED2]): Graph[VD, ED]


将当前图和other图做交集&＃xff0c;返回一个新图。如果other中的属性与原图的属性不同&＃xff0c;那么保留原图的属性。

val other &＃61;graph.subgraph(vpred&＃61;(id,attr)&＃61;>attr._2 &＃61;&＃61; "professor").mapVertices((id,attr)&＃61;>attr._1 &＃43;":"&＃43;attr._2)
other.triplets.map(x&＃61;>"["&＃43;x.srcId&＃43;":"&＃43;x.srcAttr&＃43;"-->"&＃43;x.attr&＃43;"-->"&＃43;x.dstId&＃43;":"&＃43;x.dstAttr&＃43;"]").collect.foreach(println) //输出other
graph.mask(other).triplets.map(x&＃61;>"["&＃43;x.srcId&＃43;":"&＃43;x.srcAttr&＃43;"-->"&＃43;x.attr&＃43;"-->"&＃43;x.dstId&＃43;":"&＃43;x.dstAttr&＃43;"]").collect.foreach(println)


• groupEdges 合并两条边

def groupEdges(merge: (ED, ED) &＃61;> ED): Graph[VD, ED]


合并两条边&＃xff0c;通过函数合并边的属性。【注意&＃xff1a;两条边要在一个分区内。】

4.聚合

collectNeighbors

def collectNeighbors(edgeDirection: EdgeDirection): VertexRDD[Array[(VertexId, VD)]]


收集邻居节点的数据&＃xff0c;根据指定的方向。返回的数据为RDD[(VertexId,Array[(VertexId,VD)])] 顶点的属性的一个数组。数组中包含邻居节点的顶点。

graph.collectNeighbors(EdgeDirection.In).collect


collectNeighborIds

def collectNeighborIds(edgeDirection: EdgeDirection): VertexRDD[Array[VertexId]]


与上一个方法类似&＃xff0c;只收集ID。

graph.collectNeighborIds(EdgeDirection.In).collect


aggregateMessages

def aggregateMessages[A: ClassTag](sendMsg: EdgeContext[VD, ED, A] &＃61;> Unit,mergeMsg: (A, A) &＃61;> A,tripletFields: TripletFields &＃61; TripletFields.All): VertexRDD[A]


每个边都会通过sendMsg发送一个消息&＃xff0c;每个顶点都会通过mergeMsg来处理它收到的消息&＃xff0c;tripletFields存在主要用于定制EdgeContext对象中的属性的值是否存在&＃xff0c;为了减少数据通信量。

//初始化顶点集合val vertexArray &＃61; Array((1L, ("Alice", 28)),(2L, ("Bob", 27)),(3L, ("Charlie", 65)),(4L, ("David", 42)),(5L, ("Ed", 55)),(6L, ("Fran", 50)))//创建顶点的RDD表示val vertexRDD: RDD[(Long, (String, Int))] &＃61; sc.parallelize(vertexArray)//初始化边的集合val edgeArray &＃61; Array(Edge(2L, 1L, 7),Edge(2L, 4L, 2),Edge(3L, 2L, 4),Edge(3L, 6L, 3),Edge(4L, 1L, 1),Edge(2L, 5L, 2),Edge(5L, 3L, 8),Edge(5L, 6L, 3))//创建边的RDD表示val edgeRDD: RDD[Edge[Int]] &＃61; sc.parallelize(edgeArray)//创建一个图val graph: Graph[(String, Int), Int] &＃61; Graph(vertexRDD, edgeRDD)
graph.aggregateMessages[Array[(VertexId, (String, Int))]](ctx &＃61;> ctx.sendToDst(Array((ctx.srcId.toLong, (ctx.srcAttr._1, ctx.srcAttr._2)))), _ &＃43;&＃43; _).collect.foreach(v &＃61;> {println(s"id: ${v._1}"); for (arr <- v._2) {println(s" ${arr._1} (name: ${arr._2._1} age: ${arr._2._2})")}})

5.关联操作

joinVertices

def joinVertices[U: ClassTag](table: RDD[(VertexId, U)])(mapFunc: (VertexId, VD, U) &＃61;> VD): Graph[VD, ED]


将相同顶点ID的数据进行加权&＃xff0c;将U这种类型的数据加入到VD这种类型的数据上&＃xff0c;但是不能修改VD的类型。可以使用case class 类型&＃xff0c;将VD封装为case class&＃xff0c;mapFunc对VD进行属性补全。

outerJoinVertices

def outerJoinVertices[U: ClassTag, VD2: ClassTag](other: RDD[(VertexId, U)])( mapFunc: (VertexId, VD, Option[U]) &＃61;> VD2)(implicit eq: VD &＃61;:&＃61; VD2 &＃61; null) : Graph[VD2, ED]


和joinVertices类似&＃xff0c;只是如果没有相应的节点&＃xff0c;那么join的值默认为None。

6.Pregel

前提&＃xff1a;

对于节点来说有两种状态&＃xff1a;1.钝化态&＃xff0c;类似于休眠&＃xff0c;不做任何事。2.激活态&＃xff0c;干活。
节点能够处于激活态需要有条件&＃xff1a;1.节点收到消息或者2.成功发送了任何一条消息。

def pregel[A: ClassTag](initialMsg: A,maxIterations: Int &＃61; Int.MaxValue,activeDirection: EdgeDirection &＃61; EdgeDirection.Either)(vprog: (VertexId, VD, A) &＃61;> VD,sendMsg: EdgeTriplet[VD, ED] &＃61;> Iterator[(VertexId, A)],mergeMsg: (A, A) &＃61;> A): Graph[VD, ED]


initialMsg 图初试化的时候&＃xff0c;开始模型计算的时候&＃xff0c;所有节点都会收到一个消息&＃xff0c;所有节点都是active的。
maxIterations 最大迭代次数。
activeDirection 规定了发送消息的方向。
柯里化&＃xff1a;
vprog 激活态且具有activeDirection的节点调用该消息将聚合后的数据和本节点进行属性的合并。
sendMsg 激活态的节点调用该方法发送消息。
mergeMsg 如果一个节点接收到多个消息&＃xff0c;先用mergeMsg来将多条消息聚合成一条消息。如果节点只收到一条消息&＃xff0c;则不会调用该函数。
实例&＃xff1a;求节点5到各个节点的最短距离。

代码实现&＃xff1a;

package com.dengdan.practiceimport org.apache.log4j.{Level, Logger}
import org.apache.spark.graphx.{Edge, _}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object Practice extends App {//屏蔽日志Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)//设定一个SparkConfval conf &＃61; new SparkConf().setAppName("simpleGraphx").setMaster("local[*]")val sc &＃61; new SparkContext(conf)//初始化顶点集合val vertexArray &＃61; Array((1L, ("Alice", 28)),(2L, ("Bob", 27)),(3L, ("Charlie", 65)),(4L, ("David", 42)),(5L, ("Ed", 55)),(6L, ("Fran", 50)))//创建顶点RDD表示val vertexRDD: RDD[(Long, (String, Int))] &＃61; sc.parallelize(vertexArray)//初始化边表示val edgeArray &＃61; Array(Edge(2L, 1L, 7),Edge(2L, 4L, 2),Edge(3L, 2L, 4),Edge(3L, 6L, 3),Edge(4L, 1L, 1),Edge(2L, 5L, 2),Edge(5L, 3L, 8),Edge(5L, 6L, 3))//创建边RDD表示val edgeRDD: RDD[Edge[Int]] &＃61; sc.parallelize(edgeArray)//创建图val graph: Graph[(String, Int), Int] &＃61; Graph(vertexRDD, edgeRDD)//*************************** 实用操作 ****************************************println("聚合操作")println("**********************************************************")val sourceId: VertexId &＃61; 5L //定义源点val initialGraph &＃61; graph.mapVertices((id, _) &＃61;> if (id &＃61;&＃61; sourceId) 0.0 else Double.PositiveInfinity)initialGraph.triplets.collect() foreach (println)println("找出5到各顶点的最短距离")val sssp &＃61; initialGraph.pregel(Double.PositiveInfinity, Int.MaxValue, EdgeDirection.Out)(//id为满足条件的节点的编号&＃xff0c;dist是该节点的属性值&＃xff0c;newDist为收到消息后&＃xff0c;新的属性(id, dist, newDist) &＃61;> {println("||||" &＃43; id &＃43; " 收到消息")math.min(dist, newDist)},triplet &＃61;> {println(">>>>" &＃43; triplet.srcId &＃43; " 发送消息")//源节点的属性 &＃43; 边的属性值 <目标节点的属性if (triplet.srcAttr &＃43; triplet.attr < triplet.dstAttr) {//发送成功// Interator 表示发送&＃xff0c;发送给dstId&＃xff0c;发送的内容为 triplet.srcAttr &＃43; triplet.attrIterator((triplet.dstId, triplet.srcAttr &＃43; triplet.attr))} else {//发送失败Iterator.empty}},(a, b) &＃61;> {println("$$$$$")math.min(a, b)} //当前节点所有输入的最短距离)println("--------------图信息------------------")sssp.triplets.collect().foreach(println)println("-------------节点5到各个节点的最短距离---")println(sssp.vertices.collect.mkString("\n"))sc.stop()
}

运行结果&＃xff1a;

聚合操作
**********************************************************
((2,Infinity),(1,Infinity),7)
((2,Infinity),(4,Infinity),2)
((3,Infinity),(2,Infinity),4)
((3,Infinity),(6,Infinity),3)
((4,Infinity),(1,Infinity),1)
((2,Infinity),(5,0.0),2)
((5,0.0),(3,Infinity),8)
((5,0.0),(6,Infinity),3)
找出5到各顶点的最短距离
# 初始化时
||||4 收到消息
||||6 收到消息
||||5 收到消息
||||2 收到消息
||||3 收到消息
||||1 收到消息
>>>>4 发送消息
>>>>2 发送消息
>>>>2 发送消息
>>>>5 发送消息
>>>>3 发送消息
>>>>2 发送消息
>>>>3 发送消息
>>>>5 发送消息
# 第一轮迭代
||||3 收到消息
||||6 收到消息
>>>>3 发送消息
>>>>3 发送消息
# 第二轮迭代
||||2 收到消息
>>>>2 发送消息
>>>>2 发送消息
>>>>2 发送消息
# 第三轮迭代
||||1 收到消息
||||4 收到消息
>>>>4 发送消息
# 第四轮迭代
||||1 收到消息
# 迭代结束
--------------图信息------------------
((2,12.0),(1,15.0),7)
((2,12.0),(4,14.0),2)
((3,8.0),(2,12.0),4)
((3,8.0),(6,3.0),3)
((4,14.0),(1,15.0),1)
((2,12.0),(5,0.0),2)
((5,0.0),(3,8.0),8)
((5,0.0),(6,3.0),3)
-------------节点5到各个节点的最短距离---
(1,15.0)
(2,12.0)
(3,8.0)
(4,14.0)
(5,0.0)
(6,3.0)