Towards Modeling of Distributed Graph Algorithms

The most prominent approach for distributed processing of very large graphs today is probably the Bulk Synchronous Parallel (BSP) model. BSP is a bridging model for implementing graph algorithms in such a way that they can be massively parallelized and distributed. One application can be found in the Henshin graph transformation tool, where graph transformation systems can be modeled, and code generated for the BSP framework Apache Giraph.

Although the BSP model is widely accepted and used today, there is no standard implementation framework for BSP algorithms. Frameworks that do provide BSP include Google's Pregel, Apache Giraph and GraphX in Apache Spark. For application developers it is not easy to find out which platform is the most suited one for her problem. An implementation of a graph algorithm in, say, Apache Giraph cannot be reused in other frameworks, even though the underlying concepts of BSP are the same. This is unfortunate, particularly because the overhead of developing, deploying and testing these algorithms is rather high due to the complexity of the distributed frameworks.

This problem can be solved by introducing a modeling layer for BSP algorithms. Instead of directly implementing graph algorithms in Pregel, Giraph or GraphX, the idea is to use a modeling language that supports the concepts of BSP. One possible approach is to use UML 2 state machines for the modeling. The figure below shows a state machine for a BSP-model of the shortest path algorithm.

Using such a model, one could generate code for different platforms, e.g. Pregel, Giraph or GraphX. The code generators need to be implemented only once and are straight-forward to build since the BSP concepts are more or less directly used. Using different code generators, algorithm engineers can automatically derive implementations for different platforms and benchmark them against each other without implementing a single line of code.

In Henshin, the code generation for Apache Giraph is quite complex because the concepts of graph transformations are rather different than BSP. One opportunity to simplify it would be to use BSP models as intermediate target language. Specifically, model transformations can be employed to translate a Henshin model into a BSP model. From that point on, platform-specific code generators could be used to generate the final implementation.

Using a modeling approach would enable the reuse of the many many graph algorithms already implemented in the existing BSP-frameworks. Maybe UML state machines are not the best approach for the modeling -- for instance, one could also come up with a (textual) DSL for BSP. The important point is that an abstraction from the actual implementation platforms is made.  

Henshin 1.0 (Update)

8 years after the first PoC and 4 years after its official launch, we are proud to announce the release of Henshin 1.0 -- a model transformation language and framework for EMF. Henshin 1.0 is ready for Eclipse Luna and Java 8 (including its new JavaScript engine Nashorn). The primary domain of Henshin is model-driven development and formal verification methods. With its new multi-threaded pattern matching engine and its code generator for Apache Giraph, big data is a new third main domain of Henshin.

The new version of Henshin comes with an extended transformation language with support for specifying Java imports and annotations, and a couple of new utility APIs. Besides the changes under the hood, the 1.0 release also comes with a number of visible new features, including a reworked interpreter wizard and new preferences for the diagram editor. As an eye candy, we also added a new color mode for the diagram editor which goes well together with Luna's Dark theme.

You can obtain Henshin 1.0 from the release update site or clone our Git repo to try it out!

Update from 30-07-2014

And of course we've build a critical bug into the release! Anyway, the fixed version is available at the release update site.

Transformation Tool Contest 2014 -- Call for Solutions

The 2014 Transformation Tool Contest (TTC) seeks your solutions to two challenging transformation problems, involving transforming financial XML models into object-oriented code, and transforming the very large model that underpins the IMDb movie database.

Select your favorite transformation tool and join the contest! The deadline for solution descriptions is 23 April 2014. More information, including links to the problem descriptions, at: http://www.transformation-tool-contest.eu/cfs.html

Parallel Graph Pattern Matching in Henshin (Update)

In my last post, I reported on an improvement of the graph pattern matching algorithm in Henshin that showed a major performance boost for rules where the nodes in the left-hand sides are connected only by <<require>> nodes and edges. As a concrete example, we used the transformation rule below to find couples of actors / actresses in the IMDB movie database which starred together in at least three movies.

Thanks to the tweak in the pattern matching engine we were able to apply this rule to much larger input graphs. Unfortunately, the synthetic test data that we used for our last benchmark does not have the same complexity as the real data from IMDB. In our test data, every person played in not more than 10 movies and every movie had a similarly low number of actors / actresses. In reality, the IMDB data contains movies with more than 1000 actors and obviously also persons that played in a large number of movies.

So I executed our find-couples example on the real IMDB data (thanks for the data import, Tassilo!). Not surprisingly, the results were much worse than for the synthetic data. The reason for this is that the real IMDB data is much more interconnected. So I started tweaking the pattern matcher of Henshin again. Eventually, I realized that I need something more to crack this nut.

SAP Coding Masters 2013

A possible way to improve the performance is to parallelize the graph pattern matching algorithm. In the current development version of Henshin, this is actually implemented! Here is how you can enable it:

int partitions = 10;
EGraph graph = new PartitionedEGraphImpl(model,partitions);
Engine engine = new EngineImpl();
engine.getOptines.put(Engine.OPTION_WORKER_THREADS,
   partitions);

// Execute transformation

engine.shutdown();

You need to use the new implementation of EGraph called PartitionedEGraphImpl. It basically splits up the graph to be transformed into N disjoint partitions. The second change you need to do is to set the number of worker threads in the transformation engine. In theory, this number can be different to the number of partitions, but the easiest setup is to just use the same amount of threads as there are partitions. The third point you need to ensure is that your rule uses *-stereotypes for nodes and edges (essentially it transforms all matches at once). Finally, you should shutdown the engine when your done to get rid of the worker threads.

For the example of finding couples in the IMDB database, this parallel graph pattern matching approach works actually quite well. Below you can see the performance gains achieved on an ordinary desktop computer.

With this new parallel graph pattern matcher in the interpreter, and the recently introduced Giraph code generator, Henshin really is becoming a big-data tool, which I find very exciting.

Update (07 February 2014)

I realized that the performance in absolute number is still quite bad. As I wrote in my earlier post, the optimized match finder uses path constraints to check the existence of the movies which connect the actors. The problem was though that the match finder still made an additional check whether the <<require>> elements exist, i.e., it performed a complete matching of the PAC (positive application condition). This is actually not necessary here because the path constraint is sufficient already.

To fix this problem, I added a static analysis which finds out once in the beginning whether path constraints are enough to check the PAC. This significantly reduces the run-times as you can see below. Using 12 worker threads, the processing of 2 million nodes takes 173 seconds. But to be honest, I have my doubts whether these engine optimizations will be helpful also in other cases.

Nodes	Couples	Time (ms)
499,995	79,470	6,582
1,004,463	244,627	27,239
1,505,143	585,689	87,721
2,000,900	1,017,752	173,142

Path Constraints in Henshin

In our recent paper on the Giraph code generator in Henshin, we conducted some benchmarks using a real-data example. For this we used the publicly available data sets from the IMDB movie database. We parsed the movie, actor and actress records into an EMF model as shown below. Note that I omitted the attributes here for simplicity.

The goal of our transformation on the IMDB data set was to find all pairs of persons (actors or actresses) which starred together in at least three movies. For every such pair we want to create an instance of the Couple class. We used the Henshin rule below to do this. First, note that all nodes / edges have *-actions and thus the rule is applied to all found matches at once. Second, we used <<require*>>-actions for the movies, and not <<preserve*>>. This is important because we want to ensure the existence of these movies, but be don't want to generate separate matches for every combination of movies. There is only one small subtlety in this rule: due to the symmetry of person nodes, we will actually create two couple nodes for every pair. But this could be easily fixed by enforcing an alphabetical order of the persons' names using an attribute condition.

To compare the performance of the Giraph-based engine, we also tried to execute this example with the normal Henshin interpreter. To our surprise, even for a small part of the IMDB data set, the Henshin interpreter was not able to compute a match. To study the problem in a more controlled setting, we defined a small transformation that generates a synthetic test data set of variable size. We then ran the interpreter again and got the following numbers.

Nodes	Couples	Time (ms)
1,700	200	1,625
3,400	400	3,753
5,100	600	7,828
6,800	800	14,386
8,500	1,000	23,301

This looks pretty bad. The input graphs are very small (< 10,000 nodes) and it already takes more than 20 seconds. The main problem is that the execution time grows non-linearly (should be O(n²)) with the size of the input graph. What is the reason for this? Well, the Henshin interpreter first fixes a match for the LHS (<<preserve>> and <<delete>> elements), and after that it tries to match the application conditions (<<require>> and <<forbid>> elements). In our example, this means that it first picks to arbitrary persons, and then checks whether they starred together in 3 movies. Thus. it is basically a brute-force approach that the interpreter runs here.

The pattern matching is implemented in Henshin as a constraint-solving problem. To fix the bad performance of the movies example (and similar problems), we added a new type of constraint to the pattern matching algorithm: path constraints. Path constraints are internally used to check whether there exists a path between two EObjects along a given list of EReferences. The neat thing about these paths is that they can also go via <<require>> nodes. In our example, this indeed works because the movies-reference has an opposite reference: persons. If we now fix one of the two persons, the path constraint via the references [movies, persons] allows the match finder to restrict the solution set for the second person to only those persons who played together with the first one in at least one movie. This drastically prunes the search space which gives us a much better performance.

Nodes	Couples	Time (ms)
1,700	200	133
3,400	400	89
5,100	600	35
6,800	800	36
8,500	1,000	49
...	...	...
170,000	20,000	1,419
850,000	100,000	6,183
1,700,000	200,000	9,836

This looks much better: now we can process 1.7 million nodes in less than 10 seconds. This should be also enough to run the example on the original IMDB data set. The example including the benchmark is part of the examples plug-in of the development version of Henshin.

Henshin + Giraph: Enforced Parallelism

I recently announced the availability of a Giraph code generator for Henshin, which allows you to execute graph transformations in a parallel and distributed manner. It is important to know that there is a fundamental difference in the execution semantics of graph transformations in this approach. Concretely, the application of rules is always maximum parallel, i.e., every rule is always applied to all found matches in parallel. The idea behind this is that we want to enforce the use of parallelism already at the modeling level and to make it hard for developers to use conventional sequential algorithms that cannot be parallelized automatically. For example, the rule AddTriangles below would be applied to all found triangles in parallel. Executing the iterated unit BuildSierpinski to a 3-node triangle therefore generates the full Sierpinski triangle of depth N.

Due to the change in the execution semantics of rules, other existing and also some new modeling concepts specifically tailored for parallel graph transformations become important. For example, vertices and edges with <<require>> and <<forbid>> stereotypes are particularly useful to avoid overlapping matches which could result in inconsistent parallel rule applications. We also introduced concepts that allow you to aggregate over attribute values of all found matches. I will discuss this in more detail in another post.

Henshin + Giraph = Big Data Graph Transformations

Henshin is a graph and model transformation tool for Eclipse that lets you define rules and workflows for transformation in a graphical editor. By default, these transformations are executed using an interpreter on an in-memory EMF model.

Apache Giraph is a distributed and parallel graph processing framework built for high scalability. Giraph is for instance used at Facebook to analyze the underlying graphs of social networks.

So how do Henshin and Giraph fit together? Easy: Henshin provides an expressive graph transformation language with an intuitive graphical syntax. Giraph provides an infrastructure for highly parallel and distributed graph processing. The sort of obvious way of combining the two is to use Henshin as modeling tool and Giraph as execution engine. And that is what we are currently working on.

We implemented a code generator that produces Java code for Giraph from Henshin models. This generated code contains pattern matching code and graph manipulations for rules, and the required coordination to execute transformation units (workflows). The code generation is still an experimental feature but we plan to stabilize it and ship it with the next release of Henshin. We have already a small test suite and conducted some promising benchmarks. More details later.

The Giraph code generator is available in the development version of Henshin (for Eclipse Kepler). You can get it from our nightly build update site or by getting the source code directly.