Community detection has become a popular field of research over the past few decades. Sadly, it did not move as fast as the digital world that a true data scientist lives in, with more and more data collected every second. As a result, most of the proposed solutions are simply not suitable for a big data environment.
Although a lot of algorithms suggest a new scalable way for detecting communities, none of them is actually meaning scalable in a sense of distributed algorithms and parallel computing.
Louvain algorithm is probably the most popular and widely used algorithm for detecting communities on undirected weighted graphs.
The idea is to start with each vertex being the center of its own community. At each step, we look for community neighbors and check whether or not merging both communities together would result in any gain in the modularity values. Going through each vertex, we compress the graph so that all nodes being part of the same community become a unique community vertex, with all community internal edges becoming a self-edge with aggregated weights. We repeat this process until the modularity can no longer be optimized. The process is reported as follows in Figure 3:
Because modularity will be updated any time a vertex changes, and because the change of each vertex will be driven by the global modularity update, vertices need to be processed in a serial order; making the modularity optimization a cut-off point to the nature of parallel computing. Recent studies have reported that the quality of results may decrease as the size of graph increases excessively so that modularity is unable to detect small and well-defined communities.
To the very best of our knowledge, the only distributed version of Louvain that is publicly available has been created by Sotera, a national security technology supplier (https://github.com/Sotera/distributed-graph-analytics/tree/master/dga-graphx). With different implementations on either MapReduce, Giraph, or GraphX, their idea is to make vertices choices simultaneously and update the graph state after each change. Because of the parallel nature, some of the vertex choices will be incorrect as they may not maximize a global modularity, but eventually become more and more consistent after repeated iterations.
This (potentially) slightly less accurate, but definitely highly scalable, algorithm was worth investigating, but because there is no right or wrong solution to the community detection problem and because each data science use case is different, we decided to build our own distributed version of a different algorithm rather than describing an existing one. For convenience, we repackaged this distributed version of Louvain and made it available in our GitHub repository.
By searching for some documentation material on graph algorithms, we came across a fantastic and recent white paper mentioning both scalability and parallel computing. We invite our readers to read this paper first before moving forward in the implementation.
Note
For more information about WCC algorithm, refer to the publication: A. Prat-Perez, D. Dominguez-Sal, and J.-L. Larriba-Pey, "High quality, scalable and parallel community detection for large real graphs," in Proceedings of the 23rd International Conference on World Wide Web, ser. WWW '14. New York, NY, USA: ACM, 2014, pp. 225-236
Although no implementation could be found, and the authors are discrete about the technologies they were using, we were particularly interested by the heuristic used as a measure of the graph partitioning since detection can be done in parallel without having to re-compute a global metric such as the graph modularity.
Also interesting is the assumption they use, inspired from real-life social networks, as a quality measure for detecting communities. Because communities are groups of vertices that are tightly connected together and loosely connected with the rest of the graph, there should be a high concentration of triangles closed within each community. In other words, vertices that form part of a community should be closing many more triangles in their own community than they would be closing outside:
As per the preceding equation, the clustering coefficient (WCC) for a given vertex x in a community C will be maximized when x will be closing more triangles inside of its community than outside (communities will be well defined) and/or when the number of its neighbors where it does not close any triangle with will be minimal (all nodes are interconnected). As reported in the following equation, the WCC of a community S will be the average WCC of each of its vertices:
Similarly, the WCC of a graph partition P will be the weighted average of each community's WCC:
The algorithm consists of three different phases explained next. A preprocessing step that creates an initial set of communities, a community-back propagation to ensure initial communities are consistent, and finally an iterative algorithm that optimizes the global clustering coefficient value.
The first step is to define a graph structure with vertices containing all the variables we need to compute the WCC metrics locally, including the current community a vertex belongs to, the number of triangles each vertex is closing inside and outside of its communities, the number of nodes it shares triangles with and the current WCC metric. All these variables will be wrapped into a VState class:
class VState extends Serializable {
var vId = -1L
var cId = -1L
var changed = false
var txV = 0
var txC = 0
var vtxV = 0
var vtxV_C = 0
var wcc = 0.0d
}In order to compute the initial WCC, we first need to count the number of triangles any vertex is closing within its neighborhood. Counting the number of triangles usually consists of aggregating the neighbors IDs for each vertex, sending this list to each of its neighbors, and searching for common IDs in both vertex neighbors and vertex neighbors' neighbors. Given two connected vertices A and B, the intersection between A's and B's respective list of neighbors is the number of triangles vertex A closes with B, and the aggregation in A returns the total number of triangles vertex A is closing across the graph.
In large networks with highly connected vertices, sending a list of adjacent vertices to each neighbor can be time consuming and network intensive. In GraphX, the triangleCount function has been optimized so that for each edge, only the least important vertex (in term of degrees) will be sending its list to its adjacent nodes, hence minimizing the associated cost. This optimization requires the graph to be canonical (a source ID is lower than a destination ID) and partitioned. Using our graph of persons, this can be done as follows:
val cEdges: RDD[Edge[ED]] = graph.edges
.map { e =>
if(e.srcId > e.dstId) {
Edge(e.dstId, e.srcId, e.attr)
} else e
}
val canonicalGraph = Graph
.apply(graph.vertices, cEdges)
.partitionBy(PartitionStrategy.EdgePartition2D)
canonicalGraph.cache()
canonicalGraph.vertices.count()A prerequisite of the WCC optimization is to remove edges that are not part of any triangle since they will not be contributing to communities. We therefore need to count the number of triangles, the degree of each vertex, the neighbor's IDs, and we finally remove edges where the intersection of neighbor's IDs is empty. Filtering out these edges can be done using the subGraph method that takes both a filter function for edges' triplets and a filter function for vertices as input arguments:
val triGraph = graph.triangleCount()
val neighborRdd = graph.collectNeighborIds(EdgeDirection.Either)
val subGraph = triGraph.outerJoinVertices(neighborRdd)({ (vId, triangle, neighbors) =>
(triangle, neighbors.getOrElse(Array()))
}).subgraph((t: EdgeTriplet[(Int, Array[Long]), ED]) => {
t.srcAttr._2.intersect(t.dstAttr._2).nonEmpty
}, (vId: VertexId, vStats: (Int, Array[Long])) => {
vStats._1 > 0
})Because we removed all edges that were not closing any triangle, the number of degrees for each vertex becomes the number of distinct vertices a given vertex is closing triangles with. Finally, we create our initial VState graph as follows, where each vertex becomes a center node of its own community:
val initGraph: Graph[VState, ED] = subGraph.outerJoinVertices(subGraph.degrees)((vId, vStat, degrees) => {
val state = new VState()
state.vId = vId
state.cId = vId
state.changed = true
state.txV = vStat._1
state.vtxV = degrees.getOrElse(0)
state.wcc = degrees.getOrElse(0).toDouble / vStat._1
state
})
initGraph.cache()
initGraph.vertices.count()
canonicalGraph.unpersist(blocking = false)The second step of this phase is to initialize communities using these initial WCC values. We define our initial set of communities as being consistent if and only if the following three requirements are all met:
- Any community must contain a single center node and border nodes, and all border vertices must be connected to the community center
- Any community center must have the highest clustering coefficient in its community
- A border vertex that is connected to two different centers (therefore two different communities according to rule 1) must be part of the community whose center has the highest clustering coefficient
In order to define our initial communities, each vertex needs to send information to its neighbors, including its ID, its clustering coefficient, its degrees, and the current community it belongs to. For convenience, we will send the main vertex attribute VState class as a message as it already contains all this information. Vertices will receive these messages from their neighborhood, will select the best one with the highest WCC score (within our getBestCid method), highest degree, highest ID, and will update their community accordingly.
This communication across vertices is a perfect use case for the aggregateMessages function, the equivalent of the map-reduce paradigm in GraphX. This function requires two functions to be implemented, one that sends a message from one vertex to its adjacent node, and one that aggregates multiple messages at the vertex level. This process is called message passing and is described as follows:
def getBestCid(v: VState, msgs: Array[VState]): VertexId = {
val candidates = msgs filter {
msg =>
msg.wcc > v.wcc ||
(msg.wcc == v.wcc && msg.vtxV > v.vtxV) ||
(msg.wcc == v.wcc && msg.vtxV > v.vtxV && msg.cId > v.cId)
}
if(candidates.isEmpty) {
v.cId
} else {
candidates
.sortBy {
msg =>
(msg.wcc, msg.vtxV, msg.cId)
}
.last
.cId
}
}
def sendMsg = (ctx: EdgeContext[VState, ED, Array[VState]]) => {
ctx.sendToDst(
Array(ctx.srcAttr)
)
ctx.sendToSrc(
Array(ctx.dstAttr)
)
}
def mergeMsg = (m1: Array[VState], m2: Array[VState]) => {
m1 ++ m2
}
def msgs = subGraph.aggregateMessages(sendMsg, mergeMsg)
val initCIdGraph = subGraph.outerJoinVertices(msgs)((vId, vData, msgs) => {
val newCId = getBestCid(vData, msgs.getOrElse(Array()))
vData.cId = newCId
vData
})
initCIdGraph.cache()
initCIdGraph.vertices.count()
initGraph.unpersist(blocking = false)An example of this community initialization process is reported in Figure 4. The left graph, with its nodes proportionally resized to reflect their true WCC coefficients, has been initialized with four different communities, 1, 11, 16, and 21.
Although one will surely appreciate that a single aggregateMessages function was returning relatively consistent communities, this initial partitioning violates the third of the rules we defined earlier. Some vertices (such as 2, 3, 4, and 5) belongs to a community whose center is not a center node (vertex 1 belongs to community 21). This same issue is noticed for community 11.
In order to address this inconsistency and respect our third requirement, any vertex x must broadcast its updated community to all of its neighbors having a lower coefficient, as, according to our second rule, only these lower ranked vertices could potentially become a border node of x. Any further update will result in a new message to be passed to lower rank vertices, and so on, until no vertices will change community, at which point the third of our rules will be satisfied.
Since no global knowledge of the graph is required between iterations (such as counting the global WCC value), community updates can be extensively parallelized using the Pregel API of GraphX. Initially developed at Google, Pregel allows vertices to receive messages from previous iterations, send new messages to their neighborhood and modify their own state until no further messages could be sent.
Note
For more information about Pregel algorithm, refer to the publication: G. Malewicz, M. H. Austern, A. J. Bik, J. C. Dehnert, I. Horn, N. Leiser, and G. Czajkowski, "Pregel: A system for large-scale graph processing," in Proceedings of the 2010 ACM SIGMOD International Conference on Management of Data, ser. SIGMOD '10. New York, NY, USA: ACM, 2010, pp. 135-146. [Online]. Available: http://doi.acm.org/10.1145/1807167.1807184
Similar to the aggregateMessages function mentioned earlier, we will send the vertex attribute VState as a message across vertices, with, as an initial message for Pregel super step, a new object initialized with default values (WCC of 0).
val initialMsg = new VState()
When more than one message is received at the vertex level, we only keep the one with the highest clustering coefficient, and given the same coefficient, the one with the highest degree (and then the highest ID). We create an implicit ordering on VState for that purpose:
implicit val VSOrdering: Ordering[VState] = Ordering.by({ state =>
(state.wcc, state.vtxV, state.vId)
})
def compareState(c1: VState, c2: VState) = {
List(c1, c2).sorted(VStateOrdering.reverse)
}
val mergeMsg = (c1: VState, c2: VState) => {
compareState(c1, c2).head
}Following the same principle as per recursive algorithms, we need to properly define a breaking clause at which point Pregel should stop sending and processing messages. This will be done within the send function that takes an edge triplet as an input and returns an iterator of messages. A vertex will send its VState attribute if and only if its community has changed over the previous iteration. In that case, the vertex will inform its lower ranked neighbors about its community update but will also send a signal to itself to acknowledge this successful broadcast. The latter is our breaking clause as it ensures no further message will be sent from that given node (unless its community gets updated in the further steps):
def sendMsg = (t: EdgeTriplet[VState, ED]) => {
val messages = mutable.Map[Long, VState]()
val sorted = compareState(t.srcAttr, t.dstAttr)
val (fromNode, toNode) = (sorted.head, sorted.last)
if (fromNode.changed) {
messages.put(fromNode.vId, fromNode)
messages.put(toNode.vId, fromNode)
}
messages.toIterator
}The last function to implement is the core function of the Pregel algorithm. Here we define the logic to be applied at the vertex level given the unique message we selected from the mergeMsg function. We identify four different possibilities of messages, each of them defined with the logic to be applied on the vertex status.
- If the message is the initial message sent from Pregel (vertex ID is not set, WCC is null), we do not update the vertex community ID.
- If the message comes from the vertex itself, this is an acknowledgement from the
sendMsgfunction, we set the vertex status to silent. - If the message (with higher WCC) comes from a center node of a community, we update the vertex attribute to be a border node of this new community.
- If the message (with higher WCC) comes from a border node of a community, this vertex becomes a center of its own community and will broadcast this update further to its lower-ranked network.
def vprog = (vId: VertexId, state: VState, message: VState) => {
if (message.vId >= 0L) {
// message comes from myself
// I stop spamming people
if (message.vId == vId) {
state.changed = false
}
// Sender is a center of its own community
// I become a border node of its community
if (message.cId == message.vId) {
state.changed = false
state.cId = message.cId
}
// Sender is a border node of a foreign community
// I become a center of my own community
// I broadcast this change downstream
if (message.cId != message.vId) {
state.changed = true
state.cId = vId
}
}
state
}Finally, we chain all these three functions together using the apply function of the Pregel object. We set the maximum number of iterations to infinity as we rely on the breaking clause we defined using an acknowledgment type message:
val pregelGraph: Graph[VState, ED] = Pregel.apply( initCIdGraph, initialMsg, Int.MaxValue )( vprog, sendMsg, mergeMsg ) pregelGraph.cache() pregelGraph.vertices.count()
Although the concept of Pregel is fascinating, its implementation is certainly not. As a reward to this tremendous effort, we display the resulting graph in Figure 5 next. Vertices 1 and 11 are still part of community 21 which remains valid, but communities 1 and 11 have now been replaced with communities 15 and 5 respectively, vertices having the highest clustering coefficient, degree, or ID in their community, hence validating the third requirement:
We used the Pregel API to create our initial set of communities with respect to the rules introduced earlier, but we are not set yet. The preceding figure definitely suggests some improvements that will be addressed in the following subsection. However, before moving forward, one can notice that no particular partitioning was used here. If we were to send multiple messages across communities' nodes, and if those vertices were located on different partitions (hence different executors), we would certainly not optimize the network traffic related to message passing. Different sorts of partitioning exist in GraphX, but none of them allow us to use a vertex attribute such as the community ID as a measure of the partitioning.
In the following simple function, we extract all graph triplets, build a hashcode out of a community tuple and repartition this edge RDD using the standard key value HashPartitioner class. We finally build a new graph out of this repartitioned set so that we guarantee all vertices connected from a community C1 to a community C2 will all belong to the same partition:
def repartition[ED: ClassTag](graph: Graph[VState, ED]) = {
val partitionedEdges = graph
.triplets
.map {
e =>
val cId1 = e.srcAttr.cId
val cId2 = e.dstAttr.cId
val hash = math.abs((cId1, cId2).hashCode())
val partition = hash % partitions
(partition, e)
}
.partitionBy(new HashPartitioner(partitions))
.map {
pair =>
Edge(pair._2.srcId, pair._2.dstId, pair._2.attr)
}
Graph(graph.vertices, partitionedEdges)
}The purpose of this stage is to iteratively let all vertices choose between the following three options until the WCC value cannot be longer optimized, at which point our community detection algorithm would converge to its optimal graph structure:
- STAY: To stay in its community
- TRANSFER: To move from its community and become part of its neighbor's community
- REMOVE: To leave its community and become part of its own community
For each vertex, the best movement is the one that maximizes the total WCC value. Similar to the Louvain approach, each movement depends on a global score to be computed, but the reason we turned to this algorithm is that this score can be approximated using a heuristic defined in High quality, scalable and parallel community detection for large real graphs from Arnau Prat-Pérez et. al. Because this heuristic does not require the computation of all internal triangles, vertices can all move simultaneously and this process can therefore be designed in a fully decentralized and highly scalable way.
In order to compute this heuristic, we first need to aggregate basic statistics at the community level such as the number of elements and the number of inbound and outbound links, both of them expressed as a simple word count function here. We combine them in memory as the number of communities will be considerably smaller than the number of vertices:
case class CommunityStats(
r: Int,
d: Double,
b: Int
)
def getCommunityStats[ED: ClassTag](graph: Graph[VState, ED]) = {
val cVert = graph
.vertices
.map(_._2.cId -> 1)
.reduceByKey(_+_)
.collectAsMap()
val cEdges = graph
.triplets
.flatMap { t =>
if(t.srcAttr.cId == t.dstAttr.cId){
Iterator((("I", t.srcAttr.cId), 1))
} else {
Iterator(
(("O", t.srcAttr.cId), 1),
(("O", t.dstAttr.cId), 1)
)
}
}
.reduceByKey(_+_)
.collectAsMap()
cVert.map {
case (cId, cCount) =>
val intEdges = cEdges.getOrElse(("I", cId), 0)
val extEdges = cEdges.getOrElse(("O", cId), 0)
val density = 2 * intEdges / math.pow(cCount, 2)
(cId, CommunityStats(cCount, density, extEdges))
}
}Finally, we collect both the number of vertices and the community statistics (including the community edge density) and broadcast the results to all of our Spark executors:
var communityStats = getCommunityStats(pregelGraph) val bCommunityStats = sc.broadcast(communityStats)
Tip
It is important to understand the use of the broadcast method here. If the community statistics are used within a Spark transformation, this object will be sent to the executors for each record the latter has to process. We compute them once, broadcast the result to the executors' caches so that any closure can locally make use them, hence saving lots of unnecessary network transfer.
According to the set of equations defined earlier, each vertex must have access to the community statistics it belongs to and the number of triangles it closes with any vertex inside of its community. For that purpose, we collect neighbors via a simple message passing, but only on the vertices within the same community, thus limiting the network traffic:
def collectCommunityEdges[ED: ClassTag](graph: Graph[VState, ED]) = {
graph.outerJoinVertices(graph.aggregateMessages((e: EdgeContext[VState, ED, Array[VertexId]]) => {
if(e.dstAttr.cId == e.srcAttr.cId){
e.sendToDst(Array(e.srcId))
e.sendToSrc(Array(e.dstId))
}
}, (e1: Array[VertexId], e2: Array[VertexId]) => {
e1 ++ e2
}))((vid, vState, vNeighbours) => {
(vState, vNeighbours.getOrElse(Array()))
})
}Similarly, we count the number of shared triangles using the following function. Note that we use the same optimization as per the default triangleCount method using the smallest set only to send messages to the largest one.
def collectCommunityTriangles[ED: ClassTag](graph: Graph[(VState, Array[Long]), ED]) = {
graph.aggregateMessages((ctx: EdgeContext[(VState, Array[Long]), ED, Int]) => {
if(ctx.srcAttr._1.cId == ctx.dstAttr._1.cId){
val (smallSet, largeSet) = if (ctx.srcAttr._2.length < ctx.dstAttr._2.length) {
(ctx.srcAttr._2.toSet, ctx.dstAttr._2.toSet)
} else {
(ctx.dstAttr._2.toSet, ctx.srcAttr._2.toSet)
}
val it = smallSet.iterator
var counter: Int = 0
while (it.hasNext) {
val vid = it.next()
if (
vid != ctx.srcId &&
vid != ctx.dstId &&
largeSet.contains(vid)
) {
counter += 1
}
}
ctx.sendToSrc(counter)
ctx.sendToDst(counter)
}
}, (e1: Int, e2: Int) => (e1 + e2))
}We compute and update the new WCC score of each vertex as a function of the community neighborhood size and the number of community triangles. This equation is the one described earlier while introducing the WCC algorithm. We compute a score as a ratio of triangles closed inside versus outside of a community C given a vertex x:
def updateGraph[ED: ClassTag](graph: Graph[VState, ED], stats: Broadcast[Map[VertexId, CommunityStats]]) = {
val cNeighbours = collectCommunityEdges(graph)
val cTriangles = collectCommunityTriangles(cNeighbours)
cNeighbours.outerJoinVertices(cTriangles)(
(vId, vData, tri) => {
val s = vData._1
val r = stats.value.get(s.cId).get.r
// Core equation: compute WCC(v,C)
val a = s.txC * s.vtxV
val b = (s.txV * (r - 1 + s.vtxV_C).toDouble)
val wcc = a / b
val vtxC = vData._2.length
s.vtxV_C = s.vtxV – vtxC
// Triangles are counted twice (incoming / outgoing)
s.txC = tri.getOrElse(0) / 2
s.wcc = wcc
s
})
}
val wccGraph = updateGraph(pregelGraph, bCommunityStats)The global WCC value is a simple aggregation of each vertex WCC normalized with the number of elements in each community. This value must be broadcast to Spark executors too as it will be used inside of a Spark transformation:
def computeWCC[ED: ClassTag](graph: Graph[VState, ED], cStats: Broadcast[Map[VertexId, CommunityStats]]): Double = {
val total = graph.vertices
.map {
case (vId, vState) =>
(vState.cId, vState.wcc)
}
.reduceByKey(_+_)
.map {
case (cId, wcc) =>
cStats.value.get(cId).get.r * wcc
}
.sum
total / graph.vertices.count
}
val wcc = computeWCC(wccGraph, bCommunityStats)
val bWcc = sc.broadCast(wcc)Given the cost of inserting a vertex x into a community C, the costs of removing/transferring x from/to a community C can be expressed as a function of the former, and can be derived from three parameters Θ1, Θ2, and Θ3. This heuristic states that for each vertex x, a single computation is needed for each of its surrounding communities C, and can be done in parallel assuming we gathered all the community statistics in the first place:
The computation of Θ1, Θ2, and Θ3 will not be reported here (it is available on our GitHub), but depends on the community density, the external edges, and the number of elements, all of them available within our broadcasted set of CommunityStats objects defined earlier. Finally, it is worth mentioning that this computation has a linear time complexity.
At each iteration, we will collect the different communities surrounding any vertex, and will aggregate the number of edges using the mappend aggregation from Scalaz that we introduced in Chapter 6, Scraping Link-Based External Data. This helps us to limit the amount of code written and avoids using mutable objects.
val cDegrees = itGraph.aggregateMessages((ctx: EdgeContext[VState, ED, Map[VertexId, Int]]) => {
ctx.sendToDst(
Map(ctx.srcAttr.cId -> 1)
)
ctx.sendToSrc(
Map(ctx.dstAttr.cId -> 1)
)
}, (e1: Map[VertexId, Int], e2: Map[VertexId, Int]) => {
e1 |+| e2
})Using the community statistics, the WCC value from the previous iteration, the number of vertices and the above edges count, we can now estimate the cost of inserting each vertex x into a surrounding community C. We find the local best movement for each vertex and for each of its surrounding communities, and finally apply the best one that maximizes the WCC value.
Finally, we call back the set of methods and functions defined earlier in order to update the new WCC value for each vertex, for each community, and then for the graph partition itself to see whether or not all these changes resulted in any WCC improvement. If the WCC value cannot be optimized any longer, the algorithm has converged to its optimal structure and we finally return a vertex RDD containing both the vertex ID and the final community ID this vertex belongs to.
Our test community graph has been optimized (not without its fair share of effort) and reported as shown in Figure 6:
We observe all the changes we were expecting from the previous figures. Vertices 1 and 11 are now part of their expected communities, respectively 5 and 11. We also note that vertex 16 has now been included within its community 11.