Spark GraphX 聚合求和答案

【问题标题】：Spark GraphX Aggregation SummationSpark GraphX 聚合求和
【发布时间】：2017-05-18 01:02:18
【问题描述】：

我正在尝试计算 spark graphx 图中节点值的总和。简而言之，该图是一棵树，顶部节点（根）应该对所有子节点及其子节点求和。我的图表实际上是一棵看起来像这样的树，预期的总和值应该是 1850：

                                     +----+
                     +--------------->    |  VertexID 14
                     |               |    |  Value: 1000
                 +---+--+            +----+
    +------------>      | VertexId 11
    |            |      | Value:     +----+
    |            +------+ Sum of 14 & 24  |  VertexId 24
+---++                +-------------->    |  Value: 550
|    | VertexId 20                   +----+
|    | Value:
+----++Sum of 11 & 911
      |
      |           +-----+
      +----------->     | VertexId 911
                  |     | Value: 300
                  +-----+

第一次尝试是这样的：

val vertices: RDD[(VertexId, Int)] =
      sc.parallelize(Array((20L, 0)
        , (11L, 0)
        , (14L, 1000)
        , (24L, 550)
        , (911L, 300)
      ))

  //note that the last value in the edge is for factor (positive or negative)
    val edges: RDD[Edge[Int]] =
      sc.parallelize(Array(
        Edge(14L, 11L, 1),
        Edge(24L, 11L, 1),
        Edge(11L, 20L, 1),
        Edge(911L, 20L, 1)
      ))

    val dataItemGraph = Graph(vertices, edges)


    val sum: VertexRDD[(Int, BigDecimal, Int)] = dataItemGraph.aggregateMessages[(Int, BigDecimal, Int)](
      sendMsg = { triplet => triplet.sendToDst(1, triplet.srcAttr, 1) },
      mergeMsg = { (a, b) => (a._1, a._2 * a._3 + b._2 * b._3, 1) }
    )

    sum.collect.foreach(println)

这将返回以下内容：

(20,(1,300,1))
(11,(1,1550,1))

它正在对顶点 11 进行求和，但它没有向上滚动到根节点（顶点 20）。我错过了什么或者有更好的方法吗？当然，树可以是任意大小，每个顶点可以有任意数量的子边。

【问题讨论】：

我现在想知道 Pregel 是否是正确的使用方法。

标签： scala apache-spark spark-graphx

【解决方案1】：

鉴于图形是有向的（就像在您的示例中那样），应该可以编写一个 Pregel 程序来满足您的要求：

val result = 
 dataItemGraph.pregel(0, activeDirection = EdgeDirection.Out)(
  (_, vd, msg) => msg + vd, 
  t => Iterator((t.dstId, t.srcAttr)), 
  (x, y) => x + y
 )

 result.vertices.collect().foreach(println)

// Output is:
// (24,550)
// (20,1850)
// (14,1000)
// (11,1550)
// (911,300)

我正在使用EdgeDirection.Out，因此消息只能从下往上发送（否则我们会陷入无限循环）。

【讨论】：