Spark - Graphx: mapReduceTriplets против aggregateMessages

Question

Я бегу на уроке 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 
) 

Так я бегу точно такой же код, но затем У меня нет выхода. Является ли это ожидаемым результатом или я должен что-то изменить из-за того, что он связан с совокупностью сообщений?

Answer 1

Возможно, вам нужно что-то вроде этого:

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} 
) 

Вы можете найти aggregateMessages сигнатуру функции и полезные примеры на Grapx proggraming guide странице. Надеюсь это поможет.