“像顶点一样思考”

屏幕快照 2017-11-14 18.54.08.png

​
        Spark GraphX是一个分布式图处理框架，Spark GraphX基于Spark平台提供对图计算和图挖掘简洁易用的而丰富多彩的接口，极大的方便了大家对分布式图处理的需求。Spark GraphX由于底层是基于Spark来处理的，所以天然就是一个分布式的图处理系统。图的分布式或者并行处理其实是把这张图拆分成很多的子图，然后我们分别对这些子图进行计算，计算的时候可以分别迭代进行分阶段的计算，即对图进行并行计算。
​
       设计GraphX时，点分割和GAS都已成熟，在设计和编码中针对它们进行了优化，并在功能和性能之间寻找最佳的平衡点。如同Spark本身，每个子模块都有一个核心抽象。GraphX的核心抽象是Resilient Distributed Property Graph，一种点和边都带属性的有向多重图。它扩展了Spark RDD的抽象，有Table和Graph两种视图，而只需要一份物理存储。两种视图都有自己独有的操作符，从而获得了灵活操作和执行效率。

相关知识

• Scala
  • 对象-函数式编程语言
  • 类型推断
  • 类的申明
  • map和reduce
  • 一切都是函数
  • 与java的互操作性
• Spark
  • 分布式内存数据：RDD
  • 集群和术语解释
• 图术语
  • 基础图： 有向图和无向图、有环图和无环图、有标签的图和无标签的图、二分图
  • RDF图和属性图
  • 邻接矩阵

GraphX基础

类成员

       在GraphX中，图的基础类为Garph，它包含两个RDD：一个为边RDD，另一个为顶点RDD。可以用给定的边RDD和顶点RDD构建一个图。一旦构建好图，就可以用edges()和vertices()来访问边和顶点的集合。VD和ED代表了用户自定义的顶点和边类，对应的图是参数化类型的泛类型Graph[VD,ED]。GraphX中图必须要有顶点和边属性。GraphX中Vertice和Edge持有VerticeId值，而不是顶点的引用。图在集群中是分布式存储的，不属于单个JVM，因此一条边的顶点可能在不同的集群节点上。

• 顶点： Vertice(VertexId, VD)       

  abstract class VertexRDD[VD] extends RDD[(VertexId, VD)]
  

  • 抽象值成员
    innerJoin leftJoin mapValues ···
  • 具体值成员
    collect count distinct filter foreach groupBy isEmpty persist map reduce sortBy toString ···

• 边： Edge(VertexId, VertexId, ED）   

  class Edge[ED](srcId:VertexId, dstId:VertexId, attire:E
  

  abstract class EdgeRDD[ED] extends RDD[Edge[ED]]
  

  • 抽象值成员
    innerJoin mapValues reverse
  • 具体值成员
    ++ aggregate cache collect count distinct filter foreach groupBy isEmpty map persist reduce sortBy toString ···

  class EdgeTriplet[VD, ED] extends Edge[ED]
  

  • 值成员
    Attr srcId srcAttr dstId dstAttr

• 图： Graph(VD, ED)       

  abstract class Graph[VD,ED] extend Serializable
  

  • 抽象值成员
    cache edges mapEdges mapTriplets mapVertices mask outerJoinVertices persist reverse subgraph triplets vertices ···
  • 具体值成员
    aggregateMessages mapEdges mapTriplets ···

  class GraphOps[VD,ED] extends Serializable
  

  • 值成员
    collectEdges collectNeiborIds collectNeibors degrees filter inDegrees joinVertices numEdges numVertices outDegrees pageRank personalizedPageRank pickRandomVertex pregel triangleCount ···

GraphX实例

图片 1.png

• 引用

  import org.apache.spark._
  import org.apache.spark.graphx._
  import org.apache.spark.rdd.RDD
  

• 构图

         有很多方式从一个原始文件、RDD构造一个属性图。最一般的方法是利用Graph object。 下面的代码从RDD集合生成属性图。

  // 假设SparkContext已经被构造
  val sc: SparkContext
  // 创建点RDD
  val users: RDD[(VertexId, (String, String))] =sc.parallelize(
                            Array((3L, ("rxin", "student")),
                                  (7L, ("jgonzal","postdoc")),
                                  (5L, ("franklin", "prof")), 
                                  (2L, ("istoica", "prof"))))
  // 创建边RDD
  val relationships: RDD[Edge[String]] =  sc.parallelize(
                             Array(Edge(3L, 7L, "collab"),    
                                   Edge(5L, 3L, "advisor"),
                                   Edge(2L, 5L, "colleague"), 
                                   Edge(5L, 7L, "pi")
                                   Edge(5L, 0L, "colleague")))
  // 定义一个默认用户，避免有不存在用户的关系
  val defaultUser = ("John Doe", "Missing")
  // 构造Graph
  val graph = Graph(users, relationships, defaultUser）
  

• 缓存

    //缓存。默认情况下,缓存在内存的图会在内存紧张的时候被强制清理，采用的是LRU算法
    graph.cache()
    graph.persist(StorageLevel.MEMORY_ONLY)
    graph.unpersistVertices(true)
  

• 点、边和三元组

        下面的代码用到了Edge样本类。边有一个srcId和dstId分别对应于源和目标顶点的标示符。另外，Edge类有一个attr成员用来存储边属性。可以分别用graph.vertices和graph.edges成员将一个图解构为相应的顶点和边。graph.vertices返回一个VertexRDD[(String, String)]，它继承于 RDD[(VertexID, (String, String))]。所以我们可以用scala的case表达式解构这个元组。另一方面，graph.edges返回一个包含Edge[String]对象的EdgeRDD，我们也可以用到case类的类型构造器。

         除了属性图的顶点和边视图，GraphX也包含了一个三元组视图，三元视图逻辑上将顶点和边的属性保存为一个RDD[EdgeTriplet[VD, ED]]，它包含EdgeTriplet类的实例。EdgeTriplet类继承于Edge类，并且加入了srcAttr和dstAttr成员，这两个成员分别包含源和目的的属性。我们可以用一个三元组视图渲染字符串集合用来描述用户之间的关系。

  // 找出职业为postdoc的人
  graph.vertices.filter { case (id, (name, pos)) => pos == "postdoc" }.collect
  // 计算源顶点ID大于目标顶点ID的边的数量
  graph.edges.filter(e => e.srcId > e.dstId).count
  graph.edges.filter { case Edge(src, dst, prop) => src > dst }.count
  // 使用三元组视图描述关系事实
  val facts: RDD[String] = graph.triplets.map(triplet =>triplet.srcAttr._1 + 
                     " is the " + triplet.attr + " of " + triplet.dstAttr._1)
  facts.collect.foreach(println(_))
  

• 度、入度、出度

         正如RDDs有基本的操作map, filter和reduceByKey一样，属性图也有基本的集合操作，这些操作采用用户自定义的函数并产生包含转换特征和结构的新图。定义在Graph中的 核心操作是经过优化的实现。表示为核心操作的组合的便捷操作定义在GraphOps中。然而， 因为有Scala的隐式转换，定义在GraphOps中的操作可以作为Graph的成员自动使用。例如，我们可以通过下面的方式计算每个顶点(定义在GraphOps中)的入度。区分核心图操作和GraphOps的原因是为了在将来支持不同的图表示。每个图表示都必须提供核心操作的实现并重用很多定义在GraphOps中的有用操作。

  val degrees: VertexRDD[Int] = graph.degrees;
  degrees.collect().foreach(println)
  val inDegrees: VertexRDD[Int] = graph.inDegrees
  inDegrees.collect().foreach(println)
  val outDegrees: VertexRDD[Int] = graph.outDegrees
  outDegrees.collect().foreach(println)
  

• 属性操作：修改顶点和边的属性

         属性操作每个操作都产生一个新的图，这个新的图包含通过用户自定义的map操作修改后的顶点或边的属性。Map操作根据原图的一些特性得到新图，原图结构是不变的。这些操作的一个重要特征是它允许所得图形重用原有图形的结构索引(indices)。下面的两行代码在逻辑上是等价的，但是第一个不是图操作，它不保存结构索引，所以不会从GraphX系统优化中受益。Map操作根据原图的一些特性得到新图，原图结构是不变的。这些操作经常用来初始化的图形，用作特定计算或者用来处理项目不需要的属性。例如，给定一个图，这个图的顶点特征包含出度，我们为PageRank初始化它。

  • map 操作

  //顶点转换，顶点age+1
  //RDD操作，再构造新图，不保存结构索引，不会被系统优化
  val newVertices = graph.vertices.map { case (id, attr) => 
                            (id, (attr._1 + "-1", attr._2 + "-2")) }
  val newGraph1 = Graph(newVertices, graph.edges)
  //图Map操作，被系统优化  
  val newGraph2 = graph.mapVertices((id, attr) => 
                             (id, (attr._1 + "-1", attr._2 + "-2")))
  

  • Join 操作

  //构造一个新图，顶点属性是出度
  val inputGraph: Graph[Int, String] = graph.outerJoinVertices(
                 graph.outDegrees)((vid, _, degOpt) => degOpt.getOrElse(0))
  //根据顶点属性为出度的图构造一个新图，依据PageRank算法初始化边与点
  val outputGraph: Graph[Double, Double] =inputGraph.mapTriplets(
                triplet => 1.0 / triplet.srcAttr).mapVertices((id, _) => 1.0)
  

  • 自定义类型

  //创建一个新图，顶点 VD 的数据类型为 User，并从 graph 做类型转换
  case class User(name: String, pos: String, inDeg: Int, outDeg: Int)
  val initialUserGraph: Graph[User, String] = graph.mapVertices { 
                  case (id, (name, age)) => User(name, pos, 0, 0)}
  //initialUserGraph 与 inDegrees、outDegrees（RDD）进行连接，并修改 initialUserGraph中 inDeg 值、outDeg 值
  val userGraph = initialUserGraph.outerJoinVertices(
        initialUserGraph.inDegrees) {
           case (id, u, inDegOpt) => 
                User(u.name, u.pos, inDegOpt.getOrElse(0), u.outDeg)
           }.outerJoinVertices(
        initialUserGraph.outDegrees) {
           case (id, u, outDegOpt) => 
                User(u.name, u.pos, u.inDeg, outDegOpt.getOrElse(0))
        }
  userGraph.vertices.collect.foreach(v => 
           println(s"${v._2.name} inDeg: ${v._2.inDeg} outDeg: ${v._2.outDeg}"))
  //出度和入读相同的人员
  userGraph.vertices.filter {
        case (id, u) => u.inDeg == u.outDeg
      }.collect.foreach {
        case (id, property) => println(property.name)
      }
  

• 结构操作

  • 子图

  //由已定义的顶点构成的子图
  val subGraph = graph.subgraph(vpred = (id, attr) => attr._2 != "Missing" )
  subGraph.vertices.collect().foreach(println(_))
  subGraph.triplets.map(triplet => 
                 triplet.srcAttr._1 + " is the " + triplet.attr + 
                 " of " + triplet.dstAttr._1).collect().foreach(println(_))
  

  • 图反向

  //图的反向操作，新的图形的所有边的方向相反，不修改顶点或边性属性、不改变的边的数目，它可以有效地实现不必要的数据移动或复制        
    var rGraph = graph.reverse
  

  • Mask

  //Mask操作也是根据输入图构造一个新图，达到一个限制制约的效果
  val ccGraph = graph.connectedComponents()
  val validGraph = graph.subgraph(vpred = (id, attr) => attr._2 != "Missing")
  val validCCGraph = ccGraph.mask(validGraph)
  

• 聚合操作

  //计算年龄大于自己的关注者的总人数和总年龄
  val olderFollowers: VertexRDD[(Int, Double)] = 
                           graph.mapReduceTriplets[(Int, Double)](
    //Map函数
      triplet => {
        if (triplet.srcAttr > triplet.dstAttr) {
          Iterator((triplet.dstId, (1, triplet.srcAttr)))
        } else {
          Iterator.empty
        }
      },
      //Reduce函数
      (a, b) => (a._1 + b._1, a._2 + b._2)
    )
  //计算年龄大于自己的关注者的平均年龄
  val avgAgeOfOlderFollowers: VertexRDD[Double] =
          olderFollowers.mapValues((id, value) => 
               value match {case (count, totalAge) => totalAge / count })
  avgAgeOfOlderFollowers.collect.foreach(println(_))
    
  //定义一个Reduce函数来计算图中较大度的点
  def max(a: (VertexId, Int), b: (VertexId, Int)): (VertexId, Int) = {
      if (a._2 > b._2) a else b
    }
  val maxInDegree: (VertexId, Int) = graph.inDegrees.reduce(max)
  println(s"maxInDegree: $maxInDegree")
  val maxOutDegree: (VertexId, Int) = graph.outDegrees.reduce(max)
  println(s"maxOutDegree: $maxOutDegree")
  val maxDegrees: (VertexId, Int) = graph.degrees.reduce(max)
  println(s"maxDegrees: $maxDegrees")
  

• 相邻聚合

  //计算邻居相关函数，这些操作是相当昂贵的，需要大量的重复信息作为他们的通信，因此相同的计算还是推荐用mapReduceTriplets 
  val neighboorIds:VertexRDD[Array[VertexId]] =
                        graph.collectNeighborIds(EdgeDirection.Out)
  val neighboors:VertexRDD[Array[(VertexId, Double)]]=
                            graph.collectNeighbors(EdgeDirection.Out);
  

• Pregel API

  //Pregel API。计算单源最短路径
  //通过GraphGenerators构建一个随机图
  val numVertices = 100
  val numEParts = 2
  val mu = 4.0
  val sigma = 1.3
  val graph1 = GraphGenerators.logNormalGraph(sc, numVertices, 
                    numEParts, mu, sigma).mapEdges(e=> e.attr.toDouble)
  //定义一个源值 点
  val sourceId: VertexId = 42
  //初始化图的所有点，除了与指定的源值点相同值的点为0.0以外，其他点为无穷大
  val initialGraph = graph1.mapVertices((id, _) => 
                   if (id == sourceId) 0.0 else Double.PositiveInfinity)
  //Pregel有两个参数列表，第一个参数列表包括的是：初始化消息、迭代较大数、边的方向(Out)。第二个参数列表包括的是：用户定义的接受消息、计算消息、联合合并消息的函数。
  val sssp = initialGraph.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))
        } else {
          Iterator.empty
        }
      },
      //合并消息
      (a, b) => math.min(a, b)
      )
  println(sssp.vertices.collect.mkString("\n"))
  

• 主要图算法

  • PageRank

  val pageRankGraph = graph1.pageRank(0.001)
  pageRankGraph.vertices.sortBy(_._2，false).saveAsTextFile(
                            "/user/hadoop/data/temp/graph/graph.pr")
  pageRankGraph.vertices.top(5)(Ordering.by(_._2)).foreach(println)
  

  • Connected Components

  val connectedComponentsGraph = graph1.connectedComponents()
  connectedComponentsGraph.vertices.sortBy(_._2, false).saveAsTextFile(
                                "/user/hadoop/data/temp/graph/graph.cc")
  connectedComponentsGraph.vertices.top(5)(Ordering.by(_._2)).foreach(println)
  

  • TriangleCount

  //TriangleCount主要用途之一是用于社区发现 保持sourceId小于destId
  val graph2 = GraphLoader.edgeListFile(sc, path, true)
  val triangleCountGraph = graph2.triangleCount()
  triangleCountGraph.vertices.sortBy(_._2,false).saveAsTextFile(
                               "/user/hadoop/data/temp/graph/graph.tc")
  triangleCountGraph.vertices.top(5)(Ordering.by(_._2)).foreach(println)
  

• 其他操作

  • GraphLoader构建Graph

  var path = "/user/Hadoop/data/temp/graph/graph.txt"
  var minEdgePartitions = 1
  var canonicalOrientation = false // if sourceId < destId this value is true
  val graph1 = GraphLoader.edgeListFile(sc, path, canonicalOrientation,
          minEdgePartitions,StorageLevel.MEMORY_ONLY, StorageLevel.MEMORY_ONLY)
  

  • 随机图

  //通过GraphGenerators构建一个随机图
  val numVertices = 100
  val numEParts = 2
  val mu = 4.0
  val sigma = 1.3
  val graph: Graph[Double, Int] = GraphGenerators.logNormalGraph(sc, 
        numVertices, numEParts, mu, sigma).mapVertices((id, _) => id.toDouble)
  graph.triplets.collect.foreach(triplet => 
          println(triplet.srcId + "-" + triplet.srcAttr + "-" + triplet.attr +
                     "-" + triplet.dstId + "-" + triplet.dstAttr))
  

  • aggregateUsingIndex操作

  val setA: VertexRDD[Int] = VertexRDD(
           sc.parallelize(0L until 100L).map(id => (id, 1)))
  val rddB: RDD[(VertexId, Double)] = sc.parallelize(
           0L until 100L).flatMap(id => List((id, 1.0), (id, 2.0)))
  val setB: VertexRDD[Double] = setA.aggregateUsingIndex(rddB, _ + _)
  val setC: VertexRDD[Double] = setA.innerJoin(setB)((id, a, b) => a + b)
  

参考资料

• 《Spark GraphX实战》
• 《Spark入门实战系列--9.Spark图计算GraphX介绍及实例》
• 《Spark中文手册8：spark GraphX编程指南》
• 《Spark_GraphX大规模图计算和图挖掘V3.0》