OWL 2 EL ontologies are used to model and reason over data from diverse domains such as biomedicine, geography and road tra c. Data in these domains is increasing at a rate quicker than the increase in main memory and computation power of a single machine. Recent e orts in OWL reasoning algorithms lead to the decrease in classi cation time from several hours to a few seconds even for large ontologies like SNOMED CT. This is especially true for ontologies in the description logic EL+ (a fragment of the OWL 2 EL pro le). Reasoners such as Pellet, Hermit, ELK etc. make an assumption that the ontology would t in the main memory, which is unreasonable given projected increase in data volumes. Increase in the data volume also necessitates an increase in the computation power. This lead us to the use of a distributed system, so that memory and computation requirements can be spread across machines. We present a distributed system for the classi cation of EL+ ontologies along with some results on its scalability and performance.
The OWL 2 EL pro le [
cation. Similar to other distributed systems, the design decisions and the performance of our distributed system, DistEL3 involve answering the following questions e ectively.
2 EL+ extended with ABoxes can be handled with essentially the same algorithm. 3 The source code is available at https://github.com/raghavam/DistEL. Communication How do the distributed processes communicate and how can this be minimized? Data Duplication Is data duplication required? How many copies are maintained? Result Collection After all the processes terminate, will the results be spread across the cluster? Note that several other characteristics of a distributed system such as fault tolerance, transparency, etc. have been excluded since they are not yet supported by our system. In the following sections, we describe how our system solves the issues mentioned above, and show that our system can handle large ontologies.
The plan of the paper is as follows. In Section 2 we recall preliminaries concerning EL+. In Section 3 we describe our distributed approach. In Section 4 we present and discuss our experimental evaluation. In Section 5 we discuss limitations of our approach and future work. In Section 6 we discuss related work, and in Section 7 we conclude. 2
details, please refer to [
C ::= A j > j C u D j 9r:C;
where A ranges over concept names, r over role names, and C; D over (possibly
complex) concepts. An ontology in EL+ is a nite set of general concept inclusions
C v D and role inclusions r1 rn v r, where r; r1; : : : ; rn are role names,
n 2 Z+. For a general introduction to description logics, and for the formal
semantics of the constructors available in EL+, please refer to [
and Table 1 presents a slightly modi ed set of completion rules which is easily
checked to be sound and complete as well [
R3-1: If A 2 S(Y ) and 9r:A v B 2 O, then 9r:Y v B R3-2: If 9r:Y v B and (X; Y ) 2 R(r), then S(X) := S(X) [ fBg This helps in the division and distribution of work that needs to be done by our reasoner. The axioms in the ontology are in one of the normal forms given on the left column of Table 1. S(X) contains all the subsumers of X i.e., A 2 S(X) means X v A. Likewise, R(r) stands for f(X; Y )jX v 9r:Y g. Our Normal Form
A v B A1 u 9r:A v B 9r:A v B A v 9r:B
Completion Rule R1-1 If A 2 S(X), A v B 2 O, and B 62 S(X)
then S(X) := S(X) [ fBg u An v B R1-2 If A1; : : : ; An 2 S(X), A1 u u An v B 2 O, B 62 S(X) then S(X) := S(X) [ fBg R2 If A 2 S(X), A v 9r:B 2 O, and (X; B) 62 R(r)
then R(r) := R(r) [ f(X; B)g R3-1 If A 2 S(Y ), 9r:A v B 2 O
then P = P [ f9r:Y v Bg R3-2 If (X; Y ) 2 R(r), 9r:Y v B 2 P and B 62 S(X)
then S(X) := S(X) [ fBg r v s R4 If (X; Y ) 2 R(r), r v s 2 O, and (X; Y ) 62 R(s)
then R(s) := R(s) [ f(X; Y )g r s v t R5 If (X; Y ) 2 R(r), (Y; Z) 2 R(s), r s v t 2 O, (X; Z) 62 R(t) then R(t) := R(t) [ f(X; Z)g
Table 1. Axioms (in normal forms) and modi ed completion rules of CEL Algorithm 1 Pseudocode for CEL classi cation xpoint iteration S(X) fX; >g, for each concept X in the ontology.
R(r) fg, for each role r in the ontology.
P fg repeat
Used below, Old.S(X) stands for S(X) now, and Old.R(r) stands for R(r) and Old.P stands for P ;
S(X) apply R1-1 using S(X); S(X) apply R1-2 using S(X); R(r) apply R2 using S(X) and R(r); P apply R3-1 using S(X); S(X) apply R3-2 using R(r) and P ; R(r) apply R4 using R(r);
R(r) apply R5 using R(r); until ((Old.S(X) = S(X)) and (Old.R(r) = R(r)) and (Old.P = P )) goal is to compute S(X) for each concept X in the ontology O. The R(r) is used in deriving all such subclass relationships. P is a set which holds the axioms generated by rule R3-1. Instead of representing these axioms by P , the other option is to add these axioms back in the ontology O, but we would like to keep the ontology read only.
atively until no new output is generated, as shown in Algorithm 1. Notations We use U (X) to refer to the \inverse" of S(X) i.e., U (X) = fA j X w Ag. The advantage and performance bene t that is obtained by using U (X) instead of S(X) is explained later in Section 3. It is typical in computer science to think of U (X) as applying the de nition of a function U to the argument X and thus U (X) does not yield di erent results on repeated use. In constrast with this, we use U [X] to stand for the value stored in an associative array U indexed by a possibly non-integer value X. The same applies to R(r). Hence U [X] and R[r] are conceptually treated as associative arrays, but implementation details might vary. In the rest of the paper we use U [X] and R[r] instead of S(X) and R(r). Sometimes, we refer to all R[r] collectively as R-data. 3
Architecture Each group of nodes in the cluster (see Figure 1) is dedicated to work on only one particular completion rule Ri, i.e., on axioms belonging to Ri's normal form. Axioms of the ontology are split into disjoint collections based on their normal form. Each disjoint set of axioms are assigned to a particular group, Gi, responsible for processing rule Ri. Within each group, axioms are again split among the nodes of the group. Nodes of Group11, Group12, Group32 produce results which are collected by a single node. We use a set of key-value pairs to represent the axioms and the sets U [X], R[r] that are distributed over the cluster. Figure 1 also shows the dependency among the completion rules i.e., the axioms that are to be processed in the next iteration are determined by the updates done by other nodes. For example, the axioms that will be considered in rule R2 in the next iteration will depend on the output from rules R1-1, R1-2 and R3-2. This is explained further in the optimization section. Algorithm 2 Pseudocode for rule R1-1
K 0 for all axioms of the form A v B do
K K + (U [B] [= U [A]) //add U[A] to U[B] end for return K Algorithm 3 Pseudocode for rule R1-2
K 0 for all axioms of the form A1 u
K K + (U [B] [= U [A1] \ end for return K u An v B do
\ U [An]) Algorithm 4 RolePairHelper(fr, Xg, B)
K Send triplet (fr, Bg, X) to Group32 if there exists s such that r v s then
K Send (fr, Xg, B) to Group4 //send to D1 end if if there exist s and t such that r s v t then
K Send (fr, Bg, X) to Group5 //send to D0 end if if there exist s and t such that s r v t then
K Send (fs, Xg, B) to Group5 //send to D1 end if return K //send to D1 Key-Value Store Redis4 is an open source, high performance key-value store implementation. It provides several data structures like sets, sorted sets, hash, lists. It also supports atomic operations and server-side Lua5 scripting along with client-side sharding. All these features are used in our implementation. Redis runs on each node of the cluster holding the axioms, R[r] and U [X] values.
We use a Java client named Jedis6 to interact with Redis.
Pseudocode for completion rules Pseudocode for some rules use a notation such as D0, D1. They denote databases of Redis. Each Redis instance can have several databases associated with it. If not mentioned speci cally then all the data goes into database-0 (or D0). Pseudocode for all the rules captures the total number of updates made and returns this number.
The pseudocode of R1-1 is given in Algorithm 2. The operator [= performs the set union and returns the number of elements that were added to the destination set { the latter is needed for termination checking, discussed
Algorithm 5 Pseudocode for rule R2
K 0 for all axioms of the form A v 9r:B do for all X 2 U [A] do
K K + RolePairHelper(fr, Xg, B) end for end for return K Algorithm 6 Pseudocode for rule R3-1
K 0 for all r, A, B in axioms of the form 9r:A v B do for all Y 2 U [A] do
K K + (Send axiom 9r:Y v B to Group32) end for end for return K Algorithm 7 Pseudocode for rule R3-2
K 0 for all axioms of the form 9r:Y v B received from R3-1 do
Tr := fX j (fr; Y g; X) 2 D1g; //received from RolePairHelper
K K + (U [B] [= Tr) end for return K Algorithm 8 Pseudocode for rule R4
K 0 T3 := set of triplets received from RolePairHelper(); Tr := fr j (fr; Xg; Y ) 2 T3g; for all r 2 Tr do for all roles s such that r v s is an axiom do
K K + RolePairHelper(fs, Xg, Y) end for end for return K below. All U [X] are stored on the result node. So [= also implicitly involves contacting the result node for read (U [A]) and write (U [X]) operations.
The pseudocode of R1-2 is given in Algorithm 3. As mentioned below, it sufces to nd the intersection of all U [A] involved in the conjuncts, i.e., A1 : : : An.
Expanding on these two rules, let us brie y come back to an issue mentioned earlier, namely why we chose to use U [X] instead of S[X] for our implementation.
Let O be an ontology and let K u L u M v N 2 O. Furthermore, assume that there are ve concepts in the ontology, K; L; M; N and P . During some iteration of the classi cation assume S(K) = fK; L; N; >g, S(L) = fL; P; M; >g, Algorithm 9 Pseudocode for rule R5
K 0 T0 := set of triplets received from RolePairHelper(); for all (fr; Y g; X) 2 T0 do
Tr := fr j (fr; Y g; Z) 2 D1g; for all t 2 axioms r s v t do
K K + RolePairHelper(ft, Xg, Z) end for end for return K S(M ) = fM; N; K; >g, S(N ) = fN; >g; and S(P ) = fP; K; L; M; >g. Now, according to rule R1-2, we have to check for the presence of K; L and M in each of the ve S(X), where X = K; L; M; N; P . Since only S(P ) has K; L; M , we have to add N to S(P ).
On the other hand, use instead U [K] = fK; M; P g, U [L] = fL; K; P g, U [M ] = fM; L; P g, U [N ] = fN; K; M; P g, U [P ] = fP; Lg. In this case, instead of checking all U [X], we can compute the intersection of U [K], U [L], U [M ], which is P . So, P v N which means U [N ] [= fP g. In large ontologies, the number of concepts would be in the millions or more, but the number of conjuncts in axioms like A1 u u An v B would be very less in number. So the performance is better by using U [X] since set intersection needs to be performed only on a very small number of sets in this case.
Rules R2, R4 and R5 deal with R[r] values. RolePairHelper, with pseudocode given in Algorithm 4, provides functionality that is common to these three rules. The R-data, R[.], is an associative array indexed by roles. RolePairHelper(frole r, concept Xg, concept B) is invoked in rules R2, R4 and R5. RolePairHelper() informs all nodes (namely Group32, Group4 and Group5) of these updates. Group4 does not care to know the updates to R[r] unless role r has a super role s, that is for some s, r v s. Similarly, Group5 does not care to know of these updates unless there exist roles s and t such that r s v t. Note that the R[.] across all the nodes will, in general, not be the same because of these selective updates. Note also that replicating R[.] across all nodes causes no semantic harm. This is done to facilitate local reads of R-data on nodes dealing with role axioms, i.e., Group32, Group4 and Group5. The Send primitive in Algorithms 4 and 6, returns 0 if the message sent is duplicate, or 1 otherwise.
Nodes handling rule R2, i.e., Group2, do not make use of R[r] values. So they do not need to be stored locally on Group2 nodes as shown in Algorithm 5. Rules R2, R4 and R5 potentially add new entries to the R-data, and every such update implies a triggering of rules R3-2, R4, and R5. RolePairHelper(fr, Xg, B) broadcasts such updates.
The pseudocode for rule R3-1 is given in Algorithm 6. Here, newly formed axioms do not need to be sent to all the nodes in Group32 but can be sent to only a speci c node. For the sake of clarity, this is not shown in the pseudocode and is explained in the section on optimizations. Algorithm 10 Modeling of One Iteration, OIPi(D)
K apply rule Ri using (D) if K == 0 then
return true else
return false end if Algorithm 11 Modeling of a Process, Pi repeat isNew OIPi(D) broadcast isNew to all Pj receive tj from all Pj t t1 _ t2 _ ::: _ tj until :t
The pseudocode for rule R3-2 is given in Algorithm 7. Here, database-1 (D1) is queried with key fr; Y g and Tr holds all such X.
Regarding R4, in Algorithm 8, T3 receives only such triples whose r participates in an axiom of the form r v s. Tr is formed for each r found in T3.
Concerning R5, in Algorithm 9, using the key fr; Y g, the database D1 is queried and the results are referenced by Tr. All axioms of the form r s v t in which r participates in, are retrieved. Note, that the value of s does not matter, since it is already taken care of in RolePairHelper. The following example illustrates how Algorithms 4 and 9 are connected. Let k; m and n be roles, where k m v n, (X; Y ) 2 R(k); (Y; Z) 2 R(m). RolePairHelper(fk, Xg, Y) sends (fk, Yg, X) to D0 of Group5. RolePairHelper(fm, Yg, Z) sends (fk, Yg, Z) to D1 of Group5. In Algorithm 9, Tk contains (fk, Yg, X). D1 is queried with fk, Yg as the key and (fn, Xg, Z) is produced. n is obtained from the Rolechain axiom. Termination One iteration of the process, OIP, is modeled by the pseudocode shown in Algorithm 10. Each OIP reads and writes to a Redis instance D (database).
The appropriate pseudocode for rule Ri is processed and the return value is collected in K, which holds the number of updates made. Depending on the value of K, either true or false is returned.
Each process Pi performs the computations in Algorithm 11. The receive() is a blocking operation; i.e., until a message is received, the calling process (Pi) does not proceed to the next state. However, even though the pseudocode does not show it, we assume that the messages from Pj can be received in any order.
On each node, Ni, the process Pi processes all the axioms local to it and keeps track of whether this resulted in any changes (isNew ) to either its local Redis database or to the one on other nodes. This boolean value is broadcast to all the processes. When all the isNew messages are received, each process on all the nodes knows whether any of the other process made some updates or not. If at least one process makes an update then all the processes continue with the next iteration. If none of the processes makes an update then all processes terminate.
All the nodes in the cluster keep processing the axioms over several iterations until no new output is generated by any of the nodes. In sequential computation, this is fairly easy to check, but in a distributed system, all the nodes should be coordinated in order to check whether any of them have produced a new output.
The coordination among the processes on all nodes is achieved by message passing. Process Pi on each node is associated with a channel Ci. At the end of each iteration, Pi broadcasts its status message to channels on all the nodes. It then does a blocking wait until it receives messages from all the processes on its channel. This is generally known as barrier synchronization.7 If all the messages that Pi receives are false, i.e., none of the processes made an update to any of the key-value pairs, then Pi terminates.
Optimizations The following optimizations were put in place to speed up the processing of rules. 1. All the concepts and roles in the ontology are assigned numerical identi ers.
This saves space and is easier to process. 2. If X v A, normally this would be stored in a set whose key would be X and value would be A. But we reverse it, and make A the key and X its value. This makes the check A 2 S(X), a single read call. This check is required in rules R1-1, R1-2, R2 and R3-1. 3. As shown in Figure 1, the output of a rule can a ect the processing of another rule. For example, rule R2 works on axioms of the form A v 9r:B. R2 then depends on the rules which a ect A, which are R1-1, R1-2 and R3-2. If R1-1, R1-2 and R3-2 do not make any changes to U [A], then the axiom A v 9r:B need not be considered in the next iteration. We keep track of these dependencies and thereby reduce the number of axioms to work on in subsequent iterations. 4. Extending on the optimization just mentioned, if there is a change in U [A], then not all elements of U [A] need to be considered again. In fact, we need to consider only the newly added elements. This can be achieved by assigning scores to each element in the set U [A]. A node working on rule R2 and axiom A v 9r:B keeps track of the scores of elements in U [A], i.e., up to what it has read in the previous iteration, and only considers elements whose scores are greater than that. 5. In the pseudocode of Algorithm 6 and Algorithm 4, it is shown that the newly formed axiom and the triple is sent to all the nodes of a particular group. Instead, they can only be sent to a particular node in the group, and this node is selected based on the key. For the same key within the same group, however, we can ensure that always the same node gets selected. This reduces duplication of data. 7 http://en.wikipedia.org/wiki/Barrier_(computer_science) Ontology #Logical Axioms #Concepts #Roles Not-Galen 8,015 4,242 413 GO 28,897 20,465 1 NCI 46,870 27,653 70 SNOMED 1,038,481 433,106 62 SNOMED-DUP-2 2,076,962 866,212 124 SNOMED-DUP-3 3,115,443 1,299,318 186 SNOMED-GALEN-GO 1,075,393 456,319 476
Table 2. Sizes of (normalized) ontologies we used 4
To evaluate our implementation, we made use of the seven ontologies that are listed in Table 2. The numbers in Table 2 are obtained after normalizing the ontologies. The rst three ontologies have been obtained from http: //lat.inf.tu-dresden.de/~meng/toyont.html. SNOMED is from http:// www.ihtsdo.org/snomed-ct. SNOMED-DUP-2 and SNOMED-DUP-3 are ontologies with axioms from SNOMED, but each axiom replicated twice and thrice, respectively, while concept and role names are systematically renamed for each copy. SNOMED-GALEN-GO is a merge of the three ontologies, SNOMED, NotGalen and GO, which was obtained synthetically as follows: Upon normalization of each of these ontologies, new class names and role names were created which were assigned to a local namespace. However, class names and role names introduced in the normalization are shared between the ontologies. We thus obtain a merged ontology which, albeit the merge is synthetic, retains some of the real-life character of each of these ontologies.
DistEL is implemented in Java and makes use of Redis for storage. Our cluster consists of 13 Linux nodes, but our implementation scales to larger clusters. Each node has two quad-core AMD Opteron 2300MHz processors with 16GB RAM. DistEL treats a Redis instance as a node, so in order to show the scalability aspect of our implementation, we ran 2-3 Redis instances on a single node. This allowed us to e ectively run tests for more than 13 nodes on the cluster. Since each node has 8 cores, and data on an instance of Redis generally doesn't go beyond 5GB (for our experiments), running 2-3 Redis instances does not adversely a ect the evaluation.
Table 3 has the classi cation time of ontologies when run on Pellet (version 2.3.0), jCEL (0.18.2) and ELK (version 0.3.2). Heap space given to run all the ontologies is 12GB. Timeout limit given was 2 hours. All the reasoners are invoked through the OWL API. Time taken by the OWL API to load the ontology is not taken into consideration. Note that SNOMED-GALEN-GO could not be processed by any of the state-of-the-art systems. This is remarkable because the Ontology Pellet jCEL ELK Not-Galen 12.0 3.0 1.0 GO 5.0 5.0 2.0 NCI 6.0 7.0 3.0 SNOMED 1,845.0 327.0 24.0 SNOMED-DUP-2 OutOfMemory 687.0 64.0 SNOMED-DUP-3 OutOfMemory 1149.0 93.0
SNOMED-GALEN-GO OutOfMemory TIME OUT TIME OUT Table 3. Classi cation time of ontologies using Pellet, jCEL and ELK 8 We are not entirely certain yet what causes this explosion. Our guess is that it is caused by the large number of role chains in Galen, together with the sheer size of SNOMED. 9 http://owlapi.sourceforge.net the e ect of parallelization is very good indeed. E.g., using twice the number of nodes almost halves the runtime in most cases { so the e ect of the parallelization is indeed near optimal.
As expected, for the smaller ontologies, the parallelization does not have much e ect as soon as a certain threshold is reached. We see, however, that even when using many more nodes than necessary to reach optimal runtime, we do not get a signi cant amount of additional time lost due to communication overhead.
After having retrieved the runtime gures just discussed, we noticed that some of the measured times for 15 nodes were in fact higher than those of 12 nodes, and a similar e ect showed for 28 versus 25 nodes. When additional nodes are added to the cluster, they should ideally be assigned to the slowest processing nodes. But if this is not the case, then we do not see any noticeable improvement in performance. On the contrary, there is a possibility of reduction in performance because of additional communication overhead. Since we are currently assigning nodes by intuition and rule-of-thumb, we have to expect to get such performance drops sometimes.
To further check on this e ect, we repeated the run of SNOMED on 15 nodes using a di erent assignment of nodes to rules, and it resulted in a classi cation time of 606.05 seconds, which is a signi cant improvement compared to the timing obtained by the node assignment in Table 5. In fact, it is better than using 18 nodes in the previous assignment. This shows that our manual rule-ofthumb assignments are likely not optimal, and that, in fact, a signifcantly better performance should be achievable by automating the node assignment, or by using methods for dynamic load balancing. However, in this paper we only want to show that signi cant parallelization can be achieved, and do not yet focus on a most e cient implementation. This is left for future work.
Correctness of the results produced by DistEL is veri ed by comparing the output with that of ELK, in the cases where ELK does not time out.
There is a signi cant di erence in performance between our distributed implementation and ELK. One of the primary reasons for this di erence is that our implementation requires cross-node communication (key-value pairs are sent across the nodes) and among the nodes. On the other hand, our architecture can be extended by adding more nodes, and thus can scale up to datasets which ELK cannot handle, such as SNOMED-GALEN-GO.
We list some further insights which we gained from our implementation and system. 1. For our manual assignment of nodes to rules, it is very important to gure out the slowest processing nodes in the cluster, so that, if additional nodes are available, it would be easy to determine, to which group these new nodes should be assigned to. 2. The majority of the time is in fact spent on reading and writing to local as well as remote databases. Design choices and architecture should be formulated in such a way so as to reduce cross-node communication. 3. We cannot estimate the runtime or node assignment to rules just by counting the number of axioms. Some axioms are harder to process than others although their number might be less. A case in point is SNOMED-GALEN-GO compared with SNOMED-DUP-3. 5
Some of the limitations and planned next steps are presented below.
1. Compared to other popular distributed frameworks like Hadoop,10 our
architecture does not currently provide support for fault tolerance.
2. Axioms are distributed across the cluster by type rather than load. This
leads to improper load balancing. To improve load balancing we plan to
implement work stealing so that the idle nodes can work on axioms from
other nodes.
3. Completion rules are assigned to nodes manually, for now. This could be
done automatically by considering ontology statistics such as the number of
role axioms and subclass axioms, or other measures.
4. The OWL API is used to read the axioms from an ontology, and by design
it loads the entire ontology into memory. With the size of ontologies that we
hope to deal with using our work, this becomes a bottleneck. Going forward,
we plan to use a streaming API for XML11 to read the axioms.
5. We plan to extend our work to include ABox reasoning [
Most of the distributed reasoning approaches in the literature are focussed on
RDFS inference but there are a few that deal with a fragment of OWL, namely
OWL Horst. In [
Distributed resolution techniques were used by Stuckenschmidt et al., to
achieve scalability of various OWL fragments such as ALC [
some of which have been unsuccessful. Here, we presented an approach which
gave encouraging results due to increase in the number of axioms that are
processed locally and decrease in the cross-node communication, when compared to
our previous approaches [
With ever increasing data generation rates, large ontologies challenge reasoners from the perspective of memory and computation power. In such scenarios, distributed reasoners o er a viable solution. We presented our distributed approach to EL+ ontology classi cation, called DistEL, where we show that our classi er can handle large ontologies and the classi cation time decreases with the increase in nodes. The results are encouraging, and we plan to go ahead with adding ABox reasoning support to our work as well as explore other possible distributed classi cation approaches.
Acknowledgements. We would like to thank members of the Redis mailing list redis-db@googlegroups.com for their helpful suggestions. This work was supported by the National Science Foundation under award 1017225 "III: Small: TROn { Tractable Reasoning with Ontologies." Any opinions, ndings, and conclusions or recommendations expressed in this material are those of the author(s) and do not necessarily re ect the views of the National Science Foundation.