Spark+GraphX图

Q：什么是图？图的应用场景

A：图是由顶点集合(vertex)及顶点间的关系集合（边edge）组成的一种网状数据结构，表示为二元组：Gragh=（V，E），V\E分别是顶点和边的集合。图很好的表达了事物间的练习，常用于对事物之间的关系建模。常见应用场景有：在地图应用中寻找最短路径、社交网络关系、网页间超链接关系。

——————————————————————————————————————————

Q：有向图与无向图是什么？

A：图的顶点间的连系即边是有向的，有向<A,B>,<C,A>，源顶点到目标顶点的顺序是固定的，形成了顶点的出度和入度。

——————————————————————————————————————————

Q：有环图和无环图是什么？

A：有环图即包含一系列顶点链接的环路，即存在某一点出发还能回到自身。无环图即不存在一点从自身出发还可以回到自身。（有向图）

——————————————————————————————————————————

Q：什么是度

A：度即一个顶点所有便的数量，出度是有向图中从当前顶点指向其他顶点的边的数量，入度是有向图中从其他顶点指向当前顶点的边的数量。

——————————————————————————————————————————

Q：邻接矩阵是什么?

A：表示各顶点之间连接关系的矩阵，相连则为1，自连为2，不相连为0

——————————————————————————————————————————

[TOC]

一、GraphX的数据结构

提供分布式的图计算的API、
基于弹性分布式属性图（V+E）（被封装为RDD【】），统一了表视图与图视图
- Q:什么是弹性分布式属性图（Resilient Distributed Property Graph）
- A：顶点和边都带属性的有向多重图

———————————————————————————————————————————

1、数据结构

注：VD 和 ED 是类的泛型，不要混淆为RDD的存储类型

Graph[VD,ED]

class Graph[VD, ED] {

    //基本结构

  val vertices: VertexRDD[VD]

  val edges: EdgeRDD[ED]

  val triplets: RDD[EdgeTriplet[VD, ED]]

    //额外信息

  val numEdges: Long

  val numVertices: Long

  val inDegrees: VertexRDD[Int]

  val outDegrees: VertexRDD[Int]

  val degrees: VertexRDD[Int]}

VertexRDD[VD]

RDD[(VertexId，VD)]

//VertexId:Long的别名

//VD就是顶点数据结构类的泛型

EdgeRDD[ED]

RDD[Edge[ED]]

//Edge 样例类 (srcVid,dstVid,attr:ED)

//ED就是边的数据结构类的泛型

EdgeTriplet[VD,ED]
- 继承自Edge
- 是Edge + srcVertex+desVertex的三元组的RDD ,自动推断的
- ```
srcid，srcattr，dstid，dstattr，attr
```
Edge：
- 样例类case class（src:Long,des:Long,Edata:ED）
VertexId ：Long的别名

import org.apache.spark.graphx.GraphLoader

<dependency>

      <groupId>org.apache.spark</groupId>

      <artifactId>spark-graphx_2.11</artifactId>

      <version>2.2.0</version>

</dependency>

二、图的操作

1、图的创建

图的创建遵循图的数据结构

//通过构造函数建立

import org.apache.spark.graphx._

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] ?

//通过边文件建立

port org.apache.spark.graphx.GraphLoader

//加载边列表文件创建图，文件每行描述一条边，格式：srcId dstId。顶点与边的属性均为1

val graph = GraphLoader.edgeListFile(sc,"file:///opt/spark/data/graphx/followers.txt")

//得到的是一个边和点的属性都为Int:1的一个图

注：所有描述图的RDD内的类型都是泛型的类型，不是指图的结构类型。

2、图的修改

2.1 属性算子：Map

*	仅用于修改图中的顶点或边的属性数据，不能改变ID

*	map返回值可以与旧值不一致

class Graph[VD, ED] {

    //返回值是VD，说明会以返回值替换原Vert中的VD数据而不改变ID

  def mapVertices[VD2](map: (VertexId, VD) =>VD2): Graph[VD2, ED]

    //替换边的属性值

  def mapEdges[ED2](map: Edge[ED] => ED2): Graph[VD, ED2]

    //仅能改变边的属性值

  def mapTriplets[ED2](map: EdgeTriplet[VD, ED] => ED2: Graph[VD, ED2] //Triplets不能修改顶点的泛型

}

//图的map方法返回的是一个有新的泛型类的Graph

demo实例

val t1_graph = tweeter_graph.mapVertices { case(vertextId, (name, age)) => (vertextId, name) }

val t2_graph = tweeter_graph.mapVertices { (vertextId, attr) => (vertextId, attr._1) }

val t3_graph = tweeter_graph.mapEdges(e => Edge(e.srcId, e.dstId, e.attr*7.0))

2.2 结构算子

class Graph[VD, ED] {

  def reverse: Graph[VD, ED] //改变边的方向，调换srcid和dstid

  def subgraph(epred: EdgeTriplet[VD,ED] => Boolean,

               vpred: (VertexId, VD) => Boolean): Graph[VD, ED]

} //epred 边的条件可省略，孤立点会被过滤

2.3 Join算子

柯里化函数
map返回值类型与原VD一直

//按id相等与否join

class Graph[VD, ED] {

   //等值id的join，用结合了新节点的VD来替换旧的VD

   //是个柯里化函数

   //返回值必须与主图的VD一致

  def joinVertices[U](table: RDD[(VertexId, U)])(map: (VertexId, VD, U) => VD): Graph[VD, ED] 

    //不等的id的属性会被补null

  def outerJoinVertices[U, VD2](table: RDD[(VertexId, U)])(map: (VertexId, VD, Option[U]) => VD2)

    : Graph[VD2, ED]

}

demo实例

val tweeters_comps:RDD[(VertexId,String)]= sc.parallelize(Array((1L, "kgc.cn"), (2L, "berkeley.edu"), (3L, "apache.org")))

val t_graph = tweeter_graph.joinVertices(tweeters_comps)((id, v, cmpy) => (v._1 + " @ " + cmpy, v._2))

t_graph.vertices.collect

val s_graph = tweeter_graph.outerJoinVertices(tweeters_comps)((id, v, cmpy) => (v._1 + " @ " + cmpy, v._2))

s_graph.vertices.collect

三、图的应用算法

1、PageRank（PR）算法

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

class Graph[VD, ED] {

  def pageRank(tol: Double, resetProb: Double = 0.15): Graph[Double, Double]

}

val ranks = graph.pageRank(0.0001)

ranks.vertices.sortBy(_._2, false).collect

//res43: Array[(org.apache.spark.graphx.VertexId, Double)] = Array((1,1.7924127957615184), (6,0.9969646507526427), (2,0.9969646507526427), (4,0.9688717814927127), (3,0.6996243163176441), (5,0.5451618049228395))

2、Pregel算法

Pregel是Google提出的用于大规模分布式图计算框架
- 图遍历（BFS）
- 单源最短路径（SSSP）
- PageRank计算
Pregel的计算由一系列迭代组成，称为supersteps
Pregel迭代过程
- 每个顶点从上一个superstep接收入站消息
- 计算顶点新的属性值
- 在下一个superstep中向相邻的顶点发送消息
- 当没有剩余消息时，迭代结束
数据结构：

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

class Graph[VD, ED] {

    def pregel[A](initialMsg: A, maxIterations: Int, activeDirection: EdgeDirection)(

      //根据jmessage决定如何更新自己的VD【value,originValue】

      vprog: (VertexID, VD, A) => VD,

    /**

    *根据每个节点关联的三元组情况，决定要不要发送信息，以及发送什么信息到哪个节点

    *返回的是信息发送目标节点ID和信息message的二元组的迭代器

    */

      sendMsg: EdgeTriplet[VD, ED] => Iterator[(VertexID,A)],

     //迭代归并多个来源的信息，多个变一个

      mergeMsg: (A, A) => A

    ): Graph[VD, ED] //最终返回的是一个 结构不变、值改变 的新图

}

巴特西

Spark + GraphX + Pregel

Spark+GraphX图

一、GraphX的数据结构

1、数据结构

二、图的操作

1、图的创建

2、图的修改

2.1 属性算子：Map

2.2 结构算子

2.3 Join算子

三、图的应用算法

1、PageRank（PR）算法

2、Pregel算法

最新文章

热门文章