• 顶点和边RDDs
    • VertexRDDs
    • EdgeRDDs

    顶点和边RDDs

    GraphX暴露保存在图中的顶点和边的RDD。然而，因为GraphX包含的顶点和边拥有优化的数据结构，这些数据结构提供了额外的功能。顶点和边分别返回VertexRDD和EdgeRDD。这一章
    我们将学习它们的一些有用的功能。

    VertexRDDs

    VertexRDD[A]继承自RDD[(VertexID, A)]并且添加了额外的限制，那就是每个VertexID只能出现一次。此外，VertexRDD[A]代表了一组属性类型为A的顶点。在内部，这通过
    保存顶点属性到一个可重复使用的hash-map数据结构来获得。所以，如果两个VertexRDDs从相同的基本VertexRDD获得（如通过filter或者mapValues），它们能够在固定的时间内连接
    而不需要hash评价。为了利用这个索引数据结构，VertexRDD暴露了一下附加的功能：

    1. class VertexRDD[VD] extends RDD[(VertexID, VD)] {
    2.   // Filter the vertex set but preserves the internal index
    3.   def filter(pred: Tuple2[VertexId, VD] => Boolean): VertexRDD[VD]
    4.   // Transform the values without changing the ids (preserves the internal index)
    5.   def mapValues[VD2](map: VD => VD2): VertexRDD[VD2]
    6.   def mapValues[VD2](map: (VertexId, VD) => VD2): VertexRDD[VD2]
    7.   // Remove vertices from this set that appear in the other set
    8.   def diff(other: VertexRDD[VD]): VertexRDD[VD]
    9.   // Join operators that take advantage of the internal indexing to accelerate joins (substantially)
    10.   def leftJoin[VD2, VD3](other: RDD[(VertexId, VD2)])(f: (VertexId, VD, Option[VD2]) => VD3): VertexRDD[VD3]
    11.   def innerJoin[U, VD2](other: RDD[(VertexId, U)])(f: (VertexId, VD, U) => VD2): VertexRDD[VD2]
    12.   // Use the index on this RDD to accelerate a `reduceByKey` operation on the input RDD.
    13.   def aggregateUsingIndex[VD2](other: RDD[(VertexId, VD2)], reduceFunc: (VD2, VD2) => VD2): VertexRDD[VD2]
    14. }

    举个例子，filter操作如何返回一个VertexRDD。过滤器实际使用一个BitSet实现，因此它能够重用索引以及保留和其它VertexRDDs做连接时速度快的能力。同样的，mapValues操作
    不允许map函数改变VertexID，因此可以保证相同的HashMap数据结构能够重用。当连接两个从相同的hashmap获取的VertexRDDs和使用线性扫描而不是昂贵的点查找实现连接操作时，leftJoin
    和innerJoin都能够使用。

    从一个RDD[(VertexID, A)]高效地构建一个新的VertexRDD，aggregateUsingIndex操作是有用的。概念上，如果我通过一组顶点构造了一个VertexRDD[B]，而VertexRDD[B]是
    一些RDD[(VertexID, A)]中顶点的超集，那么我们就可以在聚合以及随后索引RDD[(VertexID, A)]中重用索引。例如：

    1. val setA: VertexRDD[Int] = VertexRDD(sc.parallelize(0L until 100L).map(id => (id, 1)))
    2. val rddB: RDD[(VertexId, Double)] = sc.parallelize(0L until 100L).flatMap(id => List((id, 1.0), (id, 2.0)))
    3. // There should be 200 entries in rddB
    4. rddB.count
    5. val setB: VertexRDD[Double] = setA.aggregateUsingIndex(rddB, _ + _)
    6. // There should be 100 entries in setB
    7. setB.count
    8. // Joining A and B should now be fast!
    9. val setC: VertexRDD[Double] = setA.innerJoin(setB)((id, a, b) => a + b)

    EdgeRDDs

    EdgeRDD[ED]继承自RDD[Edge[ED]]，使用定义在PartitionStrategy的
    各种分区策略中的一个在块分区中组织边。在每个分区中，边属性和相邻结构被分别保存，当属性值改变时，它们可以最大化的重用。

    EdgeRDD暴露了三个额外的函数

    1. // Transform the edge attributes while preserving the structure
    2. def mapValues[ED2](f: Edge[ED] => ED2): EdgeRDD[ED2]
    3. // Revere the edges reusing both attributes and structure
    4. def reverse: EdgeRDD[ED]
    5. // Join two `EdgeRDD`s partitioned using the same partitioning strategy.
    6. def innerJoin[ED2, ED3](other: EdgeRDD[ED2])(f: (VertexId, VertexId, ED, ED2) => ED3): EdgeRDD[ED3]

    在大多数的应用中，我们发现，EdgeRDD操作可以通过图操作者(graph operators)或者定义在基本RDD中的操作来完成。