Spark GraphX 聚合求和
Spark GraphX Aggregation Summation
我正在尝试计算 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)
这个returns以下:
(20,(1,300,1))
(11,(1,1550,1))
它正在为顶点 11 求和,但它没有汇总到根节点(顶点 20)。我错过了什么或者有更好的方法吗?当然,树可以是任意大小,每个顶点可以有任意数量的子边。
鉴于图表是有向的(在你的例子中它似乎是),应该可以编写一个 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
以便仅从下到上发送消息(否则我们将陷入无限循环)。
我正在尝试计算 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)
这个returns以下:
(20,(1,300,1))
(11,(1,1550,1))
它正在为顶点 11 求和,但它没有汇总到根节点(顶点 20)。我错过了什么或者有更好的方法吗?当然,树可以是任意大小,每个顶点可以有任意数量的子边。
鉴于图表是有向的(在你的例子中它似乎是),应该可以编写一个 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
以便仅从下到上发送消息(否则我们将陷入无限循环)。