Recent years have seen theoretical and practical efforts on temporalizing and streamifying ontology-based data access (OBDA). This paper contributes to the practical efforts with a description/evaluation of a prototype implementation for the stream-temporal query language framework STARQL. STARQL serves the needs for industrially motivated scenarios, providing the same interface for querying historical data (reactive diagnostics) and for querying streamed data (continuous monitoring, predictive analytics). We show how to transform STARQL queries w.r.t. mappings into standard SQL queries, the difference between historical and continuous querying relying only in the use of a static window table vs. an incrementally updated window table. Experiments with a STARQL prototype engine using the PostgreSQL DBMS show the implementability and feasibility of our approach.
This paper contributes results to recent efforts for adapting the paradigm of
ontology-based data access (OBDA) to scenarios with streaming data [
Considering streams, the main challenge is that data cannot be processed as
a whole. The simple but fundamental idea is to apply a (small) sliding window
which is updated as new elements from the stream arrive at the query-answering
system (see, e.g., [
Considering temporal reasoning for reactive diagnostics as needed for the
Optique use case provided by Siemens [
The ABox sequencing strategy for windows required to avoid inconsistencies, as argued above, makes rewriting and, more importantly, unfolding of STARQL queries a challenging task. In particular, one may ask whether it is possible to rewrite and unfold one STARQL query into a single backend query formulated in the query language provided by the backend systems. We show that (Postgre)SQL transformations are indeed possible and describe them in the paper for the special case of one stream with slide parameter identical to pulse parameter and a specific sequencing strategy (for the details see the following section). 2
We recap the syntax and the semantics of STARQL with a simple example. For a
complete formal treatment we refer the reader to [
For illustration purposes, our running example is a measurement scenario in which there is a (possibly virtual) stream SMsmt of ABox assertions. A stream of ABox assertions is an infinite set of timestamped ABox assertions of the form axhti. The timestamps stem from a flow of time (T; ) where T may even be a dense set and where is a linear order. The initial part of SMsmt , called SM5smst here, contains timestamped ABox assertions giving the value of a temperature sensor s0 at 6 time points starting with 0s.
SM5smst = fval (s0; 90 )h0si; val (s0; 93 )h1si; val (s0; 94 )h2si
val (s0; 92 )h3si; val (s0; 93 )h4si; val (s0; 95 )h5sig
Assume further, that a static ABox contains knowledge on sensors telling, e.g., which sensor is of which type. In particular, let BurnerT ipT empSens(s0) be in the static ABox. Moreover, let there be a pure DL-Lite TBox with additional information such as BurnerT ipT empSens v T empSens saying that all burner tip temperature sensors are temperature sensors.
The Siemens engineer has the following information need: Starting with time point 00:00 on 1.1.2005, tell me every second those temperature sensors whose value grew monotonically in the last 2 seconds. A possible STARQL representation of the information is illustrated in Listing 1. 1 PREFIX : <http :// www . optique - project . eu / siemens > 2 CREATE STREAM S_out AS 3 CONSTRUCT GRAPH NOW { ?s rdf : type MonInc } 4 FROM STREAM S_Msmt [NOW -2s , NOW ] - >"1S "^^ xsd : duration 5 WITH START = "2005 -01 -01 T01 :00:00 CET "^^ xsd : dateTime , 6 END = "2005 -01 -01 T02 :00:00 CET "^^ xsd : dateTime 7 STATIC ABOX <http :// www . optique - project . eu / siemens / ABoxstatic >, 8 TBOX <http :// www . optique - project . eu / siemens / TBox > 9 USING PULSE WITH 10 START = "2005 -01 -01 T00 :00:00 CET "^^ xsd : dateTime , 11 FREQUENCY = "1 S "^^ xsd : duration 12 WHERE { ?s rdf : type : TempSens } 13 SEQUENCE BY StdSeq AS seq 14 HAVING FORALL ?i < ?j IN seq ,?x ,? y: 15 IF ( GRAPH ?i { ?s : val ?x } AND GRAPH ?j { ?s : val ?y }) THEN ?x <= ?y
Listing 1: Example query in STARQL
After the create expressions for the stream and the output frequency the queries’ main contents are captured by the CONSTRUCT expressions. The construct expression describes the output format of the stream, using the named-graph notation of SPARQL for fixing a basic graph pattern (BGP) and attaching a time expression, here NOW, for the evolving time. The actual result in the example (in DL notation) is a stream of ABox assertions of the form M onInc(s0)hti. Sou5ts = fM onInc(s0)h0si; M onInc(s0)h1si; M onInc(s0)h2si; M onInc(s0)h5sig
The WHERE clause binds variables w.r.t. the non-streaming sources (ABox,
TBox) mentioned in the FROM clause by using (unions) of BGPs. We assume an
underlying DL-Lite logic for the static ABox, the TBox and the BGP (considered
as unions of conjunctive queries UCQs) which allows for concrete domain values,
e.g., DL-LiteA [
The heart of the STARQL queries is the window operator in combination with sequencing. The operator [NOW-2s, NOW]->"1S"^^xsd:duration used in Listing 1 describes a sliding window, which collects the timestamped ABox assertions in the last two seconds and slides 1s forward in time. Note that the START and END specifications over the stream: These make sense only for temporal queries over streamed historical data (see Sect. 3.1).
Every temporal ABox produced by the window operator is converted to a sequence of (pure) ABoxes. The sequence strategy determines how the timestamped assertions are sequenced into ABoxes. The sequencing method used in the example is standard sequencing (StdSeq): assertions with the same timestamp come into the same Abox. So, in the example the resulting sequence of ABoxes at time point 5s is trivial as there are no more than two ABox assertions with the same timestamp: fval (s0; 92 )gh0i; fval (s0; 93 )gh1i; fval (s0; 95 )gh2ig.
Now, at every time point, one has a sequence of ABoxes on which temporal (state-based) reasoning can be applied. This is realized in STARQL by a sorted first-order logic template in which state stamped UCQs conditions are embedded. We use here again the GRAPH notation from SPARQL. In our example the HAVING clause expresses a monotonicity condition stating that for all values ?x that are values of sensor ?s w.r.t the ith ABox (subgraph) and for all values ?y that are values of the same sensor ?s w.r.t. the jth ABox (subgraph), it must be the case that ?x is less than or equal to ?y.
The grammar for the HAVING clause (not shown here) exploits a safety
mechanism. Without it a HAVING clause such as ?y > 3, with free concrete domain
variable ?y over the reals, would be allowed: the set of bindings for ?y would
be infinite. Even more, the safety conditions guarantees that the evaluation of
the HAVING clause on the ABox sequence depends only on the active domain
not the whole domain, i.e., HAVING clauses are domain independent (d.i.) (see
[
As the HAVING clause language is d.i. (see [
As (backend) data source candidates we consider any DBMS providing a
declarative language such as SQL. This is not a limitation in comparison with
those approaches (in particular our own [
Because our transformation does not rely on a window operator on the
backend side but constructs the window contents within a window table, two different
implementations become possible: 1. Preprocesses the data in the backend in
order to materialize the window table for the whole time interval fixed within
the STARQL query. The abstract computation model for this implementation is
combined rewriting: The given query is rewritten w.r.t. the TBox and the
rewritten query is posed to a pre-processed ABox resulting from the original ABox by
(partially) materializing TBox axioms (see [
Datalog& Transforma=on&
Rules&
STARQL& Grammar& Parser&
Query&& (Data& Structure)&
Normaliza =on&Rules& Normalizer&
Query& (Normal&
Form)& Datalog& Program&
Op=mized& Datalog&
Program&
SQL& Transforma=on&
Rules& Op=mizer& Op=miza =on&Rules&
SQL&Query& and&Views& SQL&Transformer&
Mapping Rules& scenarios with multiple query processing where (many of) the queries use the same windows on the same streams.
As mentioned before, the aim of this paper is to give a proof-of-concept implementation for a stream-temporal engine. In particular, regarding the required OBDA transformations from the ontology layer to the backend sources, the engine is applicable to the following special case: There is only one stream, the slide is identical to the pulse, and the sequencing strategy is standard sequencing. 3.1
For reactive diagnosis as investigated in Optique with the Siemens power plant use case, specific patterns, aka events, are to be found in previously recorded data. Data represents 1.) measurement values of sensors and 2.) turbines with assembly groups, mounted sensors and their properties. Reactive diagnosis deals with analyzing data in order to understand, e.g., reasons for engine shutdown.
If we consider again the information need discussed in Section 2, it becomes clear that in the Siemens use case, patterns to be detected are defined w.r.t. certain time windows in which relevant events take place (e.g., monotonically increasing in a window of 10 minutes, say). Thus, in this context, temporal queries are formulated using time windows in the same spirit as time windows are used for continuous queries in stream processing systems.
The same queries can be registered as a continuous query or a historical query. For the latter, it should obviously be possible to specify a period of interest, i.e., a starting point and an end point for finding answers.
We consider the example query of Listing 1 for demonstrating the transformation of historical queries. Figure 1 shows how historical STARQL queries are processed internally. The prototype is implemented in Prolog, and the rule sets used as inputs in Fig. 1 are Prolog (DCG) rules. A query is parsed in order to produce parse tree data structures, which then are normalized. Normalization converts FORALL expressions in the HAVING clause into NOT EXISTS NOT expressions by pushing negation inside. The following listing shows the normalized HAVING part of our example query. 1 HAVING NOT EXISTS ?i < ?j IN seq ?x , ?y : 2 ( GRAPH ?i { ?s val ?x } AND GRAPH ?j { ?s val ?y } AND ?x > ?y ) ; The normalized query is translated into a datalog program (see Figure 1) with fresh predicate names being generated automatically. The WHERE expression is rewritten and unfolded due to the axioms in the TBox given in the query and the mapping rules, respectively. Here we assume that sensors are created by an SQL view defined below. In the body of the first rule, datalog code for the WHERE clause is inserted (rule q0, see Listing 2)). For every language element found in the parse tree (e.g., NOT EXISTS...), we generate a new datalog rule.
For every {?x rdf:type C} ({?x R ?y})) mentioned in the WHERE or HAVING clause we insert C(WID, X) (R(WID, X, Y)) in the datalog program. WID is an implicit parameter representing a so-called window ID. Correspondingly, for every GRAPH i {?x rdf:type C} (GRAPH i {?x R ?y}) mentioned in the HAVING clause we insert C(WID, X, i) (R(WID, X, Y, i)) with i representing the specific ABox in each window sequence.
The datalog clauses q1 to q4 in Listing 2 are generated automatically from the HAVING clause of the query above. 1 q0 (WID , S) :- sensor (WID , S). 2 q1 (WID , S) :- q0 (WID , S) , not q2 (WID , S). 3 q2 (WID , S) :- seq (WID , I) , seq (WID , J) , q3 (WID , I , J , S) , I < J. 4 q3 (WID , I , J , S) :- q4 (WID , I , J , X , Y , S). 5 q4 (WID , I , J , X , Y , S) :- val (WID , I , X , S) , val (WID , J , Y , S) , X > Y.
Listing 2: Generated datalog rules
The mapping specifications of the stream s_msmt, the relations val and sensor are predefined using datalog clauses and added to the datalog rules from a file. The additional mapping rules are shown in Listing 3. 1 val (WID , Index , Sensor , Value ) :2 window (WID , Index , Timestamp ) , 3 measurement ( Timestamp , Sensor , Value ). 4 sensor (WID , Sensor ) :- val (WID , _Index , Sensor , _Value ). 5 seq (WID , Index ) :- window (WID , Index , _Timestamp ).
Listing 3: Mapping specifications using datalog clauses
During clause generation, SQL transformation rules are generated (see Figure 1). Transformation rules represent the name of the SQL relation, the number of arguments as well as the type of each parameter. SQL transformation are statically specified also for the relations val, window, and seq.
For the CREATE STREAM and CONSTRUCT expression in the query, a clause 1 s_out (T , S) :- q1 (WID , S) , window_range (WID , T).
is added to the datalog program.
The datalog program generated by the module Datalog Transformer (see Figure 1) is then optimized. Optimization eliminates wrapper clauses. An according optimization of Listing 2 is shown below. 1 q1 (WID , S) :- sensor (WID , S) , not q2 (WID , S). 2 q3 (WID , I , J , S) :- seq (WID , I) , seq (WID , J) , val (WID , I , X , S) , 3 val (WID , J , Y , S) , X > Y , I < J.
Here, the datalog rules q0 and q1 have been reformulated to q1 and q2 to q4 from Listing 2 have been reformulated by eliminating wrapper clauses to q3.
Moreover, body atoms are removed if bindings are already generated by other
atoms. In our example we see that a seq clause is already a subclause of the val
clause (Listing 3). Using semantic query optimization [
Listing 4: Optimized datalog rules The datalog program is non-recursive and safe. So it can be translated to SQL as shown in Listing 5. The column names for relations are given by the SQL Transformation Rules generated by the Datalog Transformer and by mapping rules given as input to the processing pipeline (see Figure 1).
The translation to SQL relies on tables window and window_range (Listing 6). These are based on the stream specification(s) in the query. Given start ($start$) and end ($end$) points for accessing temporal data as well as window size ($window_size$) and window slide ($slide$), a representation for all possible windows (with specific time points for the window range) together with all indexes for states that are built for the window by the specified sequencing method specified are computed. For standard sequencing, window and window_range are generated using PostgreSQL functions such as generate_series. 3.2
The transformation above applies to temporal queries on historical data stored
in a RDBMS. So, the window table generation as part of the transformation
above is a one-step generation. In order to cope with streaming data, the
transformation process has slightly to be adapted. The window table now is assumed
to be incrementally updated by some function. Apart from that, the same
transformation as for temporal queries can be applied to realize continuous querying
with STARQL. In fact, the second implementation of the transformation that
we tested does not materialize the whole window table, and so it can be directly
adapted to DBs with dynamically updated entries. Similar ideas for continuos
processing are used for TelegraphCQ [
CREATE VIEW q1 AS SELECT rel1 .WID , rel1 . SID AS S FROM sensor rel1 WHERE NOT EXISTS ( SELECT *
FROM q3 rel2 WHERE rel2 . WID = rel1 . WID AND rel2 .S = rel1 . SID );
Listing 5: SQL transformation result 1 CREATE TABLE window_range AS 2 SELECT row_number () OVER ( ORDER BY x. timestamp ) - 1 AS wid , 3 x. timestamp as left 4 x. timestamp + $window_size$ as right 5 FROM ( SELECT generate_series ( $start$ , $end$ , $slide$ ) AS timestamp ) x ; 6 7 CREATE VIEW wid AS 8 SELECT DISTINCT r.wid , mp . timestamp 9 FROM measurement mp , window_range r 10 WHERE mp . timestamp BETWEEN r. left AND r. right ; 11 12 CREATE TABLE window AS 13 SELECT wid , rank () OVER ( PARTITION BY wid ORDER BY timestamp ASC ) as ind , timestamp 14 FROM wid ;
Listing 6: Window (range) tables 4
The system is evaluated along two example STARQL queries for reactive diagnosis in the Siemens use case. They representatives for queries expected to demand processing times that are quadratic (Query1) and linear (Query2). The engine transforms the queries w.r.t. predefined mappings into PostgreSQL queries.
The datasets that we use and describe in the following are part of the Siemens
use case [
Values per Day/Sensor Ds1 82080 3 Days 19 1440 5 MB Ds2 210,000 1506 Days 3 46.5 10 MB Ds3 515,845,000 1824 Days 204 1386 23,000 MB dataset and two larger anonymized datasets for use inside Optique. The public dataset (approximately 100 MB) has a simplified structure. For the evaluation we used three datasets: Ds1, Ds2 and Ds3. Ds1 contains public data and has a size of approximately 5MB. Ds2 contains anonymous data and has a size of approximately 10 MB, and Ds3 contains anonymous data with size 23 GB.
The (simplified) schema of the normalized tables is as follows: CREATE TABLE measurement (timestamp,sensor,value); CREATE TABLE sensor (id,assemblypart,name)
Measurements are represented with a table measurement and consist of a timestamp, a reference to the sensor, and a value. For our evaluation we are using one dataset of about 82,000 entries with a timestamp ranging over 3 days (Ds1), another dataset with about 210,000 entries with a timestamp ranging over 5 years (Ds2), and a dataset (Ds3) with more than 500 million entries over 5 years. All datasets contain data referring to a number of sensors.
The datasets differ in the total number of recorded values and also in the number of measured values per timeframe. In Ds1 and Ds3 a value is measured every minute. In Ds2 a value is measured only every 30 minute in average. So we expect the calculation of a single time window for Ds1 and Ds3 to be more difficult compared to the calculation for Ds2.
Query1 (Listing 7) builds, within each window, a sequence of all sensor values in the last 24 hours and checks whether one sensor increased monotonically. This query is expected to run quadratically slower as window size increases, due to the comparisons of all value pairs (?x, ?y) for all pairs of states (i,j).
Query2 (Listing 8) outputs, every minute, sensors that show a value higher than 90. This query is expected to be faster because of its simple window content with at most one timestamp. Both queries can be transformed to PostgreSQL. We show only the transformation for Query1 in Listing 9.
The tests were run in a VM on a system with an i7 2.8 GHz CPU and 16 GB of ram. Mean values of several test runs with cold cache are shown in the following tables. For our tests we used two approaches corresponding to combined rewriting and classical rewriting, respectively (cf. Sect. 3), and for each of these Query1 and Query2. In the first approach we pre-calculated all time windows in one large table, where every window has a window id, evaluated both queries once and a second time with additional window index structures added to the table. 1 CREATE STREAM S_out1 AS 2 SELECT { ? sensor rdf : type : RecentMonInc }<NOW > 3 FROM burner_regulator [ NOW - 24 hours , NOW ]-> 24 hours 4 SEQUENCE BY StdSeq AS seq 5 HAVING FORALL i , j IN seq , ?x ,? y 6 IF {? sensor : hasVal ?x}<i > AND { : Regulator : hasVal ?y }<j > AND i < j 7 THEN ?x <= ?y ELSE TRUE
Listing 7: STARQL query Query1 1 CREATE STREAM S_out2 AS 2 SELECT { ? sens rdf : type : tooHigh }<NOW > 3 FROM burner_3 [ NOW , NOW ]-> 1 minute 4 SEQUENCE BY StdSeq AS seq 5 HAVING FORALL i IN seq , ?x IF { ? sens : hasVal ?x }<i > THEN ?x > 90
Listing 8: STARQL query Query2 1 CREATE OR REPLACE VIEW window_range AS 2 SELECT row_number () OVER ( ORDER BY x. timestamp ) - 1 AS wid , 3 x. timestamp as left , x. timestamp + '1 hour ':: interval as right 4 FROM ( SELECT generate_series ( MIN ( mp . timestamp ) , 5 MAX ( mp . timestamp ) , '1 hour ':: interval ) AS timestamp FROM measurement mp ) x; 6 7 CREATE OR REPLACE VIEW wid AS 8 SELECT distinct wid , timestamp 9 FROM measurement mp , window_range w 10 WHERE mp . timestamp >= w. left and mp . timestamp < w. right ;
CREATE VIEW win AS SELECT wid , rank () OVER ( PARTITION BY wid ORDER BY timestamp ASC ) as ind , timestamp FROM wid ;
Listing 9: Query1 in PostgreSQL
Results for the first approach are shown in Table 2. For every query you see a column with times in seconds for the non-indexed and indexed table. The nonindexed times consist of generating the table and evaluating the query. In the indexed version we added a B-tree index. The precalculation column shows the additional time for generating the window table, which has to be added in every case to the next two evaluation columns for non-indexed and indexed data.
There are different trade offs for both queries. For Query1 the total evaluation time increases as the window size increases, the precalculation time stays the same. The evaluation time for Ds1 increases faster as it has more measured values per window, compared to Ds2, which only has about one measured value per 30 minutes. Comparing the time for indexed and non-indexed data, one sees that with progressing time more single windows are used per query. On Ds1 we use only three windows and can decrease time from 30 to 26 seconds. On Ds2 we have about 1800 windows and can decrease the time from 17 to 8 seconds, which is more than 50 percent. Query2 produces a lot of small windows, so the evaluation time for each window is very short. The precalculation time for Ds2 increases a lot as we have about 2 million potential single windows in the window table. With a timeframe of only three days, the precalculation time for Ds1 stays small. The index has nearly no influence for Query2 as each window has only one tuple entry and all windows are evaluated once.
Ds3 could not be evaluated with the precalculation step, as it requires more than 50 GB additional memory for the window table. Therefore, we implemented a second approach by additional pl/pgsql functions. The idea was to generate each window dynamically, evaluate it, and delete the memory afterwards.
Results for the second approach are shown in Table 3. As each window is generated dynamically, there is no precalculation. On the other hand, no indexes can be added to a materialized table. The main disadvantage is that windows without values can not be filtered out in advance. As there are potentially 2 million windows for Query2 on Ds2, the system tries also to generate the empty windows, which increases the time a lot. Nevertheless, the problem of additional required space is solved and also the complete 20GB of Ds3 can be evaluated. 5
Much of the relevant work on stream processing has been done in the context of
DSMS [
First steps towards streamified OBDA are stream extensions of SPARQL
with a window operator having a multi-bag semantics where timestamps are
dropped [
The temporal OBDA approach of [
Though not directly related to OBDA, other relevant work stems from the
field of complex event processing. For example, EP-SPARQL/ETALIS [
The main objective in designing a query language that is intended to be used in industry is to find the right balance between expressibility and feasibility. OBDA goes for weak expressibility and high feasibility by choosing rewriting and unfolding as methodology for query answering. But even in OBDA, feasibility is not a feature one gets for free; rather it has to be achieved with optimizations. So, for engines that are implemented according to the OBDA paradigm one has to show that such transformations are theoretically possible and feasible.
In this paper we argued that STARQL provides an adequate solution for streamified and temporalized OBDA scenarios as that of Siemens because: 1. It offers a semantics that allows a unified treatment of querying temporal and streaming data. 2. It combines high expressivity with safeness to guarantee a smooth transformation into standard domain independent backend queries such as SQL. 3. It can be implemented in an engine that implements the transformations s.t. acceptable query execution times are achievable, if run with the optimization mentioned here.
Future work contains, amongst others, the following: 1. extensive (scalability) tests with known benchmarks for stream processing, 2. generalization of the transformation to multiple streams where the slides parameters are not equal to the pulse parameter and where non-standard-sequencing strategies are used, and 3. extensive comparison with other approaches, in particular CEP approaches.