GraphX Programming Guide

• 概述
• 入门
• 属性 Graph
  • 示例属性 Graph
• Graph 运算符
  • 运算符的汇总表
  • Property 运算符
  • Structural 运算符
  • Join 运算符
  • 邻域聚合
    • 聚合消息 (aggregateMessages)
    • Map Reduce Triplets Transition Guide (Legacy)
    • 计算级别信息
    • 收集相邻点
  • Caching and Uncaching
• Pregel API
• Graph 建造者
• Vertex and Edge RDDs
  • VertexRDDs
  • EdgeRDDs
• 优化表示
• Graph 算法
  • PageRank
  • 连接组件
  • Triangle 计数
• 示例

概述

GraphX 是 Spark 中用于图形和图形并行计算的新组件。在高层次上， GraphX 通过引入一个新的图形抽象来扩展 Spark RDD ：一种具有附加到每个顶点和边缘的属性的定向多重图形。为了支持图形计算，GraphX 公开了一组基本运算符（例如： subgraph ，joinVertices 和 aggregateMessages）以及 Pregel API 的优化变体。此外，GraphX 还包括越来越多的图形算法 和 构建器，以简化图形分析任务。

入门

首先需要将 Spark 和 GraphX 导入到项目中，如下所示：

import org.apache.spark._
import org.apache.spark.graphx._
// To make some of the examples work we will also need RDD import org.apache.spark.rdd.RDD

如果您不使用 Spark 外壳，您还需要一个 SparkContext。要了解有关 Spark 入门的更多信息，请参考 Spark快速入门指南。

属性 Graph

属性 Graph 是一个定向多重图形，用户定义的对象附加到每个顶点和边缘。定向多图是具有共享相同源和目标顶点的潜在多个平行边缘的有向图。支持平行边缘的能力简化了在相同顶点之间可以有多个关系（例如： 同事和朋友）的建模场景。每个顶点都由唯一的64位长标识符（ VertexId ）键入。 GraphX 不对顶点标识符施加任何排序约束。类似地，边缘具有对应的源和目标顶点标识符。

属性图是通过 vertex (VD)和 edge (ED) 类型进行参数化的。这些是分别与每个顶点和边缘相关联的对象的类型。

当它们是原始数据类型（例如： int ，double 等等）时，GraphX 优化顶点和边缘类型的表示，通过将其存储在专门的数组中来减少内存占用。

在某些情况下，可能希望在同一个图形中具有不同属性类型的顶点。这可以通过继承来实现。例如，将用户和产品建模为二分图，我们可能会执行以下操作：

class VertexProperty()
case class UserProperty(val name: String) extends VertexProperty case class ProductProperty(val name: String, val price: Double) extends VertexProperty // The graph might then have the type: var graph: Graph[VertexProperty, String] = null

像 RDD 一样，属性图是不可变的，分布式的和容错的。通过生成具有所需更改的新图形来完成对图表的值或结构的更改。请注意，原始图形的大部分（即，未受影响的结构，属性和索引）在新图表中重复使用，可降低此内在功能数据结构的成本。使用一系列顶点分割启发式方法，在执行器之间划分图形。与 RDD 一样，在发生故障的情况下，可以在不同的机器上重新创建图形的每个分区。

逻辑上，属性图对应于一对编码每个顶点和边缘的属性的类型集合( RDD )。因此，图类包含访问图形顶点和边的成员：

class Graph[VD, ED] { val vertices: VertexRDD[VD] val edges: EdgeRDD[ED] }

VertexRDD[VD] 和 EdgeRDD[ED] 分别扩展了 RDD[(VertexId, VD)] 和 RDD[Edge[ED]] 的优化版本。 VertexRDD[VD] 和 EdgeRDD[ED] 都提供了围绕图形计算和利用内部优化的附加功能。 我们在顶点和边缘 RDD 部分更详细地讨论了 VertexRDD 和 EdgeRDD API，但现在它们可以被认为是 RDD[(VertexId, VD)] 和 RDD[Edge[ED]] 的简单 RDD。

示例属性 Graph

假设我们要构建一个由 GraphX 项目中的各种协作者组成的属性图。顶点属性可能包含用户名和职业。我们可以用描述协作者之间关系的字符串来注释边：

生成的图形将具有类型签名：

val userGraph: Graph[(String, String), String]

从原始文件， RDD 甚至合成生成器构建属性图有许多方法，这些在图形构建器的一节中有更详细的讨论 。最普遍的方法是使用 Graph 对象。例如，以下代码从 RDD 集合中构建一个图：

// Assume the SparkContext has already been constructed
val sc: SparkContext // Create an RDD for the vertices val users: RDD[(VertexId, (String, String))] = sc.parallelize(Array((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")), (5L, ("franklin", "prof")), (2L, ("istoica", "prof")))) // Create an RDD for edges val relationships: RDD[Edge[String]] = sc.parallelize(Array(Edge(3L, 7L, "collab"), Edge(5L, 3L, "advisor"), Edge(2L, 5L, "colleague"), Edge(5L, 7L, "pi"))) // Define a default user in case there are relationship with missing user val defaultUser = ("John Doe", "Missing") // Build the initial Graph val graph = Graph(users, relationships, defaultUser)

在上面的例子中，我们使用了 Edge case 类。边缘具有 srcId 和 dstId 对应于源和目标顶点标识符。此外， Edge 该类有一个 attr 存储边缘属性的成员。

我们可以分别使用 graph.vertices 和 graph.edges 成员将图形解构成相应的顶点和边缘视图。

val graph: Graph[(String, String), String] // Constructed from above // Count all users which are postdocs graph.vertices.filter { case (id, (name, pos)) => pos == "postdoc" }.count // Count all the edges where src > dst graph.edges.filter(e => e.srcId > e.dstId).count

注意， graph.vertices 返回一个 VertexRDD[(String, String)] 扩展 RDD[(VertexId, (String, String))] ，所以我们使用 scala case表达式来解构元组。另一方面， graph.edges 返回一个 EdgeRDD 包含 Edge[String] 对象。我们也可以使用 case 类型构造函数，如下所示：

graph.edges.filter { case Edge(src, dst, prop) => src > dst }.count

除了属性图的顶点和边缘视图之外， GraphX 还暴露了三元组视图。三元组视图逻辑上连接顶点和边缘属性，生成 RDD[EdgeTriplet[VD, ED]] 包含 EdgeTriplet 该类的实例。此 连接可以用以下SQL表达式表示：

SELECT src.id, dst.id, src.attr, e.attr, dst.attr FROM edges AS e LEFT JOIN vertices AS src, vertices AS dst ON e.srcId = src.Id AND e.dstId = dst.Id

或图形为：

EdgeTriplet 类通过分别添加包含源和目标属性的 srcAttr 和 dstAttr 成员来扩展 Edge 类。 我们可以使用图形的三元组视图来渲染描述用户之间关系的字符串集合。

val graph: Graph[(String, String), String] // Constructed from above // Use the triplets view to create an RDD of facts. val facts: RDD[String] = graph.triplets.map(triplet => triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1) facts.collect.foreach(println(_))

Graph 运算符

正如 RDDs 有这样的基本操作 map， filter， 以及 reduceByKey，性能图表也有采取用户定义的函数基本运算符的集合，产生具有转化特性和结构的新图。定义了优化实现的核心运算符，并定义了 Graph 表示为核心运算符组合的方便运算符 GraphOps 。不过，由于 Scala 的含义，操作员 GraphOps 可自动作为成员使用 Graph 。例如，我们可以通过以下方法计算每个顶点的入度（定义 GraphOps ）：

val graph: Graph[(String, String), String] // Use the implicit GraphOps.inDegrees operator val inDegrees: VertexRDD[Int] = graph.inDegrees

区分核心图形操作的原因 GraphOps 是能够在将来支持不同的图形表示。每个图形表示必须提供核心操作的实现，并重用许多有用的操作 GraphOps。

运算符的汇总表

以下是两个定义的功能的简要摘要，但为简单起见 Graph， GraphOps 它作为 Graph 的成员呈现。请注意，已经简化了一些功能签名（例如，删除了默认参数和类型约束），并且已经删除了一些更高级的功能，因此请参阅 API 文档以获取正式的操作列表。

/** Summary of the functionality in the property graph */
class Graph[VD, ED] { // Information about the Graph =================================================================== val numEdges: Long val numVertices: Long val inDegrees: VertexRDD[Int] val outDegrees: VertexRDD[Int] val degrees: VertexRDD[Int] // Views of the graph as collections ============================================================= val vertices: VertexRDD[VD] val edges: EdgeRDD[ED] val triplets: RDD[EdgeTriplet[VD, ED]] // Functions for caching graphs ================================================================== def persist(newLevel: StorageLevel = StorageLevel.MEMORY_ONLY): Graph[VD, ED] def cache(): Graph[VD, ED] def unpersistVertices(blocking: Boolean = true): Graph[VD, ED] // Change the partitioning heuristic ============================================================ def partitionBy(partitionStrategy: PartitionStrategy): Graph[VD, ED] // Transform vertex and edge attributes ========================================================== def mapVertices[VD2](map: (VertexId, VD) => VD2): Graph[VD2, ED] def mapEdges[ED2](map: Edge[ED] => ED2): Graph[VD, ED2] def mapEdges[ED2](map: (PartitionID, Iterator[Edge[ED]]) => Iterator[ED2]): Graph[VD, ED2] def mapTriplets[ED2](map: EdgeTriplet[VD, ED] => ED2): Graph[VD, ED2] def mapTriplets[ED2](map: (PartitionID, Iterator[EdgeTriplet[VD, ED]]) => Iterator[ED2]) : Graph[VD, ED2] // Modify the graph structure ==================================================================== def reverse: Graph