LouvainMethod分布式运行的升级之路

1、背景介绍        

        Louvain是大规模图谱的谱聚类算法，引入模块度的概念分二阶段进行聚类，直到收敛为止。分布式的代码可以在如下网址进行下载。

GitHub - Sotera/spark-distributed-louvain-modularity: Spark / graphX implementation of the distributed louvain modularity algorithm

  该代码依赖的spark-core和spark-graphx、scala-lang是2.10版本，采用的gradle的进行打包，也可以采用maven进行打包，解决相关的依赖问题之后，本地模式可以很快跑通。但是转向集群的时候，发现集群的spark的scala版本是2.12,我采用的是maven的scala编译的版本是2.10, 编译用到的scala和运行环境的scala版本不一致，结果无法进行spark集群模式的运行。

2、LouvainMethod的升级之路

        首先更改环境，即把louvain的代码依赖保持2.10,把spark的scala版本改成2.10,但是这样会影响其他任务的执行，其他任务可能依赖2.12的版本。因此，踏上了LouvainMethod的升级之路，即由2.10升级到2.12。

        将项目依赖的版本和编译的scala版本改为2.12之后，发现在新的高版本的spark-graphx_2.12版本里Graph对象没有了mapReduceTriplet方法，通过查找发现该方法在2.12版本的GraphXUtils类里，以一个私有方法存在，只能在包graphx下被访问，对外部不可见，因此首先想到的是通过反射机制对该私有方法进行访问，参照了如下的方法：

         在任意scala对象中调用私有方法 - 问答 - 腾讯云开发者社区-腾讯云

代码调试后，私有方法带有泛类型参数和普通参数，可以正常被反射出来，然而在调用的时候，始终报 wrong-number-of-arguments的问题。原因还没有查到。继而通过高版本的api是实现低版本的mapReduceTriplets方法。   参照该文档  GraphX - Spark 3.4.1 Documentation   的api接口含义，注意到新版2.12的Graph里aggregateMessage方法和低版本的mapReduceTriplets返回值一致，参数类型有diff，高版本的参数是EdgeContext，低版本的是EdgeTriplet，高版本通过sendToDst和sendToSrc对低版本进行了简化，使用功能更强大，因此尝试用aggregateMessage实现mapReduceTriplets。

val nodeWeightMapFunc = (e:EdgeTriplet[VD,Long]) => Iterator((e.srcId,e.attr), (e.dstId,e.attr))

val nodeWeightReduceFunc = (e1:Long,e2:Long) => e1+e2

转化为：

def nodeWeightMapFunc(e:EdgeContext[VD, Long, Long]) {
e.sendToDst(e.attr)
e.sendToSrc(e.attr)
}
Msg与reduceFunc的返回值保持一致。

通过如下方式进行调用：val nodeWeights = graph.aggregateMessages[Long](nodeWeightMapFunc,nodeWeightReduceFunc)

sendMsg的低版本如下：

private def sendMsg(et:EdgeTriplet[VertexState,Long]) = {
    val m1 = (et.dstId,Map((et.srcAttr.community,et.srcAttr.communitySigmaTot)->et.attr))
   val m2 = (et.srcId,Map((et.dstAttr.community,et.dstAttr.communitySigmaTot)->et.attr))
   Iterator(m1, m2)
}

升级为：

private def sendMsg(et: EdgeContext[VertexState, Long, Map[(Long,Long),Long]]) = {
et.sendToSrc(Map((et.dstAttr.community, et.dstAttr.communitySigmaTot) -> et.attr))
et.sendToDst(Map((et.srcAttr.community, et.srcAttr.communitySigmaTot) -> et.attr))
}

4、在集群运行相关jar的及运行脚本