# Programming in Arabesque

Arabesque simplifies the programming of Graph Mining Problems as presented in our paper. In the paper, we describe the system and provide a comprehensive introduction to the concepts that we describe below.

We show how Arabesque can be used to solve three fundamental problems in Graph Mining. Finding cliques, counting motifs and frequent subgraph mining. We chose these problems because they represent different classes of graph mining problems. Finding cliques is an example of dense subgraph mining, and allows one to prune the embeddings using local information. Counting motifs requires exhaustive graph exploration up to some maximum size. Frequent subgraph mining is an example of explore-and-prune problems, where only embeddings corresponding to a frequent pattern need to be further explored. We discuss these problems below in more detail.

## Filter-Process Model: Think Like an Embedding

Our paper is the best introduction to the filter-process model, or "Think Like an Embedding" (TLE). We now summarize the TLE model, which is depicted below in its most basic form.

The systems exhaustively explores all subgraphs in the graph. We use the term **embedding** for a subgraph. Exploration proceeds in exploration steps. At each step the existing embeddings are expanded by one new neighboring vertex or incident edge, depending if the application wants vertex-based or edge-based exploration. If we add vertices, then every time a new vertex is added to an embedding, all the edges between the new vertex and the ones currently in the embedding are also added to the embedding, i.e., subgraphs are vertex-induced. If we add edges, then every time a new edge is added to an embedding, the vertex at the other end of the edge is also added (if it wasn't already in the embedding).

Each newly generated embedding, called candidate embedding, is streamed through a user-defined **filter** function, which returns a boolean saying whether the embedding is interesting or not. If the embedding is interesting, it is streamed over a user-defined **process** function, which typically derives some output from the embedding. After the process function, the new embedding is added to the set of embeddings to be expanded on the next exploration step. Before that, the embedding is streamed through an optional user-defined termination filter function, which is true by default and operates similarly to a filter function.

Applications that need to aggregate values across multiple embeddings can also define optional aggregation functions. Their operation is depicted below:

The process function can call a **map** function, which sends a key-value pair *(k, v)* to a certain aggregator. This aggregator then runs a user-defined **reduce** function to merge all *(k, v)* pairs with the same key *k*. The aggregated values become available for reads by user-defined functions at the end of the exploration step. Before the filter function, Arabesque executes two user-defined functions that can take advantage of aggregated values: **aggregate-filter** and **aggregate-process**, which are analogous to the filter and process function. These two additional functions allow pre-filtering and pre-processing based on values aggregated in the previous steps.

## Arabesque API

Let us see how to create a new computation in Arabesque. Arabesque operates in two modes: vertex-induced and edge-induced expansion. For algorithms using vertex-induced expansion, it is necessary to create a new class that extends *VertexInducedComputation*; for algorithms using edge-induced expansion, the new class must extend *EdgeInducedComputation*. Implementing a computation entails overriding methods of the parent class. Both parent classes extend the following interface (we report only the most important methods).

```
public interface Computation<E extends Embedding> {
void init();
boolean filter(E embedding);
void process(E embedding);
boolean shouldExpand(E newEmbedding);
void initAggregations();
boolean aggregationFilter(E Embedding);
void aggregationProcess(E embedding);
}
```

The class *E* can be either *VertexInducedEmbedding* or *EdgeInducedEmbedding*, depending on the type of computation we are using.

Computation classes can have a local state, which will be replicated at each Arabesque worker thread running the job. The init method initializes this internal state (it must start with a call to *super.init()*). The filter method executes the filter function and the process method executes the process function.
Finally, the method shouldExpand implements the termination filter function. These methods are sufficient to implement the basic filter-process model.

Algorithms using aggregation can also override the aggregation filter and aggregation process functions. In order to use aggregations, applications need to register them by specifying the key and value classes, as well as the reduction function. This must be done in the initAggregations method, which typically looks like the following example.

```
@Override
public void initAggregations() {
super.initAggregations();
Configuration conf = Configuration.get();
conf.registerAggregation(
"my-aggregation", // Aggregation name
KeyClass.class, // The class of key objects
ValueClass.class, // The class of value objects
false, // Is this aggregation persistent (i.e an output aggregation)
new ReductionFunctionImplementation() // This specifies the reduction function to use.
);
}
```

The method registerAggregation takes as input the specification of the reducer of an aggregation. Arabesque supports using multiple aggregations. Each aggregation is associated with a unique name (a String), has multiple keys, and is processes independently from the others. In the example, the aggregation is called "my-aggregation". The following two arguments are the classes of keys and values, respectively, which must implement the *Writable* interface. The third argument is a boolean with value *true* if the aggregated values are persisted until the end of the computation (in this case we talk about an *output* aggregator); a value *false* indicates that values are recomputed anew at each exploration step. Finally, the last argument indicates the user-defined class implementing the reducer logic. This class must implement the following simple interface, to aggregate two values into one:

```
public interface ReductionFunction<V extends Writable> {
V reduce(V k1, V k2);
}
```

Applications can map values to a reducer by invoking the `void map(String name, K key, V value)`

method and read aggregated values using `AggregationStorage<K, V> readAggregation(String name)`

.

## Application Examples

We now show implementation examples for three applications: finding cliques, counting motifs, and frequent subgraph mining.

### Finding Cliques

Finding cliques has many variations. Here, we present the variation of finding cliques of a maximum size. Following, we have the implementation in Arabesque:

```
public class CliqueComputation
extends VertexInducedComputation<VertexInducedEmbedding> {
private static final String MAXSIZE = "arabesque.clique.maxsize";
private static final int MAXSIZE_DEFAULT = 4;
int maxsize;
@Override
public void init() {
super.init();
maxsize = Configuration.get().getInteger(MAXSIZE, MAXSIZE_DEFAULT);
}
@Override
public boolean filter(VertexInducedEmbedding embedding) {
return isClique(embedding);
}
private boolean isClique(VertexInducedEmbedding embedding) {
return embedding.getNumEdgesAddedWithExpansion()
== embedding.getNumVertices() - 1;
}
@Override
public void process(VertexInducedEmbedding embedding) {
if (embedding.getNumWords() == maxsize) {
output(embedding);
}
}
@Override
public boolean shouldExpand(VertexInducedEmbedding newEmbedding) {
return newEmbedding.getNumWords() < maxsize;
}
}
```

In Arabesque, the user must define the Computation class for the problem at hand. In this particular case, we have defined the class **CliqueComputation** that extends the **VertexInducedComputation**, which dictates that the exploration Arabesque will perform is vertex-induced. As we explained before, this exploration extends an embedding by connecting it to a new vertex at a time. All edges connecting that new vertex to existing vertices of the embedding are also added.

To solve the problem and control the exploration the user defines two main functions. The **filter** function decides whether the passed embedding is a valid clique and, thus, if it should be further expanded and processed. The **process** function dictates what to do with the embeddings that passed the filter function. In this particular problem, we only want to output the embeddings. Note, that the complexity of the exploration and the required checks for avoiding redudant work, and the canonicality checks are completely transparent to the end-user. This filter function trivially respects the anti-monotonicity property as if a parent is not a clique, there's no chance for any of its children to be cliques.

For performance reasons, the **filter** is implemented in an incremental way. The current embedding was constructed by extending a parent embedding with some new vertex and all the edges that connect it to existing vertices in the embedding. Thus we can efficiently implement the filter function by only considering whether the vertex we added connects to all previous vertices in the parent embedding. This can be done very easily, by getting the number of new edges added to this embedding *getNumEdgesAddedWithExpansion()* and checking if this number is equal to the number of vertices in the parent embedding which by default is one less than the current embedding.

To produce the output, we simply check whether we have reached the size of cliques we are interested in. The *output* function saves that embedding to HDFS.

To avoid going to deeper depths than the one we want to discover, the user can override the optional **shouldExpand** function that checks whether we need to further expand the embedding and thus continue the exploration. By default, this function returns true, and by overriding it we can stop the processing faster.

### Counting Motifs

A motif is defined as a connected pattern of vertex induced embeddings that exists in an input graph. Further, a set of motifs is required to be non-isomorphic, i.e., there should obviously be no duplicate patterns. In motif mining there is no minimum frequency threshold; rather the goal is to extract all motifs that occur in the graph along with their frequency distribution. Since this task is inherently exponential, motifs are typically restricted to patterns of order (i.e., number of vertices) at most k. For example, for k = 3 we have two possible motifs: a chain where ends are not connected and a (complete) triangle. Following, we have the Motif counting implementation in Arabesque:

```
public class MotifComputation
extends VertexInducedComputation<VertexInducedEmbedding> {
private static final String AGG_OUTPUT = " output";
private static final String MAXSIZE = "arabesque.motif.maxsize";
private static final int MAXSIZE_DEFAULT = 4;
private static LongWritable reusableLongWritableUnit = new LongWritable(1);
private int maxsize;
@Override
public void init() {
super.init();
maxsize = Configuration.get().getInteger(MAXSIZE, MAXSIZE_DEFAULT);
}
@Override
public void initAggregations() {
super.initAggregations();
Configuration conf = Configuration.get();
conf.registerAggregation(
AGG_OUTPUT,
conf.getPatternClass(),
LongWritable.class,
true,
new LongSumReduction()
);
}
@Override
public void process(VertexInducedEmbedding embedding) {
if (embedding.getNumWords() == maxsize) {
map(AGG_OUTPUT, embedding.getPattern(), reusableLongWritableUnit);
}
}
@Override
public boolean shouldExpand(VertexInducedEmbedding newEmbedding) {
return newEmbedding.getNumWords() < maxsize;
}
}
```

The Computation class, **MotifComputation**, extends the **VertexInducedComputation**, which dictates that the exploration that Arabesque will perform is Vertex Induced in a similar way to finding cliques.

For counting Motifs, we are interested in all possible variations of the embeddings and thus we don't need to implement a special filter function. The default filter function returns true and is sufficient for this problem. For the process implementation, we need to compute for every embedding the pattern it corresponds to (motif for this problem), and then group it over all embeddings to compute the frequencies of the motifs. In Arabesque, this is performed through the user of aggregations, which work similarly to Giraph's Aggregators, and allow one to compute aggregate statistics on a pattern level or any other arbitrary key. In the process function, we simply generate a new unit key-pair containing the pattern that this embedding corresponds to as the key and 1 as the value. To be able to use an aggregation, we first need to initialize it. This is done in the **initAggregations** function. Here, we define the key and value types as well as how to reduce values belonging to the same key. For motifs, a simple sum reduction is sufficient.

The shouldExpand behaves identically to the Finding Cliques computation, terminating the expansion when the desired depth has been reached.

### Frequent Subgraph Mining

Frequent subgraph mining (FSM) focuses on finding patterns whose frequency in the input graph is above a certain threshold. The FSM task outputs all frequent patterns and their respective embeddings (i.e., matching subgraphs) from a single input graph.

In our implementation, frequency is counted using an anti-monotonic metric called "Minimum Image-Based" (MNI) support (see B. Bringmann and S. Nijssen. "What is frequent in a single graph?" PAKDD 2008). We chose this metric because it is anti-monotonic. By contrast, simply counting the number of subgraphs that match a pattern is not an anti-monotonic frequency metric. According to such metric, the expansion of an infrequent pattern may be frequent. We refer the reader to our SOSP paper for a precise description of the MNI metric.

In frequent subgraph mining, we use aggregation to calculate the frequency/support of each pattern we see. The frequency/support metric is based on the notion of domain, which is defined as the set of distinct mappings between a vertex in a pattern and the matching vertices in any automorphism of an embedding. The **process** function maps the domains of an embedding to its respective pattern. The *DomainSupportReducer* function merges 2 domains into one by merging all internal mappings for each vertex position. Note that in this example, we use one more parameter when registering the aggregation. This is called an *end aggregation* method, which is invoked every time aggregation completes. This method is useful for cases where we have a very large number of keys, as it happens with FSM. Finalized aggregated values, in fact, are broadcasted to all workers in the system. End aggregation methods can be used to eliminate keys whose value is useless and that do not need to be further use, and therefore broadcasted. In this example, the *DomainSupportEndAggregationFunction* function will look at all aggregated (Pattern, DomainSupport) pairs and only keep those where the DomainSupport is above a predetermined threshold. This is a performance optimization to prevent having to broadcast a large amount of pairs that are of no value to us because we already know they are not frequent. The **aggregationFilter** function then consults the surviving key-value mapping to see if the pattern of the current embedding is contained in that mapping. If it is, that pattern is frequent so the current embedding should be processed. This is then done via the **aggregationProcess** function which outputs all embeddings that survive the aggregation-filter.

```
public class FSMComputation
extends EdgeInducedComputation<EdgeInducedEmbedding> {
public static final String AGG_SUPPORT = "support";
public static final String CONF_SUPPORT = "arabesque.fsm.support";
public static final int CONF_SUPPORT_DEFAULT = 1000;
public static final String CONF_MAXSIZE = "arabesque.fsm.maxsize";
public static final int CONF_MAXSIZE_DEFAULT = -1;
private DomainSupport reusableDomainSupport;
private AggregationStorage<Pattern, DomainSupport> previousStepAggregation;
private int maxSize;
private int support;
@Override
public void init() {
super.init();
Configuration conf = Configuration.get();
support = conf.getInteger(CONF_SUPPORT, CONF_SUPPORT_DEFAULT);
maxSize = conf.getInteger(CONF_MAXSIZE, CONF_MAXSIZE_DEFAULT);
reusableDomainSupport = new DomainSupport(support);
previousStepAggregation = readAggregation(AGG_SUPPORT);
}
@Override
public void initAggregations() {
super.initAggregations();
Configuration conf = Configuration.get();
conf.registerAggregation(
AGG_SUPPORT,
conf.getPatternClass(),
DomainSupport.class,
false,
new DomainSupportReducer(),
new DomainSupportEndAggregationFunction()
);
}
// Previous Step embeddings
@Override
public boolean aggregationFilter(EdgeInducedEmbedding previousStepEmbedding) {
// Using the DomainSupportEndAggregationFunction, we
// removed all mappings for non-frequent patterns.
// So we simply have to check if the mapping has
// the pattern for the corresponding key
return previousStepAggregation.containsKey(previousStepEmbedding.getPattern());
}
@Override
public void aggregationProcess(EdgeInducedEmbedding previousStepEmbedding) {
output(previousStepEmbedding);
}
// Current step embeddings
@Override
public void process(EdgeInducedEmbedding embedding) {
reusableDomainSupport.setFromEmbedding(embedding);
map(AGG_SUPPORT, embedding.getPattern(), reusableDomainSupport);
}
@Override
public boolean shouldExpand(EdgeInducedEmbedding newEmbedding) {
return maxSize < 0 || newEmbedding.getNumWords() < maxSize;
}
}
```

The Computation class **FSMComputation** extends the EdgeInducedComputation class, which determines that the exploration is edge induced, a different exploration strategy to the previous two Applications. Similar to Motifs, no filter function is defined. The reasoning for such an omission, however, is different than in the case of Motifs. With FSM we cannot filter based on the information of a single subgraph alone and need to wait until the aggregated frequency values are ready to then filter the embeddings with the aggregationFilter function.

To obtain the aggregated frequencies, we use a non persistent aggregation. These behave in a similar manner to output aggregations with the difference that their values do not persist until the end of the execution. That is, at each different step, the content of interstep aggregators are reset.

The process function for FSM is responsible for creating the key-value pairs that, when aggregated, will result in (Pattern, DomainSupport) from which we can extract the set of frequent patterns.

Similar to the other algorithms, the user can define the optional shouldExpand function to prune explorations of very big subgraphs.

The FSM algorithm also contains an optional MasterComputation. This MasterComputation is called after the end of a superstep and allows you complete access to the values of the aggregations before the start of the next step. In this example, we use this access to output the frequent patterns that we see at each step as well as to terminate execution earlier if we immediately detect that there are no frequent patterns (and, therefore, all aggregationFilter calls will return false at the next step).

```
public class FSMMasterComputation extends MasterComputation {
@Override
public void compute() {
AggregationStorage<Pattern, DomainSupport> aggregationStorage =
readAggregation(FSMComputation.AGG_SUPPORT);
if (aggregationStorage.getNumberMappings() > 0) {
System.out.println("Frequent patterns:");
int i = 1;
for (Pattern pattern : aggregationStorage.getKeys()) {
System.out.println(i + ": " + pattern);
++i;
}
}
// If frequent patterns is empty and superstep > 0, halt
else if (getStep() > 0) {
haltComputation();
}
}
}
```

# Developing your own algorithms

The easiest way to start developing your own graph mining algorithms on top of Arabesque is to clone our Arabesque-Skeleton repository. This gives you a preconfigured Maven project with detailed instructions on how to get your code running in a cluster.