Spark GraphX（一）

桃扇骨 2022-08-10 03:39 217阅读 0赞

# 1.基础 #

Spark中属性图是由VertexRDD和EdgeRDD两个参数构成的。其中，每个vertex由一个唯一的64位长的标识符（VertexId）作为key。同时，属性图也和RDD一样，是不可变的、分布式的、可容错的。属性图Graph的定义如下：

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

即Graph中包含了三个属性点集、边集、triplets集（三元组）。可以用图形象的表示为：

![Center][]

*  Vertices：由VertexId（Long类型）、attribute（属性描述或距离）构成。如，(3L, ("San Francisco", "CA"))，（1L, 10）。
 *  Edges：由srcId（起始结点VertexId）、dstId（终结点VertexId）、attribute（边的权值）构成。如，Edge(1L, 2L, 20)
 *  Triplets：由srcId、srcAttr（起始Vertex）和dstId、dstAttr（终止Vertex），以及attr构成。如，((1, (Santa Clara, CA)), (2, (Fremont, CA)), 20)

# 2.操作符 #

正如RDDs有基本的操作map、filter和reduceByKey一样，Graph也有基本操作，这些操作采用用户自定义的函数并产生包含转换特征和结构的新图。定义在Graph中的核心操作时经过优化的实现，表示核心操作的组合的便捷操作定义在GraphOps中。因为Scala的隐式转换，定义在GraphOps中的操作可以作为Graph的成员自动使用。例如，初始化Graph的时候调用mapVertices：

val initialGraph = graph.mapVertices((id, _) => if (id == sourceId) 0.0 else Double.PositiveInfinity)

Graph有很多操作，具体的可以参考Spark API。

# 3.Pregel #

在GraphX中，更高级的Pregel操作是一个约束到图拓扑的批量同步（bulk-synchronous）并行消息抽象。Pregel操作者执行一系列的超级步骤（super steps），在这些步骤中，顶点从之前的超级步骤中接收进入（inbound）消息的总和，为顶点属性计算一个新的值，然后在以后的超级步骤中发送消息到邻居顶点。不像Pregel而更像GraphLab，消息作为一个边为三元组的函数被并行计算，消息计算既访问了源顶点特征也访问了目的顶点特征。在super steps中，没有收到消息的顶点被跳过。当没有消息遗留时，Pregel操作停止迭代并返回最终的图。

与更标准的Pregel实现不同的是，GraphX中的顶点仅仅能发送消息给邻居顶点，并利用用户自定义的消息函数构造消息。这些限制允许在GraphX进行额外的优化。

Pregel有两个参数列表（graph.pregel(argslist)(argslist)）。第一个参数列表包含配置参数初始消息、最大迭代次数、发送消息的边的方向（默认是沿边的方向）。第二个参数列表包含用户自定义的函数用来接收消息（vprog）、计算消息（sendMsg）、合并消息（mergeMsg）。

而Pregel处理的数据流（Dataflow）是：1.每一次迭代计算从计算指定点的邻结点和出边开始；2.使用triplets视图，重新计算每个triplet的消息，然后在终结点合并消息；3.在所有顶点，信息被vertex programs收到。在伯克利的论文《 **GraphX: Graph Processing in a Distributed Dataflow Framework**》中，关于Graph计算的数据流的描述如下：

> *Each iteration begins by executing the join stage to bind active vertices with their outbound edges. Using the triplets view, messages are computed along each triplet in a map stage and then aggregated at their destination vertex in a groupby stage. Finally, the messages are received by the vertex programs in a map stage over the vertices.*

在Google关于Pregel的论文《**Pregel: A System for Large-Scale Graph Processing**》中，举了一个图形化的例子描述如下：

![Center 1][]

pregel的定义如下：

def pregel[A: ClassTag](
    	initialMsg: A,
    	maxIterations: Int = Int.MaxValue,
    	activeDirection: EdgeDirection = EdgeDirection.Either)(
    	vprog: (VertexId, VD, A) => VD,
    	sendMsg: EdgeTriplet[VD, ED] => Iterator[(VertexId, A)],
    	mergeMsg: (A, A) => A)

例一，我们可以用Pregel操作表达计算单源最短路径( single source shortest path)。

import org.apache.spark.graphx._
    // Import random graph generation library
    import org.apache.spark.graphx.util.GraphGenerators
    // A graph with edge attributes containing distances
    val graph: Graph[Int, Double] = GraphGenerators.logNormalGraph(sc, numVertices = 100).mapEdges(e => e.attr.toDouble)
    val sourceId: VertexId = 42 // The ultimate source
    // Initialize the graph such that all vertices except the root have distance infinity.
    val initialGraph = graph.mapVertices((id, _) => if (id == sourceId) 0.0 else Double.PositiveInfinity)
    val sssp = initialGraph.pregel(Double.PositiveInfinity)(
      vprog = (id, dist, newDist) => math.min(dist, newDist), // Vertex Program
      sendMsg = triplet => {  // Send Message
    				if (triplet.srcAttr + triplet.attr < triplet.dstAttr) {
    				  Iterator((triplet.dstId, triplet.srcAttr + triplet.attr))
    				} else {
    				  Iterator.empty
    				}
    			},
      mergeMsg = (a,b) => math.min(a,b) // Merge Message
      )
    println(sssp.vertices.collect.mkString("\n"))

例二，用Pregel实现广度优先遍历（Breadth First Search）

import org.apache.spark.SparkConf
    import org.apache.spark.SparkContext
    import org.apache.spark.graphx.GraphLoader
    
    /**
     * @author Administrator
     */
    object BFS {
    
      def main(args: Array[String]): Unit = {
    
        if(args.length != 4){
            System.err.println("Usage: BFS <input file> <output file> <source vertex> <iteration number>")
            System.exit(1)
        }
    
        val sAllTime = System.currentTimeMillis()
        val conf = new SparkConf()
        val sc = new SparkContext(conf)
        val fname = args(0)
        val outPath = args(1)
        val srcVertex = args(2).toInt
        val numIter = args(3).toInt
    
        val sLoadTime = System.currentTimeMillis()
        val graphFile = GraphLoader.edgeListFile(sc, fname).cache()
        val eLoadTime = System.currentTimeMillis()
    
        val graph = graphFile.mapVertices((id, _) => if (id == srcVertex) 0.0 else Double.PositiveInfinity)
        val sComTime = System.currentTimeMillis()
        val bfs = graph.pregel(Double.PositiveInfinity, numIter)(
            vprog = (id, attr, msg) => math.min(attr, msg),
            sendMsg = triplet => {
                if (triplet.srcAttr != Double.PositiveInfinity) {
                  Iterator((triplet.dstId, triplet.srcAttr+1))
                  }
                else {
                  Iterator.empty
                  }
                },
            mergeMsg = (a,b) => math.min(a,b) )
        val eComTime = System.currentTimeMillis()
        //bfs.vertices.saveAsTextFile(outPath)
    
        sc.stop()
        val eAllTime = System.currentTimeMillis()
        println("Load time: " + (eLoadTime - sLoadTime) / 1000)
        println("Compute time: " + (eComTime - sComTime) / 1000)
        println("Total time: " + (eAllTime - sAllTime) / 1000)
      }
    }

[Center]: /images/20220731/c5f1c889ce4344448818439a8e987695.png
[Center 1]: /images/20220731/f95941697031476098bd758f5fa68cf5.png