版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/lm709409753/article/details/85019833
1.LPA 标签传播算法,主要是顶点计算函数,选择label标签最多的项,更新顶点的属性。根据相应的业务,可以修改graphx的源码进行修改,改为我们业务中需要的标签值。由于LPA很难保证收敛,所以要设定迭代次数。
2.代码
object LabelPropagationAlgorithm {
/**
* Run static Label Propagation for detecting communities in networks.
*
* Each node in the network is initially assigned to its own community. At every superstep, nodes
* send their community affiliation to all neighbors and update their state to the mode community
* affiliation of incoming messages.
*
* LPA is a standard community detection algorithm for graphs. It is very inexpensive
* computationally, although (1) convergence is not guaranteed and (2) one can end up with
* trivial solutions (all nodes are identified into a single community).
*
* @tparam ED the edge attribute type (not used in the computation)
* @param graph the graph for which to compute the community affiliation
* @param maxSteps the number of supersteps of LPA to be performed. Because this is a static
* implementation, the algorithm will run for exactly this many supersteps.
* @return a graph with vertex attributes containing the label of community affiliation
*/
def run[VD, ED: ClassTag](graph: Graph[VD, ED], maxSteps: Int): Graph[VertexId, ED] = {
require(maxSteps > 0, s"Maximum of steps must be greater than 0, but got ${maxSteps}")
val lpaGraph = graph.mapVertices { case (vid, _) => vid }
def sendMessage(e: EdgeTriplet[VertexId, ED]): Iterator[(VertexId, Map[VertexId, Long])] = {
Iterator((e.srcId, Map(e.dstAttr -> 1L)), (e.dstId, Map(e.srcAttr -> 1L)))
}
def mergeMessage(count1: Map[VertexId, Long], count2: Map[VertexId, Long])
: Map[VertexId, Long] = {
(count1.keySet ++ count2.keySet).map { i =>
val count1Val = count1.getOrElse(i, 0L)
val count2Val = count2.getOrElse(i, 0L)
i -> (count1Val + count2Val)
}.toMap
}
def vertexProgram(vid: VertexId, attr: Long, message: Map[VertexId, Long]): VertexId = {
if (message.isEmpty) attr else message.maxBy(_._2)._1
}
val initialMessage = Map[VertexId, Long]()
Pregel(lpaGraph, initialMessage, maxIterations = maxSteps)(
vprog = vertexProgram,
sendMsg = sendMessage,
mergeMsg = mergeMessage)
}
}