官网指导https://spark.apache.org/docs/latest/graphx-programming-guide.html#pregel-api

一般关系图中，事物为顶点，关系为边

定义一个图：

1、对于每条边，矩阵中相应单元格值为1 2、对于每个循环，矩阵中相应单元格值为2，方便在行或列上求得顶点度数

GraphX是Spark提供分布式图计算API

GraphX特点

针对某些领域，如社交网络、语言建模等，graph-parallel系统可以高效地执行复杂的图形算法，比一般的data-parallel系统更快

Graphx是将graph-parallel的data-parallel统一到一个系统中。允许用户将数据当成一个图或一个集合RDD，而简化数据移动或复杂操作。

对Graph视图的所有操作，最终都会转换成其关联的Table视图的RDD操作来完成

创建一个简单的GraphX:

输出结果：

这里每一个操作产生一个新图，其顶点和边被用户定义的map函数修改了。 注意： 在每一个实例图结构不受影响。这是这些操作的关键特征，这允许结果图重复利用原始图的结构索引。下面的代码片段逻辑上是等同的，但是第一个没有保存结构索引，其不会从GraphX系统优化中获益。

演化一下这四个结构算子。

首先先创建一个新图，三个参数：

然后打印一下这个图：

输出图中各点的关系：打印结果如下，这个graph表示的是a到f这几个人之间的关系，a是b的客户，c是b的客户，c是d的病人，c是d的病人，c是d的朋友，e是f的爸爸

reverse算子的作用就是把edge的方向反过来，在这里就是把每个人的关系反过来一下 代码如下：

输出结果：

如图比较所有的关系双方都反过来了

subgraph顾名思义就是取原来graph的子graph，获取子graph肯定是有条件过滤掉一部分数据，剩下来的就是子graph 代码如下：

输出结果：

这个子graph,只保留了父graph中第一个属性比b的ascii码大的vertex

mask算子就是求当前graph和另外一个graph的交集 代码如下，我们使用了上一个算子的结果作为当前graph进行mask的参数：

输出结果：

groupEdges的作用是将2个vertex之间的所有edge进行合并，我们知道graphx处理的是多重图，多重图的特征就是2个顶点之间可能有多个平行边，这里的groupEdges就可以把这些平行边合并 代码如下：

我们这里将平行边的元素用and连接起来了，这里要注意的是，使用groupEdges算子之前，必须先用一下partitionBy，不然不起作用的 打印结果如下：

定义一个graph来测试join：

这个graph描述了每个人的名字和工作，这里我们给每个人增加除了名字和工作的其他属性，这个属性就是年龄属性 因此，我们需要定义一个rdd，描述每个人的年龄。 代码如下：

这里我们定义了id为3到7的这5个人的年龄，注意我们原来的graph的所有人的id为1到5，接下来有2种方法来把这个年龄属性加到我们graph中的每个人上面：

第一种方法就是outerJoinVertices,代码如下：

第二种方法就是joinVertices,代码如下：

输出结果：

例子2：

//Pregel val minGraph = graph.pregel(initialMsg, Int.MaxValue, EdgeDirection.Out)(vprog, sendMsg, mergeMsg) minGraph.vertices.collect.foreach{ case (vertexId, (value, original_value)) => println(value) }

val initialMsg = 9999 def vprog(vertexId: VertexId, value: (Int, Int), message: Int): (Int, Int) = { if (message == initialMsg) value else (message min value._1, value._1) }

def sendMsg(triplet: EdgeTriplet[(Int, Int), Boolean]): Iterator[(VertexId, Int)] = { val sourceVertex = triplet.srcAttr if (sourceVertex._1 == sourceVertex._2) Iterator.empty else Iterator((triplet.dstId, sourceVertex._1)) }

def mergeMsg(msg1: Int, msg2: Int): Int = msg1 min msg2