Spark2.2+Scala2.11+GraphX中mapReduceTriplets报错及替换方法咨询

嘿，我之前在Spark 2.x版本的GraphX里也踩过这个坑！mapReduceTriplets在Spark 2.0之后就被标记为废弃，到2.2版本已经彻底被移除了，官方推荐用更灵活的aggregateMessages来替代它。我来帮你把代码转换成对应的实现：

你的原代码是要计算每个节点关联的所有边属性之和，用aggregateMessages可以完美实现，步骤如下：

1. 定义消息发送函数 #

这个函数和你原来的nodeWeightMapFunc作用一致：遍历每个边三元组，给源节点和目标节点分别发送边的属性值：

val sendMsg: EdgeTriplet[VD, Long] => Iterator[(VertexId, Long)] = triplet => {
  Iterator((triplet.srcId, triplet.attr), (triplet.dstId, triplet.attr))
}

2. 定义消息合并函数 #

和你原来的nodeWeightReduceFunc逻辑完全相同，把同一个节点收到的所有消息值相加：

val mergeMsg: (Long, Long) => Long = (a, b) => a + b

3. 调用aggregateMessages计算结果 #

最后调用Graph的aggregateMessages方法，传入上面两个函数，同时指定要访问的三元组字段（这里我们需要源节点ID、目标节点ID和边属性，所以用TripletFields.All）：

val nodeWeights: VertexRDD[Long] = graph.aggregateMessages(sendMsg, mergeMsg, TripletFields.All)

补充小提示 #

• aggregateMessages的第三个参数TripletFields是用来做性能优化的：如果你的逻辑不需要访问节点属性，只需要边的信息，可以用TripletFields.EdgeOnly来减少数据传输；如果只需要源节点属性，用TripletFields.Src，以此类推。
• 返回的VertexRDD[Long]和mapReduceTriplets返回的类型完全一致，所以你后续对nodeWeights的操作可以保持不变。

内容的提问来源于stack exchange，提问作者Markus