扫二维码与项目经理沟通
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流
本篇内容主要讲解“GraphX的基础知识有哪些”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“GraphX的基础知识有哪些”吧!
创新互联公司自2013年创立以来，先为徐水等服务建站，徐水等地企业，进行企业商务咨询服务。为徐水企业网站制作PC+手机+微官网三网同步一站式服务解决您的所有建站问题。
Spark GraphX是一个分布式图处理框架，Spark GraphX基于Spark平台提供对图计算和图挖掘简洁易用的而丰富多彩的接口，极大的方便了大家对分布式图处理的需求。Spark GraphX由于底层是基于Spark来处理的，所以天然就是一个分布式的图处理系统。图的分布式或者并行处理其实是把这张图拆分成很多的子图，然后我们分别对这些子图进行计算，计算的时候可以分别迭代进行分阶段的计算，即对图进行并行计算。
设计GraphX时，点分割和GAS都已成熟，在设计和编码中针对它们进行了优化，并在功能和性能之间寻找最佳的平衡点。如同Spark本身，每个子模块都有一个核心抽象。GraphX的核心抽象是Resilient Distributed Property Graph，一种点和边都带属性的有向多重图。它扩展了Spark RDD的抽象，有Table和Graph两种视图，而只需要一份物理存储。两种视图都有自己独有的操作符，从而获得了灵活操作和执行效率。
类成员
在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]]
class EdgeTriplet[VD, ED] extends Edge[ED]
值成员Attr srcId srcAttr dstId dstAttr
抽象值成员innerJoin mapValues reverse
具体值成员++ aggregate cache collect count distinct filter foreach groupBy isEmpty map persist reduce sortBy toString ···
图： Graph(VD, ED)
abstract class Graph[VD,ED] extend Serializable
class GraphOps[VD,ED] extends Serializable
值成员collectEdges collectNeiborIds collectNeibors degrees filter inDegrees joinVertices numEdges numVertices outDegrees pageRank personalizedPageRank pickRandomVertex pregel triangleCount ···
抽象值成员cache edges mapEdges mapTriplets mapVertices mask outerJoinVertices persist reverse subgraph triplets vertices ···
具体值成员aggregateMessages mapEdges mapTriplets ···
GraphX实例
引用
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初始化它。
//顶点转换，顶点age+1
//RDD操作，再构造新图，不保存结构索引，不会被系统优化
val newVertices = graph.vertices.map { case (id, attr) =>
(id, (attr._1 + "-1", attr._2 + "-2")) }
val newGraph2 = Graph(newVertices, graph.edges)
//图Map操作，被系统优化
val newGraph3 = graph.mapVertices((id, attr) =>
(id, (attr._1 + "-1", attr._2 + "-2")))
//构造一个新图，顶点属性是出度 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)
}自定义类型
Join 操作
map 操作
结构操作
//由已定义的顶点构成的子图 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操作也是根据输入图构造一个新图，达到一个限制制约的效果 val ccGraph = graph.connectedComponents() val validGraph = graph.subgraph(vpred = (id, attr) => attr._2 != "Missing") val validCCGraph = ccGraph.mask(validGraph)
Mask
图反向
子图
聚合操作
//计算年龄大于自己的关注者的总人数和总年龄
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 graph2 = GraphGenerators.logNormalGraph(sc, numVertices,
numEParts, mu, sigma).mapEdges(e=> e.attr.toDouble)
//定义一个源值 点
val sourceId: VertexId = 42
//初始化图的所有点，除了与指定的源值点相同值的点为0.0以外，其他点为无穷大
val initialGraph = graph2.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"))主要图算法
val pageRankGraph = graph2.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)
val connectedComponentsGraph = graph2.connectedComponents() connectedComponentsGraph.vertices.sortBy(_._2, false).saveAsTextFile( "/user/hadoop/data/temp/graph/graph.cc") connectedComponentsGraph.vertices.top(5)(Ordering.by(_._2)).foreach(println)
//TriangleCount主要用途之一是用于社区发现 保持sourceId小于destId val graph3 = GraphLoader.edgeListFile(sc, path, true) val triangleCountGraph = graph3.triangleCount() triangleCountGraph.vertices.sortBy(_._2,false).saveAsTextFile( "/user/hadoop/data/temp/graph/graph.tc") triangleCountGraph.vertices.top(5)(Ordering.by(_._2)).foreach(println)
TriangleCount
Connected Components
PageRank
其他操作
var path = "/user/Hadoop/data/temp/graph/graph.txt" var minEdgePartitions = 1 var canonicalOrientation = false // if sourceId < destId this value is true val graph2 = 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))
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)
到此，相信大家对“GraphX的基础知识有哪些”有了更深的了解，不妨来实际操作一番吧！这里是创新互联网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流
微信二维码
移动版官网