In the last few years there has been an increasing number of application elds, like the Semantic Web, social networks, bioinformatics, astronomical databases, etc., where large graph datasets are analyzed, queried, and, more generally, manipulated. Graphs are usually queried by specifying reachability patterns through regular path expressions; this leads to the need for e cient and scalable tools for processing regular path queries on large graphs. In this work we present a distributed implementation of GXPath and show that this implementation, built on top of Hadoop MapReduce, can scale linearly with the number of vertices and/or edges.
In the last few years there has been a growing interest in
querying and, more generally, analyzing huge graph datasets.
These datasets arise in many real-life contexts, such as
social and instant messaging networks, biology, crime
detection and prevention, etc. As an example, social networks
like Facebook and Twitter attract everyday more and more
users from all over the world. The di usion of these services
is so wide that Facebook reported that in the second quarter
of 2015 its network comprises more than 1.4 billion active
users [
Social network interactions can be naturally modeled by using directed graphs, whose vertices denote users and edges describe user-to-user interactions: indeed, Facebook and Twitter already use graphs and graph tools to model and analyze their networks.
Given the wide di usion of graph datasets, analyzing these graphs is becoming more and more important. This task, however, is very challenging, partly because of the exibility of graphs and partly because of their ever increasing size. In the following example, we describe a paradigmatic application of graph analysis related to crime detection and prevention. Example 1.1 Consider a government intelligence agency that must seek for terrorist cells. Unlike what happens for traditional criminal organizations, terrorists usually exploit Internet services by exchanging messages or uploading videos to make proselytizing campaigns, and they often leave evidences of these interactions. Security and intelligence agencies, hence, try to nd these evidences, by analyzing the large amount of information exchanged on the Internet and, in particular, on social networks like Facebook or Twitter.
The structure of terrorist cells can be usually translated into well-determined reachability patterns in social network graphs. A potential way to identify these cells, therefore, is to adopt an analysis technique that inspects social network graphs and looks for those vertices that are connected by paths matching these reachability patterns; the results of these queries can be further re ned by linking graph vertices to other databases, e.g., criminal records, and by manually inspecting collected results. Hence, while single users of a social network may not give rise to the suspicion of belonging to a terrorist cell, the links among them could take down the whole organization.
In Figure 1 we show the connections between the terrorists
that participated to the 9-11 attacks in Washington and New
York, and other people involved with terrorist activities [
Graph reachability patterns are usually expressed by means
of Regular Path Queries (RPQs) [
Unfortunately, traditional relational database systems are
not well-suited for this kind of queries and do not scale well
on large graphs. On the other hand, most existing graph
database systems are inherently centralized (e.g., Neo4j [
Our Contribution. In this paper we describe a system for processing GXPath queries on large data graphs. In this system, built on top Hadoop MapReduce, a query is compiled into an acyclic graph of MapReduce jobs, similar in spirit to a database query plan. Intermediate results are stored in a compressed format to decrease the I/O overhead implied by MapReduce. As proved by several experiments, our system scales linearly with the number of vertices and/or edges. Paper Outline. The rest of the paper is organized as follows. In Section 2 we describe the data model and the query language being used. In Section 3, then, we sketch our query processing technique. In Section 4, next, we present an extensive experimental evaluation of our system. In Sections 5 and 6, nally, we discuss some related works and draw our conclusions.
To describe a graph, we consider a model in which edges are labeled by symbols from a nite alphabet and vertices can contain data values from a countably in nite set D. For the sake of simplicity, we assume that a vertex can contain a single data value.
De nition 2.1 A data graph (over G = hV; E; i where:
E V
When we deal with purely navigational queries, we refer to a graph as G = hV; Ei omitting .
A path from a vertex v1 to a vertex vn in a graph is a sequence = v1a1v2 : : : vn 1an 1vn such that each (vi; ai; vi+1) for i < n is an edge in E. We use ( ) to denote the ordered concatenation of the labels of . 2.2
Most navigational formalisms for querying data graphs are based on RPQs and their extensions. An RPQ is an expression of the form x !L y, where L is a regular language over (typically represented by a regular expression or a NFA). Given a -labeled graph G = hV; Ei, the answer to an RPQ as above is the set of pairs of vertices (v; v0) such that there is a path from v to v0 with ( ) 2 L.
GXPath, proposed by Libkin et al. in [
In this paper we focus our attention on the navigational fragment of GXPath, without value tests, as described by the following grammar: := j j a j a j + j j j m;n j [ ] Here, denotes the empty word, is a wildcard matching any symbol, 1 2 and 1 + 2 are the standard concatenation and union operators. m;n denotes the repetition of from m to n times (m n; m 2 N; n 2 N [ f g); a , nally, denotes the backward navigation, is the complement of , and [ ] is a nested condition. The Kleene star can be represented as 0; .
Given a data graph G = hV; E; i, the semantics J KG of a query on G is a set of pairs of vertices de ned as follows: J KG J KG a G J K Ja KG J 1 2KG J 1 + 2KG J KG [ ]KG Jam;n G J K = f(v; v) j v 2 V g = f(u; v) j 9a 2 (_u; a; v) 2 Eg = f(u; v) j (u; a; v) 2 Eg = f(u; v) j (v; a; u) 2 Eg = J 1KG J 2KG = J 1KG [ J 2KG = V V J KG = f(kv=;vm)J2rKkGG j (v; u) 2 J KGg = Sn where is the concatenation of binary relations and Ri denotes the concatenation of R with itself i times. Example 2.2 Consider the graph depicted in Figure 2 and the query a0; [b]. This query returns all the pairs of vertices (x; y) where x and y are connected by an a-labeled path of any length and y has an outgoing b edge. In the case of the graph of Figure 2, this query returns f(v4; v4); (v2; v4); (v1; v4); (v3; v4)g.
Consider now the following query: a2;3 (b + d). This query returns all vertex pairs (u; v) connected by the following paths: aab; aaab; aad; aaad. The result of this query is f(v1; v6); (v1; v5); (v2; v5); (v3; v6)g.
: V ! D is a function that assigns data values to vertices.
In [
d
We rst illustrate our query processing technique through examples and, then, present in more detail the operators used in our system. 3.1
In our system, before processing queries on an input graph G = hV; E; i, G must be indexed. During graph indexing, the system creates a distinct binary relation per label, containing all the edges in E having that label. These relations are stored in the distributed le system and form the actual input data that will be analyzed. In the following, we will assume that the indexing phase returns a set R = fRa j a 2 g, where Ra = f(x; y) j (x; a; y) 2 Eg; more generally and with a little abuse of notation, we will use R and R to denote the result of the evaluation of with and without duplicates, respectively.
When a query is submitted to the system, it is rst translated into its abstract syntax tree. After a simpli cation phase, where common rewritings (e.g., = = ) are applied, the resulting AST is transformed in an acyclic graph of MapReduce jobs; each job consumes and produces binary relations of vertices.
Example 3.1 Consider again the query a2;3 (b + d) on the graph of Figure 2. This query can be translated in the AST of Figure 3.
In our system this AST is transformed in the query plan shown in Figure 4. Here, Symbol(a), Symbol(b), and Symbol(d) are operators that access the system catalog and return the location of binary relations Ra, Rb, Rd in the distributed le system. In the case of our input graphs, these relations have the following extensions:
Count(Ra; 2; 3), then, evaluates the bounded transitive
and re exive closure of Ra. To this aim, Count(Ra; 2; 3)
exploits a MapReduce implementation of the SemiNaive
algorithm described by Bancilhon in [
Union(Rb; Rd) accesses Rb and Rd, and returns their union.
The output of Count(Ra; 2; 3) and Union(Rb; Rd) is passed as input to the concatenation operator Concat(Ra2;3 ; Rb+d). Concat(R; S) takes as input two binary relations R and S, and returns their composition, by applying a standard matrix product algorithm. In our example, Concat(Ra2;3 ; Rb+d) returns the following binary relation: f(v1; v6); (v2; v5); (v1; v5); (v3; v6)g.
Finally, the output of Concat(Ra2;3 ; Rb+d) is sent to the duplicate elimination operator Distinct( ). In our system, indeed, query operators do not return sets of vertex pairs, as prescribed by GXPath set-based semantics, but collections that may contain repeated entries. Duplicates are eliminated at the end of query evaluation by relying on the Distinct( ) operator. We made this choice as early experiments showed that set-based operators were much slower than sequence-based ones.
Example 3.2 Consider now the query b [c] a1;3. This query rst looks for all vertex pairs (x; y) such that (x; y) 2 Jb [c] a1;3KG, and, then, computes the complement of Jb [c] a1;3KG (i.e., V V Jb [c] a1;3KG). In particular, in each pair (x; y) y is reachable from x by traversing backward an incoming b-labeled edge, by ltering out vertices without c-labeled outgoing edges, and by traversing a path labeled with a, aa, or aaa.
In the case of the graph of Figure 2, Jb [c] a1;3 G =
K f(v6; v3); (v6; v1); (v6; v2)g and Jb [c] a1;3KG = f(vi; vj ) j i = 1; : : : ; 7; j = 1; : : : ; 7g f(v6; v3); (v6; v1); (v6; v2)g.
To evaluate the result of this query, our system creates the query plan shown in Figure 5. Here, BackSymbol(b) accesses relation Rb, created during the indexing phase and stored in the distributed le system, and returns a relation Rb obtaining by adding a pair (vx; vy) for each pair (vy; vx) 2 Rb. Operator Cond(Rc), instead, just accesses relation Rc and outputs a relation R[c] containing a pair (vx; vx) for each (vx; vy) 2 Rc.
The Compl( ) operator, nally, computes the complement of the binary relation passed as input. To this aim, it just
Concat(Rb [c]; Ra1;3 ) generates all the pairs of vertices (u; v) 2 V V and, during the generation process, it discards those belonging to its input. As Compl() already suppresses duplicates by itself, the system does not introduce a Distinct() operator. 3.2
As already stated, each graph must be indexed before becoming available for querying. To this aim, our system preprocesses each graph by using a MapReduce job whose Map phase returns, for each edge (x; label; y), the key-value pair (label; (x; y)); during the Reduce phase, each reducer receives a pair (label; L), where L = f(x; y) j (x; label; y) 2 Eg, creates a compressed le for label, and add each pair (x; y) to the le. As a result, the indexer creates a collection of binary relations Ra, stored as compressed les on HDFS by relying on the Snappy codec of Hadoop. During this indexing phase, the system also records in the system catalog the number of edges and vertices of the graph (for the sake of simplicity, we assume that input graphs have no isolated vertices, i.e., vertices without incoming and outgoing edges). 3.3
Our system exploits several basic query operators. In this section we will discuss the most prominent ones, and assume, unless otherwise noted, that each operator reads and writes compressed les.
Epsilon. Epsilon() just returns a set of pairs (u; u), where u 2 V . This operator is implemented as a Map-only MapReduce job, where each mapper is in charge of generating pairs (u; u) for a given interval of vertices (vertices are uniquely identi ed by integers). By relying on the NLineInputFormat class of Hadoop, each mapper is forced to process exactly one vertex interval, i.e., the system creates as many mappers as intervals.
Symbol and BackSymbol. Symbol(a) takes as input a symbol a 2 , and returns the relation Ra. As input graphs are assumed to have been previously indexed, Symbol(a) just looks in the catalog for the compressed le storing Ra, and returns its location in the distributed le system. BackSymbol(a), instead, returns a relation Ra = f(y; x) j (x; a; y) 2 Eg. To this end, BackSymbol(a) is implemented as a Map-only MapReduce job, whose mappers just reads the pairs in Ra and return their inverse.
Wildcard. The Wildcard() operator implements the operator of GXPath. To this end, it exploits the fact the input graphs have been previously indexed, and just returns the locations in the distributed le system of the les that store relations Ra, for any a 2 , hence leveraging on Hadoop ability to work on multiple input les.
Concat. Concat(R 1 ; R 2 ) takes as input two binary
relations and returns their composition. This is obtained by
applying a matrix multiplication algorithm. In MapReduce
there are di erent ways to evaluate matrix multiplication
[
Union. Union(R 1 ; R 2 ) implements the union between two binary relations R 1 and R 2 . As MapReduce jobs can work on multiple input les and in our system duplicate elimination is performed at the end of the computation, Union(R 1 ; R 2 ) just returns to its father operator the location of the compressed les in the distributed le system containing the pairs of R 1 and R 2 , without the need to activate a MapReduce job.
Cond. Cond(R) takes as input a binary relation R and returns a new relation R0 such that (x; x) 2 R0 if and only if 9y:(x; y) 2 R. Cond(R) is implemented as a single MapReduce job, whose Mappers create a (x; x) pair for each (x; y) pair being read.
Compl. Compl(R) takes as input a binary relation R and returns its complement, i.e., V V R. This operator is particularly challenging as the topology of the graph cannot be used to guide the evaluation process. This operator is implemented by loading R in a persistent hash set, and by running a MapReduce job, where each Mapper creates a fragment of V V and, for each generated pair (x; y), discards (x; y) if (x; y) is contained in the hash set. The job has no Reduce phase.
Count. Count(R; m; n) takes as input a binary relation R,
and returns the result of Rm;n. To this end, our system
actually computes the bounded transitive and re exive
closure of R by exploiting a MapReduce implementation of a
variation of the Seminaive algorithm [
Distinct. The Distinct(R) operator takes as input a binary relation with duplicates R, and returns a binary relation R without duplicates. Duplicate elimination is implemented through a MapReduce job, whose Map phase, for each pair (x; y) 2 R, returns a pair (K; V ), where K = (x; y) and V is empty; in the Reduce phase, then, each reducer receives a pair ((x; y); L), where L is an empty list, and just outputs (x; y).
In this section we analyze the performance and the scalability of our query processor. As there is no standard benchmark for evaluating the performance of graph query processors, we generated a set of 15 random queries and evaluated them on three datasets comprising large synthetic power-law graphs. 4.1
We performed our experiments on a 4-node, multitenant Hadoop cluster which is part of the Cineca1 PICO cluster. Each node features an Intel Xeon E5 2650 v2 @ 2.6GHz CPU with 16 cores, 64 GB of main memory, and 32TB of local disks. Cluster nodes run RHEL Linux 6.5 and Hadoop 2.6.0, with HDFS block size set to 128MB and up to 2GB of memory per container. 4.2
We performed our experiments on three datasets
comprising synthetic graphs generated, according to the power
law, by using the R-MAT generator of GTGraph [
The R-MAT algorithm generates random graphs according to the power law distribution; the inputs of the algorithm are the desired number of vertices n and the requested number m of edges; we set the minimum and maximum label value to 0 and 1000, respectively. The graphs generated by R-MAT are summarized in Tables 1, 2, and 3.
As there is no standard benchmark for evaluating the
performance and the scalability of graph query processors, we
based our experiments on a set of random GXPath queries.
To create this set of queries, we implemented an extended
version of the random regular expression generator described
by Colazzo et al. in [
In our generator, operators are divided in two main classes: intermediate operators and terminal operators. The rst class comprises unary and binary operators, such as + , , [ ], m;n, and , while the second class only contains operators that may appear in a leaf of the parse tree of a query: a, a , , and .
The generation algorithm takes as input an expected query AST depth, as well as a distribution of probability for intermediate and terminal operators, and it works recursively. During each recursive call, it veri es whether the current AST has reached the expected depth: if so, the algorithm randomly selects a leaf operator; otherwise, it randomly chooses a unary or binary operator and performs another recursive call.
Operators are selected by using a random generator, that generates doubles between 0 (inclusive) and 1 (exclusive). These range values are mapped, according to the operator probability set in Table 4.(a), into GXPath operators. In a similar way, the algorithm chooses the leafs using the probability set in Table 4.(b). m and n indexes of counting operators are extracted in a random way, in the integer range between 0 and . Label values are randomly selected in the interval ]0; 1000[.
(a) Unary and binary operators.
Operator
+ [ ] m;n p 0,36 0,36 0,18 0,09 0,01 (b) Terminal operators.
Operator a a p 0,48 0,48 0,01 0,01
We generated, through the above random generator, 15 queries, shown in Table 5 These queries do not contain the complement operator. Indeed, the evaluation of this operator requires to materialize intermediate results whose size can be quadratic in the size of the input graph; this leaded, in all our preliminary experiments, to the exhaustion of the available HDFS space, even with compression enabled; therefore, we decided to remove queries with complement from our query set. 4.4
In our experimental evaluation we performed several kinds of experiments. As a preliminary test, we measured the indexing time for all datasets; then, we evaluated the performance and the scalability of the system by executing each query of Section 4.3 on each dataset.
1Cineca is an Italian university consortium for supercomputing services.
In these preliminary tests we measured the time required for preprocessing and indexing each graph in our datasets.
The results we obtained are shown in Figures 6(a), 6(b), and 6(c).
As it can be observed, the preprocessing and indexing time grows linearly with the size of the input graphs. 4.4.2
In these tests we evaluated the performance and the scalability of our system when processing the queries of Section 4.3 on the datasets of Section 4.2.
The results we obtained are shown in Figures 7(a), 7(b), and 7(c). As query Q6, which contains the wildcard operator, showed a signi cant overhead wrt the remaining queries, we also reported in Figures 8(a), 8(b), and 8(c) these results without query Q6. As it can be observed, the system preforms well on all queries (except for Q6), and scales linearly with the number of edges and/or vertices.
To better understand the behavior of the system, we can analyze in more detail the results obtained so far for queries Q5, Q6, Q9, and Q12. Indeed, queries Q5 and Q6 are the only ones that include, respectively, and ; query Q9 contains counting operators with relatively high values of m and n; query Q12, nally, contains all the remaining operators. Query Q5. Query Q5 includes an operator inside a union at the top level of the query. This implies that the system must materialize an intermediate relation J KG = f(u; u) j u 2 V g containing jV j pairs. In turn, this implies that the selectivity of Q5 is quite low and explains why its processing cost is higher than those of the other queries, as shown in Figs. 8(a), 8(b), and 8(c).
To understand if by itself can signi cantly worsen the processing time of a query, we compared in Figure 9 the query execution time with the time required for generating J KG. As it can be observed, on large graphs impact on the system performance can be quite signi cant.
Query Q6. From Figures 7(a), 7(b), and 7(c), we can observe that query Q6 is the most expensive one in our query set, as its execution time may exceed four hours on large graphs. Indeed, to process this query the system must read the whole input graph, even if it has been previously indexed. Indexing, however, helps in decreasing Q6 processing time, since the system can read the compressed les produced during the indexing phase.
epsilon 1E+08 2E+08 4E+08
5E+08 3E+08
Vertices Query Q9. This query includes a counting operator with relatively high values of m1 and n (59 and 63, respectively). This implies that our evaluation algorithm must perform several iterations; however, this does not a ect Q6 processing time so badly, which con rms that our system has good scalability properties.
Query Q12. This query contains all operators of the language with the exception of counting, , and . From Figures 8(a), 8(b), and 8(c) we can observe that the system behaves very well while processing this query, and scales linearly with the number of vertices and edges. This suggests that counting, , and (together with complement) are the most expensive operators to evaluate. 5.
There exist several systems for managing and querying data graphs. In this section we review a few of them. 5.1
Neo4j [
In niteGraph [
G-Path [
Q1 Q2 Q3 Q4 Q5 Q6 Q7 Q8 Q9 Q10 Q11 Q12 Q13 Q14 Q15 Q1 Q2 Q3 Q4 Q5 Q7 Q8 Q9 Q10 Q11 Q12 Q13 Q14
Q15 0
Q1 Q2 Q3 Q4 Q5 Q6 Q7 Q8 Q9 Q10 Q11 Q12 Q13 Q14 Q15 Q1 Q2 Q3 Q4 Q5 Q7 Q8 Q9 Q10 Q11 Q12 Q13 Q14 Q15 (a) G 1 results. (b) G 2 results. (c) G 3 results.
The system described by Koschmieder and Leser in [
In [
In this work we investigated the problem of querying massive graph datasets and, in particular, we implemented a scalable and e cient distributed query processor for the navigational fragment of GXPath. Our query processor exploits a MapReduce infrastructure to coordinate the work of multiple machines in a cluster and to distributed the computing load among them.
To evaluate the performance and the scalability of our implementation, we performed several experiments on a very small Hadoop cluster, using synthetic graph datasets with size up to 500 million vertices and 50 billion edges; given the lack of benchmarks for graph queries, we relied on a set of randomly generated queries. These experiments showed that our system is scalable and that it can e ciently process complex queries on large graphs.
In the near future we would like to extend our system in two ways. First of all, we would like to widen the fragment of GXPath being supported by our implementation, and, in particular, to cover the value test fragment of the language. This is a challenging task, as GXPath allows for comparing values stored in vertices reachable from di erent paths.
In the second place, while our system is very scalable, queries containing complement operations are still too expensive to be evaluated. To overcome this limitation, we plan to introduce more sophisticated processing techniques that, looking at the part of the query following a complement operator, could limit the scope of the complement.