The increasing popularity of RDF Stream Processing (RSP) has led to developments of data models and processing engines which diverge in several aspects, ranging from the representation of RDF streams to semantics. Benchmarking systems such as LSBench, SRBench, and CSRBench were introduced as attempts to compare different approaches. However, these works mainly concentrate on the operational aspects. The recent logic-based LARS framework provides a theoretical underpinning to analyze stream processing/reasoning semantics. Towards comparing RSP engines at the semantic level, in this paper, we pick two representative RSP engines, namely C-SPARQL and CQELS, and propose translations from their languages and execution modes into LARS. We show the faithfulness of the translations and discuss how they can be exploited to provide a formal analysis and comparison of RSP semantics.
Within the Semantic Web research area, RDF Stream Processing (RSP) recently emerged
to address challenges in querying heterogeneous data streams. This has led to
developments of data models and processing engines, which diverge in several aspects, ranging
from the representation of RDF streams, execution modes, to semantics [
A standardization must start from seeing the differences between existing approaches
and thus comparing RSP engines is an important topic. Initial empirical comparisons
were carried out in SRBench [
Due to the lack of a common language to express divergent RSP approaches, the three works above could just look at the output of the engines and did not have further means to explain beyond the output what caused the difference semantically.
Recently, [
Recently, a Logic-based framework for Analyzing Reasoning over Stream (LARS)
was introduced [
We show the faithfulness of the translations and discuss how they can be exploited to provide a formal analysis and comparison of RSP semantics. 2
This section briefly reviews RDF, SPARQL, RSP, and LARS, which will be illustrated
using the following running scenario inspired by [
Example 1 The Sirius Cybernetics Corporation offers shop owners a real-time geomarketing solution (RT GM ) to increase their sales. RT GM provides two services: (i) an application that allows shop owners to push instantaneous discount coupons to a server, and (ii) a free mobile app that fetches the coupons from shops near the phone, matches them with the preferences specified in the user’s shopping profile, and delivers the matched coupons to the user. Alice and Bob own shops a and b that sell shoes and glasses, resp. At time point 10, Alice sends out a coupon for a 30% discount for men’s MBT shoes. At time 15, Bob sends out a coupon for a 25% discount on Ray-Ban glasses.
Claire has the App installed on her mobile phone and is walking near shops a and b from time 18. She is neither interested in discounts on men’s products nor discount of less than 20%. Therefore, she will get only the discount from shop b. 2.1
RDF is a W3C recommendation for data interchange on the Web [
subject, p the predicate, and o the object. An RDF graph is a set of RDF triples. Example 2 (cont’d) Information in the scenario of Ex. 1 about products, offers from shops, and Claire’s relative locations to shops can be stored in the following RDF graphs: G = f : \mbt" : g classify : 1: : \rayban" : g classify : 0: : : : g g1 = f: a : offers : c1: : c1 : on : \mbt": : c1 : reduce : 30:g g2 = f: b : offers : c2: : c2 : on : \rayban": : c2 : reduce : 25:g g3 = f: \claire" : isNear : a: : \claire" : isNear : b:g A triple pattern is a tuple (sp; pp; op)2(I [ B [ V ) (I [ V ) (I [ B [ L [ V ), where V is a set of variables. A basic graph pattern is a set of triple patterns.
SPARQL [
extend Linked Stream Data and Linked Data with temporal aspects: 1. An RDF graph at timestamp t, denoted by G(t), is a set of RDF triples valid at time t and called an instantaneous RDF graph. A temporal RDF graph is a sequence G = [G(t)]; t 2 N = f0; 1; 2; : : :g, ordered by t.
t is a timestamp.
ments hgi : [ti]i, where gi, representing an offer, is of the form in Ex. 2 and ti can be either (i) the time point when the offer is announced by a shop owner (application time), or (ii) the time point when gi arrives at an RSP engine (system time).
Continuous Queries. Continuous queries are registered on a set of input streams and background data, and continuously send out the answers as new input arrives at the streams. There are two modes to execute such queries. In pull-based mode, the system is scheduled to execute periodically independent of the arrival of data and its incoming rate; in push-based mode, the execution is triggered as soon as data is fed into the system.
Continuous queries in C-SPARQL and CQELS follow the approach by the
Continuous Query Language (CQL) [
As CQL is based on SQL, the background data tables and input streams all have
schemas. This makes it crystal clear to see which input tuple comes from which stream.
On the other hand, as RDF is schema-less, it is not straightforward to get this
distinction. RSP engines use different approaches to build the snapshot datasets for R2R
evaluation [
Example 4 (cont’d) A continuous query to notify Claire with instantaneous coupons matching her preferences can be expressed in C-SPARQL and CQELS as follows. For readability, we write <coupons> instead of <http://coupons>, etc.
SELECT ?shop ?product ?percent FROM <products>
STREAM <coupons> [RANGE 30m]
STREAM <locations> [RANGE 5m] WHERE { ?shop :offers ?coupon. ?coupon :reduce ?percent. ?coupon :on ?product. ?user :isNear ?shop. ?product :g_classify ?gender.
FILTER (?percent >= 20 && ?gender != 1)}
SELECT ?shop ?product ?percent FROM <products> WHERE {
STREAM <coupons> [RANGE 30m] { ?shop :offers ?coupon. ?coupon :reduce ?percent.
?coupon :on ?product. } STREAM <locations> [RANGE 5m] {
?user :isNear ?shop. } ?product :g_classify ?gender.
FILTER (?percent >= 20 && ?gender != 1)} Q1: Notification query in C-SPARQL Q2: Notification query in CQELS 2.3
We will gradually introduce the central concepts of LARS [
Example 5 (cont’d) The offers in the running scenario (Ex. 1) can be modeled as a data stream D = (TD; D) with a timeline TD = [0; 50] whose time unit is minutes, and the evaluation function D(10) = fo er (a; \mbt "; 30)g, (15)=fo er (b; \rayban"; 25)g,
D(18) = fisNear (a); isNear (b)g and D(t) = ; for all t 2 TD n f10; 15; 18g. The evaluation function D can be equally represented as
D =
1108 77!! ffiosNeera(ra(;a\)m;ibstN"e;a3r0()bg);g15 7! fo er (b; \rayban"; 25)g; : To cope with the amount of data, one usually considers only recent atoms. Let S = (T; ) and S0 = (T 0; 0) be two streams s.t. S0 S, i.e., T 0 T and 0(t0) (t0) for all t0 2 T 0. Then S0 is called a substream of S.
as input a stream S = (T; ), a time point t 2 T , called the reference time point, and a vector of window parameters x for type and returns a substream S0 of S. Important are tuple-based and time-based window functions. The former select a fixed number of latest tuples while the latter select all atoms appearing in last n time points. Window operators . Window functions can be accessed in formulas by window operators. That is, an expression has the effect that is evaluated on the “snapshot” of the stream delivered by its associated window function w .
By dropping information based on time, window operators specify temporal relevance. For each atom in a window, we control the semantics by some temporal reference. Time Reference. Let S = (T; ) be a stream, a 2 A and B data. Then, at time point t 2 T , A static background – a holds, if a 2 (t) or a 2 B; – 3a holds, if a holds at some time point t0 2 T ; – 2a holds, if a holds at all time points t0 2 T ; and – @t0 a holds, if t0 2 T and a holds at t0.
Next, the set A+ of extended atoms is given by the grammar 3a j 2a; where a 2 A and t is any time point. Expressions of form are called window atoms. Example 6 The window atom 303o er (Sh; Pr ; Pe) takes a snapshot of the last 30 minutes of a stream and uses the 3 operator to check whether an offer from shop Sh on product Pr with a discount of Pe% appeared in the stream during this period. Similarly, 53isNear (Sh) does the same job to take a snapshot of size 5 minutes of the shops near the user. 2.4
We present a fragment of the formalism in [
Syntax. A rule r is of the form (r), where H(r) = is the head and the body of r is (r) = 1; : : : ; j ; not j+1; : : : ; not n. Here, is of form a or @ta, where a 2 AI , and each i is either an ordinary atom or a window atom.
Let B (r) = B +(r) [ B (r), where B +(r) = f i j 1 i jg is the positive and B (r) = f i j j < i ng is the negative body or r. A (LARS) program P is a set of rules. A program is positive, if none of its rules has a negative body atom. Example 7 (cont’d) Suppose we are given static background data B that contains product information in a predicate of form g classify (Pr ; Ge), where Ge = 0 (resp. 1) marks that product Pr is for women (resp., men). The following LARS rule amounts to the queries in Example 4, under the input streams in a format as in Example 5. ans(Sh; Pr ; Pe)
303o er (Sh; Pr ; Pe); 53isNear (Sh); g classify (Pr ; Ge); Pe 20; Ge 6= 1: This rule works as follows. The two window atoms provide offers announced in the last 30 minutes and the shops near the user within the last 5 minutes. Together with the gender classification of products provided by g classify , only products not for men (Ge 6= 1) and have discount rate from 20% are concluded at the head with predicate ans. Semantics. Let P be a LARS program. For a data stream D = (TD; vD), any stream I = (T; ) D that coincides with D on AE is an interpretation stream for D. A tuple M = hT; ; W; Bi is an interpretation for D, where W is a set of window functions w such that the corresponding window operator appears in P , and B is the background knowledge. Throughout, we assume W and B are fixed and thus also omit them.
Satisfaction by M at t 2 T is as follows: M; t j= for 2 A+, if holds in (T; ) at time t; M; t j= r for rule r, if M; t j= (r) implies M; t j= H(r), where M; t j= (r), if (i) M; t j= i for all i 2 f1; : : : ; jg and (ii) M; t 6j= i for all i 2 fj+1; : : : ; ng; and M; t j= P for program P , i.e., M is a model of P (for D) at t, if M; t j= r for all r 2 P . Moreover, M is minimal, if in addition no model M 0 = hT; 0; W; Bi 6= M of P exists such that 0 .
Definition 3 (Answer Stream) An interpretation stream I = (T; ) for a data stream D I is an answer stream of program P at time t, if M = hT; ; W; Bi is a minimal model of the reduct P M;t = fr 2 P j M; t j= (r)g. By AS(P; D; t) we denote the set of all such answer streams I.
Since RSP queries return just a single deterministic answer (which of course can contain multiple rows) at a time point, we consider in this paper LARS programs that have a single answer stream. By AS (P; D; t), we directly refer to the single element of AS(P; D; t).
in Ex. 2. That is, B = f: : : ; g classify (\mbt "; 1); g classify (\rayban"; 0); : : :g. Take the data stream D from Ex. 5 and let P be the LARS program consisting of the single rule in Ex. 7. Then, I = (TI ; I ) is the only answer stream of P wrt. D and B at time t = 18, where TI = TD and I = D [ f18 7! fans(b; \rayban"; 25)gg. 3
Section 2.2 shows a divergence in realizing continuous queries in C-SPARQL and CQELS. To be able to capture and analyze the difference between the two approaches, we need to have a common starting point. This section proposes a formal model of RSP queries that captures this common starting point idea, and then classifies C-SPARQL and CQELS on the model.
Similarly as in [
Example 9 Queries Q1 and Q2 in Ex. 4 stem from Q = (V; P; D; S), where V = f?shop; ?pname; ?percentg P = (P1 [ P2 [ P3) FILTER R
8 ?shop : offers ?coupon: 9 P1= < ?coupon : on ?product: =
: ?coupon : reduce ?percent: ; P2= f ?user
: isNear ?shop: g P3= f ?product : g classify ?gender: g R = (?percent D = f<products>g
20 && ?gender 6= 1) S = (<http://coupons>; [RANGE 30m]; P1); (<http://locations>; [RANGE 5m]; P2) : This query covers all common aspects of Q1 and Q2 which both access the static dataset at <products> and the input streams at <coupons> and <locations> with a window of range 30 and 5 minutes, respectively. On top of the snapshot from the input streams together with the static dataset, a pattern matching is carried out on the graph pattern P . Next, we show how this RSP query model captures the divergent C-SPARQL and CQELS queries. Consider an RSP query Q.
The corresponding C-SPARQL query, denoted by cs(Q), can be obtained from Q by setting the graph patterns in all stream graph patterns in S to ;. This goes along with the idea of C-SPARQL to merge patterns on the input streams into the default graph.
A corresponding CQELS query, however, can be obtained from Q at different levels of cautiousness: for every part of P that contains gi s.t. (si; !i; gi) 2 S, replace it with either (i) (STREAM si !i gi), or (ii) ((STREAM si !i gi) UNION gi). The former is a brave approach when one can make sure that the static dataset and the stream si do not share patterns, while the latter is more cautious when one is not sure and rather expects triples matching gi come from either the static dataset or the input streams. Therefore, Q is corresponding to a set cq (Q) of 2jSj CQELS queries, including a brave one, a cautious one, and the ones in between. Note that Q2 in Ex. 4 is the brave CQELS query of Q. 4
We will use the following strategy to capture different RSP approaches with LARS:
(1) First, the two push- and pull-based execution modes can be applied to LARS
programs in general via two straightforward translations.
(2) Then, window expressions in RSP are translated into window operators in LARS.
(3) Next, R2R operators and the approaches in building the datasets to be evaluated by
R2R operators are captured by two slightly different translations 1 and 2, based on
the translation from SPARQL to Datalog rules in [
We now go into details of (1)-(3). 4.1
This section provides two translations that capture the push- and pull-based execution modes by means of LARS itself. Given a LARS program P and a pulling period U > 0, the translations (P ) and (P; U ) encode the push- and pull-mode by LARS rules, respectively. Intuitively, we add to the body of each rule in P an ordinary atom trigger. Then, rules to conclude trigger are added depending on the mode. For push-based mode, trigger will be concluded per new incoming input triple. For pull-based mode, the condition is that the current time point is a multiple of U .
Formally speaking, for a LARS rule r, a LARS program P , a pulling period U , let trigger (r) = H (r)
B (r); trigger: trigger (P ) = ftrigger (r) j r 2 P ^ B (r) 6= ;g
(P ) = trigger (P ) [ ftrigger (P; U ) = trigger (P ) [ ftrigger
NOW p(X): j p 2 AI g Notably, the translation for the pull-based mode needs to acquire the current time point, which is achieved as follows. The logical constant true always holds, and thus @T true holds for all considered time points T . By applying window operator NOW (or equivalently 0) before, only the current time point will be selected. The following proposition shows that and faithfully capture the execution modes. Proposition 1 Let P be a LARS program, U be a positive integer, and D = (TD; D) be an input stream. For every t 2 TD, it holds that (1) If D(t) 6= ;, then AS( (P ); D; t) = AS(P; D; t) (2) If D(t) = ;, then AS( (P ); D; t) = fDg (3) If t % U = 0, then AS( (P; U ); D; t) = AS(P; D; t) (4) If t % U 6= 0, then AS( (P; U ); D; t) = fDg. 4.2
[RANGE UNBOUNDED]
(!)
L
L;0;D
N
#
0 or NOW
1
For capturing R2R operators of continuous SPARQL queries we can exploit an existing
translation from SPARQL to Datalog rules [
We propose two strategies (T1) and (T2) to extend the translation in [
LARS rules.
Towards formally presenting our translations, we start with a review of the translation
from SPARQL to Datalog in [
Intuitively, converts SPARQL operators to declarative rules.
Our purpose is to provide a translation for theoretical analysis rather than for practical implementation of RSP queries. Thus, we concentrate on (ii). For (i), we assume that – each triple (s; p; o) from the static dataset D can be accessed by triple(s; p; o; D), – each triple (s; p; o) arriving at a stream s at time t contributes to the evaluation function at t under a predicate striple, that is, striple(s; p; o; s) 2 (t). 3 http://www.kr.tuwien.ac.at/research/systems/dlvhex/ (V; (S; P; O); D; S; i) (V; (P1 AND P2); D; S; i) = ansi(V ; D; S)
triple(S; P; O; D) = (vars(P1); P1; D; S; 2i)[ (vars(P2); P2; D; S; 2i + 1)[ ansi(V ; D; S) ans2i(vars(P1); D; S); ans2i+1(vars(P2); D; S): (V; (P1 UNION P2); D; S; i) = (vars(P1); P1; D; S; 2i)
(vars(P2); P2; D; S; 2i + 1)[ ansi(V [(V n vars(P1)) ! null]; D; S) ansi(V [(V n vars(P2)) ! null]; D; S) ans2i(vars(P1); D; S): ans2i+1(vars(P2); D; S): (V; (P1 MINUS P2); D; S; i) = (vars(P1); P1; D; S; 2i)[
(vars(P2); P2; D; S; 2i + 1)[ ansi(V [(V n vars(P1)) ! null]; D; S) ans2i(vars(P1); D; S); not ans02i(vars(P1) \ vars(P2); D; S): ans02i(vars(P1) \ vars(P2); D; S) ans2i+1(vars(P2); D; S): (V; (P1 OPT P2); D; S; i) = (V; (P1 AND P2); D; S; i)[
(V; (P1 MINUS P2); D; S; i) (V; (P FILTER R); D; S; i) = (V; P; D; S; 2i)[
LT (ansi(V ; D; S) ans2i(vars(P ); D; S); R:) (V; (GRAPH g P ); D; S; i) = (V; P; g; S; i) for g 2 V [ I
ansi(V ; D) ansi(V ; g); isIRI(g); g 6= default:
Fig. 1: Extending translation in [
(!)3striple(S; P; O; s) j (s; !; g) 2 Sg: The translation for strategy (T1) is 1(V; P; D; S; i) = (V; P; D; S; i) [ 0(S). For (T2), let 2 be a function that agrees with , and moreover fulfills: 2(V; (STREAM s ! g); D; S; i)=ansi(V ; D; S) (!)(
^ (S;P;O)2g 3striple(S; P; O; s)): When it is clear from context, we will write in the sequel = i(Q) (i 2 f1; 2g), for a query Q = (V; P; D; S) instead of = i(V; P; D; S; 1).
Given an RSP query Q=(V; P; D; S), let Q02fcs(Q)g[cq (Q) and Ii=AS ( i(Q0); D; t)
for a data stream D and a time point t, where i 2 f1; 2g. We denote the set of atoms of
predicate ansj with the parameter corresponding to S projected away by
chop(I; Q) = fansj (Vj ; D) j ansj (Vj ; D; S) 2 Ig [ (I n fansj (Vj ; D; S) 2 Ig).
The following result shows that our translation preserves the translation in [
This section discusses ideas to tackle the comparison between C-SPARQL and CQELS using LARS. The main questions are: (i) What do we mean by saying “C-SPARQL and CQELS return the same output?” (ii) When do the two approaches of dealing with streaming input coincide, i.e., merging input streams into the default graph and using stream graph patterns do not make any difference in building the datasets evaluated by the engines? (iii) Under which conditions will the push- and pull-based executions fulfill (i)? The most important question is (i), which establishes the whole setup and methodology for the comparison. Since C-SPARQL and CQELS build on a pull- and push-based mode, respectively, it does not make sense to require that the output of the two engines coincides at every single time point. Instead, we need some notion of agreement between the two engines. Intuitively, we say that C-SPARQL and CQELS agree on a time interval [t1; t2], if the union of outputs returned by CQELS at every time point in (t1; t2] (i.e., the total output in that interval) coincides with the output returned by C-SPARQL at t2. Assume that both two engines start at time 0 of a timeline T and C-SPARQL is executed with a pulling period of U time units. Then, we say that the engines agree on T just if they agree on every interval [i U; (i + 1) U ] 2 T , for i 0. On the translated LARS programs, checking for agreement on the output boils down to checking the agreement on the output facts of the predicate ans1.
Imagine a user staring at his App to wait for notifications. If the two engines agree, then at the end of the pulling period, the user will see the same notifications in both cases; if the pulling period is short enough, she might even not notice any difference between the two modes. This makes the agreement notion reasonable from a practical point of view.
Now we analyze possible sufficient conditions of agreement for C-SPARQL and CQELS, by finding answers for questions (ii) and (iii).
To put (ii) more concretely in the context of LARS, take an RSP query Q and let Q1 = cs(Q) and Q2 2 cq (Q); we want to find conditions under which the answer streams of 1(Q1) and 2(Q2) at time point t on the input stream D have the same extension of predicate ans1.
Observe that 1(Q1) and 2(Q2) differ on dealing with input coming from predicate striple. The former program merges facts of striple into triple, which will be used later on to conclude ansi; the latter program concludes ansi directly from striple. If the static dataset and the input streams share some patterns, then 1 may not be capable of identifying the origin of some triple. Even if the input stream does not receive any incoming triple, 1(Q1) may still conclude some output due to facts staying in the static part. On the other hand, 2(Q2) returns no output as the stream is not updated. To avoid such confusion, it is required that the static dataset and the input streams do not share patterns. Fortunately, this usually holds in practice. For instance, in our running example, the pattern ?user :isNear ?shop can only match triples in the stream <http://locations>, since predicate :isNear will not be streamed in <http://coupons>, and the static dataset should not contain such information.
When the conditions for question (ii) are fulfilled, one has a better setting to
analyze (iii). Still, for the first step, we need to assume as in [
However, this condition cannot be guaranteed under high throughput. The reason is that with dense input streams, the snapshots taken at time points near the beginning of an interval will have high chances to collect more input than the snapshot at its end. Thus, in practice it is rather unlikely that C-SPARQL and CQELS will agree, due to the strong semantic implications of push/pull-based querying.
Conclusions and Outlook. This paper establishes first steps towards formally comparing two RSP semantics implemented in two well-known engines, namely C-SPARQL and CQELS, by proposing translations to capture the languages and execution modes of the engines, and discussing idea to formalize a notion of agreement between the two semantics as well as a condition for it to hold. Next steps include working out these ideas formally and implementing the translations to build a comparison benchmarking systems of different stream processing approaches using LARS.