Spark GraphX图形数据分析

Spark GraphX图形数据分析

图（Graph）的基本概念

• 图是由顶点集合(vertex)及顶点间的关系集合（边edge）组成的一种网状数据结构
• 图数据很好的表达了数据之间的关系

• 处理的是有向图

图的术语-4

• 出度：指从当前顶点指向其他顶点的边的数量
• 入度：其他顶点指向当前顶点的边的数量

图的经典表示法（了解）

• 邻接矩阵

1、对于每条边，矩阵中相应单元格值为1
2、对于每个循环，矩阵中相应单元格值为2，方便在行或列上求得顶点度数

Spark GraphX 简介

• GraphX特点

1)基于内存实现了数据的复用与快速读取
2)通过弹性分布式属性图（Property Graph）统一了图视图与表视图
3)与Spark Streaming、Spark SQL和Spark MLlib等无缝衔接

GraphX核心抽象

• 弹性分布式属性图
• 顶点和边都是带属性的有向多重图

顶点要带边，两个边构成编号

• 一份物理存储，两种视图（table view,Graph view）

GraphX API

Graph[VD,ED]

VD:顶点的数据类型(二元组)
var rdd=sc.makeRDD(List((1L,"A"),(2L,"B")))
ED:边的数据类型

方法一：spark API
var spark=SparkSession.builder().master("local[2]")
    .appName("hello").getOrCreate();
  val sc=spark.sparkContext
  val verticesRDD=sc.makeRDD(List((1L,1),(2L,2),(3L,3)))
  val edgesRDD=sc.makeRDD(List(Edge(1L,2L,1),Edge(2L,3L,2)))
  val graph=Graph(verticesRDD,edgesRDD)
  graph.vertices.foreach(println(_))
  graph.edges.foreach(println(_))

方法二：spark上运行
import org.apache.spark.graphx.{Edge,Graph}
var spark=SparkSession.builder().master("local[2]")
    .appName("hello").getOrCreate();
  val sc=spark.sparkContext
  val verticesRDD=sc.makeRDD(List((1L,1),(2L,2),(3L,3)))
  val edgesRDD=sc.makeRDD(List(Edge(1L,2L,1),Edge(2L,3L,2)))
  val graph=Graph(verticesRDD,edgesRDD)
  graph.vertices.collect
  graph.edges.collect
  graph.triplets.collect

图的算子-1

属性算子

• 类似于RDD的map操作

对顶点进行遍历，传给你的顶点类型，生成新的顶点

def mapVertices[VD2](map: (VertexId, VD) => VD2)
def mapEdges[ED2](map: Edge[ED] => ED2)

结构算子

• reverse、subgraph

scala> graph1.reverse.triplets.collect
scala> graph1.subgraph(vpred=(id,attr)=>attr._2<30).triplets.collect

图的算子-3

join算子：从外部的RDDs加载数据，修改顶点属性

class Graph[VD, ED] {

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

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

: Graph[VD2, ED]

}

PageRank in GraphX

• PageRank (PR)算法

• 

• 激活（active）和未激活(inactive)

  • 相当于红色为未激活，绿色为激活

  ！！！传过来的值和当前的值做对比，小（接受），大（本身去除）

  1）首先我们默认将2里面（-1）作为旧值

  2）我们将2值传送给7（-1）因为2<7：得出2（7）

  3）我们将7值传送给3（-1）因为7>3：得出3（3）

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

   --------------------------------------------------------------------------------------------------------------

 def pregel[A](initialMsg: A, maxIterations: Int, activeDirection: EdgeDirection)(
      vprog: (VertexID, VD, A) => VD,
      sendMsg: EdgeTriplet[VD, ED] => Iterator[(VertexID,A)],
      mergeMsg: (A, A) => A)
    : Graph[VD, ED]

 -----------------------------------------------------------------------------------------------------------------------

实例一：
入度:多个b对象指向a
case class User(name:String,ilike:Int,follow:Int)
import org.apache.spark.graphx._
//原始图
var points=sc.makeRDD(Array((1L,"zs"),(2L,"ls")))
var edges = sc.makeRDD(Array(Edge(2L,1L,1)))
var graph=Graph(points,edges)
//改变点的信息的结构
var newGraph=graph.mapVertices(
(id,name)=>User(name,0,0))
newGraph.inDegrees.collect//这是一张表
//将新节点表和入度表联合outerjoin
val nnGraph=newGraph.outerJoinVertices(newGraph.inDegrees)((id,lf,rf)=>User(lf.name,lf.ilike,rf.getOrElse(0)))
scala> nnGraph.pageRank(0.01).triplets.collect