Incremental, Data-Parallel Graph Clustering

Differential Dataflow [19] is an incremental computational model that has shown promise in a variety of graph processing tasks. In their paper, they demonstrated that their computational model was able to solve graph algorithms such as PageRank, Strongly Connected Components with a significant increase in performance over a selection of baselines, the speedups ranged from 1.9x to 1899.6x. Given most greedy modularity optimisation methods require multiple iterations over non-changing, already calculated vertices, it therefore makes sense to warrant an investigation on how Differential Dataflow could be used to reduce the cost of these iterations, additionally when we consider a dynamically changing graph we note that the increase in possible community swaps do not immediately increase with respect to the number of vertices but rather with respect to the degree of the inserted edge, although a swap as a result of this insertion may result in a domino effect of changes later on still.

One criteria for choosing Louvain modularity optimisation as a candidate for our exploration of incremental, data-parallel graph clustering was simply as a result of its usefulness and widespread use in both academia and in industry. Louvain despite its pitfalls has remained the most commonly used graph clustering algorithm with various open source implementations in distributed dataflow processing frameworks such as Apache Spark ¹¹1See https://spark.apache.org/. The other criteria for choosing Louvain is simply due to the unnecessary work it performs when iterating over vertices, this provides a place of improvement in terms of performance, additionally as mentioned previously the advantages of incrementally performing Louvain may result in showing that the use of Louvain like methods used for agglomerative clustering could also be used in conjunction with Differential Dataflow.

Although the Louvain modulairty algorithm always improves modularity in its serial implementation, this property is not maintained in parallel implementations. Unfortunately this means that it is possible for the parallel implementation to worsen the modularity of a graph and in this case steps must be made to prevent or mitigate this issue. This negative gain scenario present in Parallel Louvain is discussed and proven by Lu [18].

Unfortunately as a result of our computations being performed on stale data, it is possible to end up in a deadlock. This case is trivial to demonstrate, consider the simple case where there exists a graph of size two. Let’s state that when performing the version of Louvain without updates and relying on stale data there is a $\mathrm{\Delta }Q>0$ at both vertices, this means that both will perform a move to each others community. Following this move, it becomes clear that the next step could also simply be to repeat this swap, this means that some method is needed to prevent non-termination.

Given our usage of differential dataflow, we are able to mitigate some non-termination behaviour relatively cheaply. When we are about to make a decision to move to another community, we do this by examining the modularity gain of moving to the vertex of one of our edges. We can simply perform a join to determine where our neighbour would move, but more importantly we are then also able to evaluate the modularity gain for both of these operations and only perform the community transfer which maximises global modularity. While this may seem counter-intuitive since we limit the community transfers made, it mitigates both non-termination negative gain scenarios to a certain extent, unfortunately it does not however prevent non-termination nor negative-gain. The swapping technique above does not prevent non-termination. It is able to mitigate non-termination when considering two vertices however, when the number of vertices that perform cyclic swaps increase, it is obviously unable to account for such a scenario. Consider the case of a set of vertices where each vertex moves in a circular repeating pattern between communities, lets label these vertices as $i,j,k$ such that $i\in C(j),j\in C(j),k\in C(k)$ and $C(i)\ne C(j)\ne C(k)$. Consider the case where $i$ considers moving to $C(j)$, $j$ considers moving to $C(k)$ and $k$ considers moving to $C(i)$ simultaneously. In this case these vertices will see the global modularity as being $Q_{ne{w}_i}=Q_{old}+\mathrm{\Delta }{Q}_{i\to C(j)}$, $Q_{ne{w}_j}=Q_{old}+{\mathrm{\Delta }}_{Q_{j\to C(k)}}$ and $Q_{ne{w}_k}=Q_{old}+\mathrm{\Delta }{Q}_{k\to C(i)}$ respectively. The critical error here is obvious, these new values of $Q$ are completely wrong and could potentially lead to non-termination, the real value of $Q$ is in fact given by $Q_{new}=Q_{old}+\mathrm{\Delta }{Q}_{i\to C(j)}+\mathrm{\Delta }{Q}_{j\to C(k)}+\mathrm{\Delta }{Q}_{k\to C(i)}$. Note how in the above formula a determination of one transfer directly impacts the values used in the computation of $\mathrm{\Delta }Q$ in other transfers, this is why Louvain is traditionally a sequential algorithm.

Ultimately deadlocks occur when vertices continuously move around its neighbours and are also able to move back to their initial community structure, following this revelation it becomes easy to recognise that this implies that deadlocks occur in connected components although further work should be attempted to prove if this is exactly the case. A simple yet naive approach to this problem could be to simply agglomerate connected components into a single structure, however this is not always a viable approach as there exists graphs with the number of nodes in communities being smaller than the number of nodes in connected components, as a result using this approach would simply break our community detection algorithm for some graphs. As a result of our graph dataset containing graphs such that the size of one of its strongly connected components was the same size of the graph itself, this was obviously not a practical option.

Ultimately, our method is not dissimilar to the traditional Louvain [6] algorithm for modularity optimisation. The algorithm is expressed as a dataflow program, however the equation to measure both modularity and modularity gain remains the same as the traditional algorithm. One fundamental restriction that had to be made in our implementation however, is to order our nodes, this was quite simply achieved by assigning an integer to every node, this is ultimately due to implementation details of the open source library that was used to implement our algorithm. Like the initial Louvain algorithm, this modification is able to scale up to networks that measure in the millions of edges and is only limited by the memory it consumed. It is important to note that Differential Dataflow implements logic simply by running it until the output data converges to a fixed point. Our algorithm only computes the first phase and the Louvain algorithm is applied repeatedly until a fixed point is reached, modifications such as custom stopping criteria can still be made through simply applying the identity function to the various mappings to force convergence even when the stopping criteria is probabilistic. The algorithm works by first assigning a unique community to every node in the graph, that is, the number of nodes and communities are equal at the start of the first iteration. For each node $i$ in the graph, we consider the neighbours $j$ of $i$ and obtain the modularity gain ($\mathrm{\Delta }Q$) of moving it from its community and to the community of its neighbour, when performing this we need to keep track of the highest gain in modularity. After we have determined the highest gain in modularity, we perform the act of moving the node over to its neighbours community if $\mathrm{\Delta }Q>0$, otherwise the node remains in its current community.

Unfortunately as we have mentioned previously, there is no guarantee on convergence of our Louvain based method, in order to artificially induce convergence, we opted for a quick and easy method and that was to simply apply the identity function to a set of node, community pairs in the place where we would normally apply the clustering function. This is provides us with easy termination behaviour.

To our knowledge, there is no other method of combining incremental and iterative computations to dataflow computations with the exception of Differential Dataflow. Apache Spark and the Pregel based abstraction layer exposed via GraphX, isn’t completely at a disadvantage here however, since GraphX does provide incremental view maintenance capability along with other optimisations listed below in section 4.1.2.

Unfortunately, differential dataflow does not have an advantage in some tasks of dataflow computations, largely as a result of the of features present in Spark that Differential Dataflow lacks, at the end of the day, it is important to note that Differential Dataflow is an academia oriented library/framework where Apache Spark has been improved on over several years with major companies in industry using it for their ‘big-data’ processing needs. The lack of these features unfortunately negatively impacts the performance of Differential Dataflow, these potential areas of improvement for Differential Dataflow should be taken into account when comparing results. The list of features available in Apache Spark’s GraphX abstraction are listed below for the purposes of providing more perspective:

• •

  Index Reuse

• •

  Vertex Mirroring

• •

  Multicast join

• •

  Partial Materialisation

• •

  Incremental View Maintenance

• •

  Automatic Join Elimination

• •

  Variable Integer Encoding

More such optimisations and details regarding what these optimisation enable in GraphX are presented in extensive detail in [27]. Note that some of the above optimisations will not necessarily improve performance with Differential Dataflow, for example the Incremental View Maintenance optimisation in GraphX.

The datasets were obtained from the set of graph databases available from the Stanford large graph database collection ¹¹1See https://snap.stanford.edu/data/. These graphs varied in size and included fairly large datasets such as the Amazon product co-purchasing dataset.

For both processes we used the maximum number of cores available (8). Although we did test on different number of cores, these did not reveal much insight and therefore were not explored in our results.

Here we explore the different results obtained for batch processing of graph data using Apache Spark and Differential Dataflow.

We now would like to measure the cost of a single insert in an already clustered graph, we can simply measure this through evaluating the time taken for a computation to stabilise after an edge pair is inserted to an already clustered graph (to a single iteration). The given table shows the performance for the initial graph clustering and also the cost of a single insertion.

The usage of Differential Dataflow and to be more precise the result of parallelising Louvain in the manner described previously did not for the most part have a significant impact on the modularity value. Although Differential Dataflow did demonstrate a somewhat lower modularity value for the Amazon product co-purchasing network. One possible explanation for this is that these communities aren’t always well defined and can jump from categories quite easily, however, this still wouldn’t explain why the Apache Spark implementation was able to offer a near 50% increase in modularity. It is important to note still that the Differential Dataflow based method was still able to offer a better modularity for five out of the seven graph datasets, showing that a Louvain method based on Differential Dataflow is effective.

From our results, it seems that there is a certain property of the graphs that seem to impact the overall stability of both the time to terminate and in some cases the modularity itself. Consider for example the AS-733, email-Eu-core and Malaria-HVR_3 datasets which stabilised and terminated competely by themselves without our non-termination prevention technique coming into play. In the case of email-Eu-core, it is not too dissimilar to Wikipedia-Chameleon, in this case we observe that Wikipedia-Chameleon takes about two seconds compared to the fast 378ms runtime of the email-Eu-core graph, while Wikipedia-Chameleon is a bigger dataset, it is not significantly bigger than email-Eu-core, this can be shown in more perspective through contrasting the runtime and size of the Wikipedia-Squirrel dataset with Wikipedia-Chameleon. It is difficult to make a claim regarding what property of a graph is responsible for the stability of the communities, however, one likely heuristic that could be used to determine the graph stability is the edge to node ratio, we would like to make the speculation that a higher edge to node ratio generally reduces the probability of an unstable graph community structure but further research is needed here to provide evidence for this conjecture.

Unfortunately, one potential pitfall of using Differential Dataflow for the Louvain method is the degradation in performance when the advantages of using Differential Dataflow are not present and our implementation degrades to a slower version of the same algorithm due to us paying the runtime costs of Differential Dataflow. This occurs when there is a large domino effect due to a change in the classification of a vertex, simply put this change may induce another change which may induce another and so on. This is ultimately related to the Graph Stability mentioned in section 5.2.2. This is difficult to prevent as it is difficult to determine if these induced changes contributed to increasing the modularity of our graph in a meaningful way or not.

From the above results, a Differential Dataflow based Louvain method may be appropriate for some usage in real-time systems for some graph structures, but there may be some graph that has highly unstable community structures which would cause a Differential Dataflow based Louvain method to be far slower than ideal, perhaps even slower than a sequential non-incremental version. Further investigation here is warranted to investigate how to determine what exactly determines the stability properties of a graph’s community structure.

Unfortunately at this stage, we believe that Differential Dataflow is not completely ready yet to be used in general purpose incremental programs without willing to pay a significant software engineering cost as a result of the inherent complexity of using a hardly documented, research piece of software. An interesting approach to solving this problem could be perhaps to expose the ability to perform incremental computations at a higher abstraction layer, this is very much inline with the approach taken by the other ‘Big-Data’ processing frameworks such as Apache Spark. This hypothetical implementation also may gain an advantage due to its incremental computational model relying on a partial ordering of entries, although this should be explored in more detail, the advantage could be the ability to avoid implementing a variant of the Chandy-Lamport algorithm which is needed to ensure recovery from failures.

Unfortunately, even certain trivially incremental operators such as ‘max’ and ‘min’ are not available through the open source implementation, future work could examine how more operators could be added to Differential Dataflow in order to increase its expressive power.

While we have demonstrated how Differential Dataflow could be used to implement Louvain modularity optimisation, we have not explored the efficacy of other graph clustering methods, it is likely worth the effort to explore how other graph clustering algorithms could be implemented in such a way that it is able to take advantage of incremental computations, it may be the case that some traditionally more costly algorithms end up being significantly cheaper when differential dataflow is used.

If we are able to generalise even more on the partial ordering of time implemented in differential dataflow, we could use this to provide movement of already agglomerated vertices between clusters, what this means is that there is potential for modularity greater than what Louvain would normally find. The change here would be to represent time as vector of length three, this is not too different from the existing implementation, the additional entry here would simply represent the number of times a given node has been agglomerated. When new nodes are added, we would be able to examine if a certain vertex belongs in a different community due to this change, when the answer to this question is yes, a certain amount of work would be performed to propagate these changes throughout the hierarchy, the important distinction is that when communities remain the same, this work would be avoided entirely.

While “Self Adjusting Computations” are not directly usable for the computation of dataflow programs [19], there is potential for unifying “Self Adjusting Computations” for the purpose of incremental reductions. As mentioned previously the ‘reduce’ operator does not have any incremental properties due to the expressiveness of the operator, using “Self Adjusting Computations” here could prove to be a significant advantage in incrementally performing our reductions, this would reduce our max $Q$ determination from $𝒪(n)$ to $𝒪(1)$ in the best case, where $n$ is the number of neighbours.