码迷,mamicode.com
首页 > 其他好文 > 详细

Spark GraphX图形数据分析

时间:2019-08-11 01:19:11      阅读:201      评论:0      收藏:0      [点我收藏+]

标签:mic   xid   ever   属性   hello   编号   spark   sql   foreach   

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

 

Spark GraphX图形数据分析

标签:mic   xid   ever   属性   hello   编号   spark   sql   foreach   

原文地址:https://www.cnblogs.com/tudousiya/p/11333440.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!