Spark Graph基础

Spark Graph基础

文章目录

• • 1、图（Graph）的基本概念
  • • 1.1图的概念：
    • 1.2应用场景:
    • 1.3弹性分布式属性图
  • 2、图的构建
  • • 2.1图的基本属性
    • 2.2GraphX的图存储模式
    • • 2.2.1 vertices
      • 2.2.2 edges
      • 2.2.3 triplets
      • 2.2.4 案例
    • 2.3 图信息算子
    • 2.4 计算算子
    • • 2.4.1 案例1：计算用户粉丝数量
      • 2.4.2 案例2：找出用户社交网络中最重要的用户
      • 2.4.3 案例3：谁是网络红人
      • 2.4.4 案例4：PageRank应用
  • 3、连通分量
  • • 3.1 概念
    • 3.2 案例：计算连通分量
  • 4、Pregel框架
  • • 4.1 以PageRank为例说明Pregel的计算过程：
    • 4.2 Pregel概念
    • 4.3 案例：使用Pregel求出图中最小值
    • 4.4 案例：使用Pregel计算单源最短路径

1、图（Graph）的基本概念

1.1图的概念：

是由顶点集合(vertex)及顶点间的关系集合（边edge）组成的一种网状数据结构

• 通常表示为二元组：Gragh=（V，E）
• 可以对事物之间的关系建模
• 很好地表达了数据之间的关联性

1.2应用场景:

• 在地图应用中寻找最短路径

• 社交网络关系

• 网页间超链接关系

1.3弹性分布式属性图

• 分布式图（graph-parallel）计算和分布式数据（data-parallel）计算类似，分布式数据计算采用了一种record-centric的集合视图，而分布式图计算采用了一种vertex-centric的图视图。
• 分布式数据计算通过同时处理独立的数据来获得并发的目的，分布式图计算则是通过对图数据进行分区（即切分）来获得并发的目的。
• 更准确的说，分布式图计算递归地定义特征的转换函数（这种转换函数作用于邻居特征），通过并发地执行这些转换函数来获得并发的目的。
• GraphX的核心抽象是弹性分布式属性图，它是一个有向多重图，带有连接到每个顶点和边的用户定义的对象。
• 有向多重图中多个并行的边共享相同的源和目的顶点。支持并行边的能力简化了建模场景，相同的顶点可能存在多种关系(例如co-worker和friend)。
• 每个顶点用一个唯一的64位长的标识符（VertexID）作为key。GraphX并没有对顶点标识强加任何排序。同样，边拥有相应的源和目的顶点标识符。

2、图的构建

2.1图的基本属性

Graph[VD,ED]

class Graph[VD, ED] {val vertices: VertexRDD[VD]val edges: EdgeRDD[ED]val triplets: RDD[EdgeTriplet[VD, ED]]
}

VertexRDD[VD]
EdgeRDD[ED]
EdgeTriplet[VD,ED]

import org.aphx._
val vertices:RDD[(VertexId,Int)]=sc.makeRDD(Seq((1L,1),(2L,2),(3L,3)))
val edges=sc.makeRDD(Seq(Edge(1L,2L,1),Edge(2L,3L,2)))
val graph=Graph(vertices,edges)  //Graph[Int,Int] ?

Edge：样例类
VertexId：Long的别名

import org.aphx.GraphLoader
//加载边列表文件创建图，文件每行描述一条边，格式：srcId dstId。顶点与边的属性均为1
val graph = GraphLoader.edgeListFile(sc, "file:///opt/spark/data/")

2.2GraphX的图存储模式

Graphx借鉴PowerGraph，使用的是Vertex-Cut( 点分割 ) 方式存储图，用三个RDD存储图数据信息：

• VertexTable(id, data)：id为顶点id， data为顶点属性

• EdgeTable(pid, src, dst, data)：pid 为分区id ，src为源顶点id ，dst为目的顶点id，data为边属性

• RoutingTable(id, pid)：id 为顶点id ，pid 为分区id

GraphX中vertices、edges以及triplets

• vertices、edges以及triplets是GraphX中三个非常重要的概念

2.2.1 vertices

在GraphX中，vertices对应着名称为VertexRDD的RDD。这个RDD有顶点id和顶点属性两个成员变量。它的源码如下所示：

abstract class VertexRDD[VD](sc: SparkContext,deps: Seq[Dependency[_]]) extends RDD[(VertexId, VD)](sc, deps)

从源码中我们可以看到，VertexRDD继承自RDD[(VertexId, VD)]，这里VertexId表示顶点id，VD表示顶点所带的属性的类别。这从另一个角度也说明VertexRDD拥有顶点id和顶点属性。

2.2.2 edges

在GraphX中，edges对应着EdgeRDD。这个RDD拥有三个成员变量，分别是源顶点id、目标顶点id以及边属性。它的源码如下所示：

abstract class EdgeRDD[ED](sc: SparkContext,deps: Seq[Dependency[_]]) extends RDD[Edge[ED]](sc, deps)

从源码中我们可以看到，EdgeRDD继承自RDD[Edge[ED]]，即类型为Edge[ED]的RDD。

2.2.3 triplets

在GraphX中，triplets对应着EdgeTriplet。它是一个三元组视图，这个视图逻辑上将顶点和边的属性保存为一个RDD[EdgeTriplet[VD, ED]]。可以通过下面的Sql表达式表示这个三元视图的含义:

SELECT src.id, dst.id, src.attr, e.attr, dst.attr
FROM edges AS e LEFT JOIN vertices AS src, vertices AS dst
ON e.srcId = src.Id AND e.dstId = dst.Id

同样，也可以通过下面图解的形式来表示它的含义：

EdgeTriplet的源代码如下所示：

class EdgeTriplet[VD, ED] extends Edge[ED] {//源顶点属性var srcAttr: VD = _ // nullValue[VD]//目标顶点属性var dstAttr: VD = _ // nullValue[VD]protected[spark] def set(other: Edge[ED]): EdgeTriplet[VD, ED] = {srcId = other.srcIddstId = other.dstIdattr = other.attrthis}

EdgeTriplet类继承自Edge类，我们来看看这个父类：

case class Edge[@specialized(Char, Int, Boolean, Byte, Long, Float, Double) ED] (var srcId: VertexId = 0,var dstId: VertexId = 0,var attr: ED = null.asInstanceOf[ED])extends Serializable

Edge类中包含源顶点id，目标顶点id以及边的属性。所以从源代码中我们可以知道，triplets既包含了边属性也包含了源顶点的id和属性、目标顶点的id和属性。

2.2.4 案例

打印图的顶点和顶点的值，打印图的边，打印triplets带有属性的点和边

//属性图案例
object SparkGraph2 {def main(args: Array[String]): Unit = {//创建SparkSessionval spark = SparkSession.builder().master("local[*]").SimpleName).getOrCreate()val sc = spark.sparkContextval users = sc.parallelize(Array((3L, ("rxin", "student")),(7L, ("jgonzal", "postdoc")),(5L, ("franklin", "professor")),(2L, ("istoica", "professor"))))val relationships = sc.parallelize(Array((Edge(3L, 7L, "collaborator")),(Edge(5L, 3L, "advisor")),(Edge(2L, 5L, "colleague")),(Edge(5L, 7L, "PI"))))//通过点集合和边集合构建图val graph = Graph(users, relationships)//打印图的顶点和顶点的值println("打印图的顶点和顶点的值")graph.vertices.foreach(x => println(s"${x._1}-->${x._2}"))//打印图的边println("打印图的边")graph.edges.foreach(x => println(s"src:${x.srcId}，dst:${x.dstId},attr:${x.attr}"))//triplets带有属性的点和边println("triplets带有属性的点和边")iplets.foreach(x => String()))}
}

2.3 图信息算子

• 查看图形顶点的数量：println(graph.numVertices)
• 查看图形边的数量：println(graph.numEdges)
• 查看图形的入度：graph.inDegrees.foreach(println)
• 查看图形的出度：graph.outDegrees.foreach(println)
• 查看图形的度：graph.degrees.foreach(println)

object SparkGraph3 {def main(args: Array[String]): Unit = {//创建SparkSessionval spark = SparkSession.builder().master("local[*]").SimpleName).getOrCreate()val sc = spark.sparkContextval users = sc.parallelize(Array((1L, ("alice", 28)),(2L, ("bob", 27)),(3L, ("charlie", 65)),(4L, ("david", 42)),(5L, ("ed", 55)),(6L, ("fran", 50))))val cntCall = sc.parallelize(Array(Edge(2L, 1L, 7),Edge(2L, 4L, 2),Edge(3L, 2L, 4),Edge(3L, 6L, 3),Edge(4L, 1L, 1),Edge(5L, 2L, 2),Edge(5L, 3L, 8),Edge(5L, 6L, 3)))val graph = Graph(users,cntCall)//找出大于30岁的用户println("找出大于30岁的用户1")graph.vertices.filter{case (id,(name,age))=>age>30}.foreach(x=> String()))println("找出大于30岁的用户2")graph.vertices.filter(_._2._2>30).foreach(x=> String()))//假设打call超过5次，表示真爱。请找出他（她）们println("假设打call超过5次，表示真爱。请找出他（她）们")iplets.filter(_.attr>5).foreach(x=> println(s"${x.toString()}"))// 查看图信息// 顶点数量// 边数量// 度、入度、出度println("查看图形边的数量")println(graph.numVertices)println("查看图形边的数量")println(graph.numEdges)println("查看图形的入度")graph.inDegrees.foreach(println)println("查看图形的出度")graph.outDegrees.foreach(println)println("查看图形的度")graph.degrees.foreach(println)}
}

2.4 计算算子

• 对顶点遍历，生成新的图： graph.mapVertices((vertexId,attr)=>(vertexId,attr._1))
• 对边进行遍历，生成新的图：graph.mapEdges(e=>e.attr*7.0)
• 反转&#verse
• 筛选生成子图：graph.subgraph
• 图关联，合并迁移joinVertices和outerJoinVertices：graph1.joinVertices(graph2)((id1, attr1, attr2) => ())
• mask操作构造一个子图，这个子图包含输入图中包含的顶点和边。它的实现很简单，顶点和边均做inner join操作即可。这个操作可以和subgraph操作相结合，基于另外一个相关图的特征去约束一个图。
• groupEdges操作合并多重图中的并行边(如顶点对之间重复的边)。在大量的应用程序中，并行的边可以合并（它们的权重合并）为一条边从而降低图的大小。

object SparkGraph4 {def main(args: Array[String]): Unit = {//创建SparkSessionval spark = SparkSession.builder().master("local[*]").SimpleName).getOrCreate()val sc = spark.sparkContextval users = sc.parallelize(Array((1L, ("alice", 28)),(2L, ("bob", 27)),(3L, ("charlie", 65)),(4L, ("david", 42)),(5L, ("ed", 55)),(6L, ("fran", 50))))val cntCall = sc.parallelize(Array(Edge(2L, 1L, 7),Edge(2L, 4L, 2),Edge(3L, 2L, 4),Edge(3L, 6L, 3),Edge(4L, 1L, 1),Edge(5L, 2L, 2),Edge(5L, 3L, 8),Edge(5L, 6L, 3)))val graph = Graph(users, cntCall)//对顶点遍历，生成新的图val graph1 = graph.mapVertices((vertexId,attr)=>(vertexId,attr._1))//对边进行遍历，生成新的图val graph2 = graph.mapEdges(e=>e.attr*7.0)//打印println("graph")iplets.foreach(x=> String()))println("graph1")iplets.foreach(x=> String()))println("graph2")iplets.foreach(x=> String()))//对图里边的方向进行反转println("graph")iplets.foreach(x=> String()))println("对图里边的方向进行反转")val reverseGraph = iplets.foreach(x=> String()))println("subgraph")val subgraph = graph.subgraph(vpred = (id,attr)=>attr._2<iplets.foreach(x=> String()))println("打印图的顶点和顶点的值")subgraph.vertices.foreach(x => println(s"${x._1}-->${x._2}"))println("打印图的边")subgraph.edges.foreach(x => println(s"src:${x.srcId}，dst:${x.dstId},attr:${x.attr}"))//joinval tweeters_comps = sc.parallelize(Array((1L, "kgc"), (2L, "berkeley.edu"), (3L, "apache")))val t_graph = graph.joinVertices(tweeters_comps)((id, v, cmpy) => (v._1 + " @ " + cmpy, v._2))t_llect.foreach(println(_))}
}

2.4.1 案例1：计算用户粉丝数量

object FansGraph {def main(args: Array[String]): Unit = {//创建SparkSessionval spark = SparkSession.builder().master("local[*]").SimpleName).getOrCreate()val sc = spark.sparkContextval vertexArray = Array((1L, ("Alice", 28)),(2L, ("Bob", 27)),(3L, ("Charlie", 65)),(4L, ("David", 42)),(5L, ("Ed", 55)),(6L, ("Fran", 50)))val edgeArray = Array(Edge(2L, 1L, 7),Edge(2L, 4L, 2),Edge(3L, 2L, 4),Edge(3L, 6L, 3),Edge(4L, 1L, 1),Edge(5L, 2L, 2),Edge(5L, 3L, 8),Edge(5L, 6L, 3))val vertexRDD: RDD[(Long, (String, Int))] = sc.parallelize(vertexArray)val edgeRDD: RDD[Edge[Int]] = sc.parallelize(edgeArray)val graph: Graph[(String, Int), Int] = Graph(vertexRDD, edgeRDD)case class User(name: String, age: Int, inDeg: Int, outDeg: Int)//修改顶点属性val initialUserGraph: Graph[User, Int] = graph.mapVertices{case (id, (name, age)) => User(name, age, 0, 0)}//将顶点入度、出度存入顶点属性中val userGraph = initialUserGraph.outerJoinVertices(initialUserGraph.inDegrees) {case (id, u, inDegOpt) => User(u.name, u.age, OrElse(0), u.outDeg)}.outerJoinVertices(initialUserGraph.outDegrees) {case (id, u, outDegOpt) => User(u.name, u.age, u.inDeg, OrElse(0))}//顶点的入度即为粉丝数量for ((id, property) <- llect)println(s"User $id is ${property.name} and is liked by ${property.inDeg} people.")}
}

2.4.2 案例2：找出用户社交网络中最重要的用户

object SocialNet{def main(args: Array[String]): Unit = {//创建SparkSessionval spark = SparkSession.builder().master("local[*]").SimpleName).getOrCreate()val sc = spark.sparkContextval tweeters = Array((1L, ("Alice", 28)), (2L, ("Bob", 27)), (3L, ("Charlie", 65)), (4L, ("David", 42)), (5L, ("Ed", 55)), (6L, ("Fran", 50)))val vertexRDD: RDD[(Long, (String, Int))] = spark.sparkContext.parallelize(tweeters)val followRelations = Array(Edge[Int](2L, 1L, 7), Edge[Int](2L, 4L, 2), Edge[Int](3L, 2L, 4),  Edge[Int](3L, 6L, 3), Edge[Int](4L, 1L, 1), Edge[Int](5L, 2L, 2), Edge[Int](5L, 3L, 8), Edge[Int](5L, 6L, 3))val edgeRDD = spark.sparkContext.parallelize(followRelations)val graph: Graph[(String, Int), Int] = Graph(vertexRDD, edgeRDD)val ranks = graph.pageRank(0.0001)ranks.vertices.sortBy(_._2, false).collect.foreach(println)}
}

2.4.3 案例3：谁是网络红人

object NetRedPeople{def main(args: Array[String]): Unit = {val spark = SparkSession.builder().master("local[*]").SimpleName).getOrCreate()val sc = spark.sparkContextval regex = """(((User[0-9]{1,},[0-9]{1,})),((User[0-9]{1,},[0-9]{1,})))""".rval twitters = sc.textFile("files/07twitter_graph/twitter_").map(line => line match {case regex(followee, follower) => (Some(followee), Some(follower))case _ => (None, None)}).filter(x => x._1 != None && x._2 != None).map(x => (x._1.get.split(","), x._2.get.split(","))).map(x => (x._1(0), x._1(1).toLong, x._2(0), x._2(1).toLong))val verts = twitters.flatMap(x=>Array((x._2,x._1),(x._4,x._3))).distinct()val edges = twitters.map(x=>Edge(x._2,x._4,"follow"))val defaultUser = ("")val gragh = Graph(verts,edges,defaultUser)gragh.degrees.sortBy(lines=>lines._2,false).take(3).foreach(println(_))}
}

2.4.4 案例4：PageRank应用

PageRank（PR）算法，用于评估网页链接的质量和数量，以确定该网页的重要性和权威性的相对分数，范围为0到10
从本质上讲，PageRank是找出图中顶点（网页链接）的重要性，GraphX提供了PageRank API用于计算图的PageRank

//练习2：PageRank应用
object PageRanktest{def main(args: Array[String]): Unit = {val spark = SparkSession.builder().master("local[*]").SimpleName).getOrCreate()val sc = spark.sparkContext// Load the edges as a graphval graph = GraphLoader.edgeListFile(sc,"hdfs://192.168.221.140:9000/kb10/")// Run PageRankval ranks = graph.pageRank(0.0001).vertices// Join the ranks with the usernamesval users = sc.textFile("hdfs://192.168.221.140:9000/kb10/").map {line => {val fields = line.split(",")(fields(0).toLong, fields (1))}}val ranksByUsername = users.join(ranks).map {case (id, (username, rank)) => (username, rank)}ranksByUsername.sortBy(line=>line._2,false).collect.foreach(println(_))}
}

3、连通分量

3.1 概念

• 连通分量是一个子图，其中任何两个顶点通过一条边或一系列边相互连接
• 其顶点是原始图顶点集的子集，其边是原始图边集的子集
• 只要求出图的所有连通分量，就可以知道图中任意两顶点之间是否有路径可达

3.2 案例：计算连通分量

//演示示例10：计算连通分量
object ConnectedComponent{def main(args: Array[String]): Unit = {val spark = SparkSession.builder().master("local[*]").SimpleName).getOrCreate()val sc = spark.sparkContextval vertexArray = Array((1L, ("Alice", 28)),(2L, ("Bob", 27)),(3L, ("Charlie", 65)),(4L, ("David", 42)),(5L, ("Ed", 55)),(8L, ("Fran", 50)),(9L, ("aaa", 50)))val edgeArray = Array(Edge(1L, 2L, 7),Edge(2L, 3L, 2),Edge(3L, 4L, 4),Edge(4L, 5L, 3),Edge(8L, 9L, 1)
//      Edge(7L, 1L, 2),
//      Edge(5L, 7L, 8),
//      Edge(5L, 6L, 3))val vertexRDD = sc.parallelize(vertexArray)val edgeRDD = sc.parallelize(edgeArray)val graph = Graph(vertexRDD,llect.foreach(println)// 连通分量是一个子图，其中任何两个顶点通过一条边或一系列边相互连接，// 其顶点是原始图顶点集的子集，其边是原始图边集的子集// 只要求出图的所有连通分量，就可以知道图中任意两顶点之间是否有路径可达println("---连通分量")llect.foreach(println)}
}

4、Pregel框架

• Pregel是一种面向图算法的分布式编程框架，采用迭代的计算模型：在每一轮，每个顶点处理上一轮收到的消息，并发出消息给其它顶点，并更新自身状态和拓扑结构（出、入边）等。

4.1 以PageRank为例说明Pregel的计算过程：

• 顶点v首先接收来自上一次迭代的消息，计算它们的和。
• 然后使用计算的消息和重新计算PageRank
• 之后程序广播这个重新计算的PageRank的值到顶点v的所有邻居
• 最后程序判断算法是否应该停止。

def PageRank(v: Id, msgs: List[Double]) {
// 计算消息和
var msgSum = 0
for (m <- msgs) { msgSum = msgSum + m }
// 更新 PageRank (PR)
A(v).PR = 0.15 + 0.85 * msgSum
// 广播新的PR消息
for (j <- OutNbrs(v)) {
msg = A(v).PR / A(v).NumLinks
send_msg(to=j, msg)
}
// 检查终止
if (converged(A(v).PR)) voteToHalt(v)
}

4.2 Pregel概念

Pregel选择了一种纯消息传递的模式，忽略远程数据读取和其他共享内存的方式，这样做有两个原因。

第一，消息的传递有足够高效的表达能力，不需要远程读取（remote reads）。

第二，性能的考虑。在一个集群环境中，从远程机器上读取一个值是会有很高的延迟的，这种情况很难避免。而消息传递模式通过异步和批量的方式传递消息，可以缓解这种远程读取的延迟。

GraphX也是基于BSP模式。GraphX公开了一个类似Pregel的操作，它是广泛使用的Pregel和GraphLab抽象的一个融合。在GraphX中，Pregel操作者执行一系列的超步，在这些超步中，顶点从之前的超步中接收进入(inbound)消息，为顶点属性计算一个新的值，然后在以后的超步中发送消息到邻居顶点。

不像Pregel而更像GraphLab，消息通过边triplet的一个函数被并行计算，消息的计算既会访问源顶点特征也会访问目的顶点特征。在超步中，没有收到消息的顶点会被跳过。当没有消息遗留时，Pregel操作停止迭代并返回最终的图。

4.3 案例：使用Pregel求出图中最小值

• initialMsg：在“superstep 0”之前发送至顶点的初始消息
• maxIterations：将要执行的最大迭代次数
• activeDirection：发送消息方向（默认是出边方向：EdgeDirection.Out）
• vprog：用户定义函数，用于顶点接收消息
• sendMsg：用户定义的函数，用于确定下一个迭代发送的消息及发往何处
• mergeMsg：用户定义的函数，在vprog前，合并到达顶点的多个消息

object Pregeltest{def main(args: Array[String]): Unit = {val spark = SparkSession.builder().master("local[*]").SimpleName).getOrCreate()val sc = spark.sparkContextval initialMsg = 9999def vprog(vertexId: VertexId,value:(Int,Int),message:Int):(Int,Int)={if (message==initialMsg) {value}else{(message min value._1,value._1)}}def sendMsg(triplet:EdgeTriplet[(Int,Int),Boolean]):Iterator[(VertexId,Int)]={//srcAttr:源顶点属性//dstAttr:目标顶点属性val sourceVertex = triplet.srcAttrif (sourceVertex._1==sourceVertex._2) ptyelse Iterator((triplet.dstId,sourceVertex._1))}def mergeMsg(msg1:Int,msg2:Int):Int=msg1 min msg2//创建顶点集RDDval vertices = sc.parallelize(Array((1L, (7,-1)), (2L, (3,-1)),  (3L, (2,-1)), (4L, (6,-1))))//创建边集RDDval relationships = sc.parallelize(Array(Edge(1L, 2L, true), Edge(1L, 4L, true),  Edge(2L, 4L, true), Edge(3L, 1L, true),  Edge(3L, 4L, true)))//创建图val graph = Graph(vertices,relationships)//Pregelval minGraph = graph.pregel(initialMsg,Int.MaxValue,EdgeDirection.Out)(vprog,sendMsg,mergeMsg)llect.foreach{case (vertexId,(value,original_value)) => println(value)}}
}

4.4 案例：使用Pregel计算单源最短路径

object Pregeltest2{def main(args: Array[String]): Unit = {val spark = SparkSession.builder.master("local[*]").SimpleName).getOrCreate()val sc = spark.sparkContextval vertices:RDD[(VertexId,Double)]=sc.makeRDD(Seq((0L,1.0),(1L,1.0),(2L,1.0),(3L,1.0)))val edges=sc.makeRDD(Seq(Edge(0L,1L,100),Edge(0L,2L,30),Edge(0L,4L,10),Edge(2L,1L,60),Edge(2L,3L,60),Edge(3L,1L,10),Edge(4L,3L,50)))val graph=Graph(vertices,edges)val sourceId: VertexId = 0Lval initGraph=graph.mapVertices((id, _) => if (id == sourceId) 0 else Double.PositiveInfinity)val sssp=initGraph.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))pty},(dist1,dist2)=>math.min(dist1,dist2)  //合并消息)println(llect().mkString("n"))}
}