1

我正在通过教程 http://ampcamp.berkeley.edu/big-data-mini-course/graph-analytics-with-graphx.html

在某些时候，我们使用 mapReduceTriplets 操作。这将返回预期结果

// Find the oldest follower for each user
val oldestFollower: VertexRDD[(String, Int)] = userGraph.mapReduceTriplets[(String, Int)](
  // For each edge send a message to the destination vertex with the attribute of the source vertex
  edge => Iterator((edge.dstId, (edge.srcAttr.name, edge.srcAttr.age))),
  // To combine messages take the message for the older follower
  (a, b) => if (a._2 > b._2) a else b
)

但是 IntelliJ 指出 mapReduceTriplets 已被弃用（从 1.2.0 开始），应该由 aggregateMessages 替换

// Find the oldest follower for each user
val oldestFollower: VertexRDD[(String, Int)] = userGraph.aggregateMessages()[(String, Int)](
  // For each edge send a message to the destination vertex with the attribute of the source vertex
  edge => Iterator((edge.dstId, (edge.srcAttr.name, edge.srcAttr.age))),
  // To combine messages take the message for the older follower
  (a, b) => if (a._2 > b._2) a else b
)

所以我运行完全相同的代码，但没有任何输出。这是预期的结果还是我应该因为聚合消息的变化而改变一些东西？

4

1 回答 1

5

可能你需要这样的东西：

val oldestFollower: VertexRDD[(String, Int)] = userGraph.aggregateMessages[(String, Int)]
(
    // For each edge send a message to the destination vertex with the attribute of the source vertex
    sendMsg = { triplet => triplet.sendToDst(triplet.srcAttr.name, triplet.srcAttr.age) },
   // To combine messages take the message for the older follower
    mergeMsg = {(a, b) => if (a._2 > b._2) a else b}
)

您可以在Grapx 编程指南页面找到aggregateMessages函数签名和有用的示例。希望这可以帮助。

于 2015-01-31T23:08:28.867 回答