RDF Stream Processing (RSP) is gaining momentum. The RDF stream data model is progressively adopted and many SPARQL extensions for continuous querying are converging into a unified RSP query language. However, the RSP community still has to investigate when transforming data streams in RDF streams pays off. In this paper, we report on several experiments on a revolutionized version of our Streaming Linked Data framework (namely, SLD Revolution). SLD Revolution adopts i) Generic Programming, i.e. it operates on time-stamped generic data items, and ii) it applies a lazy-transformation approach, i.e. it postpones the RDF stream transformation until it can benefit from it. , processing data according to their nature (event-, tuple-, tree- and graph-based). SLD Revolution results to be a cheaper (it uses less memory and has a smaller CPU load), faster (it reaches higher maximum input throughput) yet more accurate (it provides a smaller error rate in the results) solution than its ancestor SLD.
or the March 2015 eclipse in Europe. Those deployments process in real-time up to 3000 microposts per minute on a 50 C/month machine in the cloud (8 GB of RAM and 4 cores), providing semantic analysis and sophisticated visualizations.
In five years of SLD usage, we learned that using RDF streams is valuable when
i) data are naturally represented as graphs, i.e. micro-posts in the larger social graph,
and when ii) the availability of popular vocabularies makes easy writing adapters that
semantically annotate the incoming data, e.g. we wrote adapters that annotated streams
from the major social networks using SIOC [
However, we have also found out several weaknesses of the approach: – RDF streams cannot be found in the wild, yet. JSON is largely used in practice (e.g.,
Twitter Streaming APIs3 and W3C activity stream 2.0 working draft4).
– The results of C-SPARQL queries are often relational and forcing them into an RDF
streams is not natural, i.e., a user would naturally use the REGISTER QUERY ...
AS SELECT ... form instead of REGISTER STREAM ... AS CONSTRUCT ... one.
It takes three triples to state how many times a hashtag appears in the micro-posts
observed in 1 minute, while the tuple htimestamp, hashtag, counti is more succinct.
– It is harder to express some computation using C-SPARQL over RDF streams and
graphs than writing a path expression over JSON; or writing an SQL query over
relations; or writing an EPL statement over events.
– SLD builds on the C-SPARQL engine and, thus, shares with it some shortcomings,
i.e. it gives incorrect answers when it is overloaded engine [
In this paper, we challenge the hypothesis that RDF streams should play such a central role in SLD. We investigate if i) using time-stamped generic data items (instead of focusing only time-stamped RDF graphs) and ii) processing them according to their event-, tuple-, tree- and graph-based nature, offer the opportunity to engineer a cheaper (uses less memory and CPU), faster (it reaches higher maximum input throughput) yet more accurate (i.t. with a smaller error in the results) version of SLD that we called SLD Revolution. We bring experimental evidence that supports the design decision of revolutionizing SLD in those two directions. Using our experience on social media monitoring, we design a set of experiments to evaluate our hypothesis: we chose the expected maximum rate of micro-posts per minute and the machine to deal with the worst-case scenario; by reducing the available memory and processor time requirement we push the overload status to an higher input rate.
The remainder of the paper is organized as follows. Section 2 introduces the state-ofthe-art in stream processing with a special cut on RSP. Section 3 presents SLD Revolution and its new processing model. Sections 4 and 5, respectively, describe the settings and the results of the experiments we run. Finally, in Section 6, we conclude and present our future work. 2
Data model. RSP extends the RDF data model and the SPARQL query model in order
to take into account the streaming nature of the data.
3 https://dev.twitter.com/streaming/overview
4 http://www.w3.org/TR/activitystreams-core/
A relational data stream [
An RDF Stream is defined in the same way, but di is either an RDF statement
(as done in most of the RSP approaches [
Processing Model. Also the processing model of RSP [
Figure 2 illustrates the architecture of SLD that offers: (i) a set of adapters that
transforms heterogeneous data streams in RDF streams attaching to each received
element a time-stamp that identifies the ingestion time (e.g., a stream of microposts
in JSON as an RDF stream using the SIOC vocabulary [
SLD Revolution adopts generic programming [
Figure 3 illustrates the SLD Revolution processing model. The stream and the relation concepts of CQL are mapped to generic data streams ShT i and to instantaneous generic data items IhT i.
In line with CQL and RSP-QL, SLD Revolution proposes three classes of operators: – The stream-to-instantaneous S2I hT i operators transform the infinite generic data stream ShT i in to a finite bag of instantaneous generic data items IhT i. – The instantaneous-to-instantaneous I2I hT ; T 0i operators transform instantaneous generic data items IhT i into other instantaneous generic data items IhT 0i, where T and T 0 can be of the same type or of different types. For instance, a
S2I<T > I2I<T,T' >
I2S<T >
Fig. 3. The processing model of the SLD revolution framework.
C-SPARQL query of the type REGISTER QUERY ... AS SELECT ... takes in input time-stamped RDF graphs and generates as output time-stamped relations. – The instantaneous-to-stream I2S hT i operators transform instantaneous generic data items IhT i into a generic data stream ShT i.
Source Stream
P T T H
Streaming Linked Data Revolution Server Receiver
Generic Stream Bus
Translator Recorder
Re-player
Processor
Decorator
P T T H
Sink Stre1am
SLD Revolution generalizes SLD architecture (cf. Figure 4 with Figure 2). The
Generic Stream Bus replace of the RDF stream bus. The receivers replace the adapters.
As the adapters they allow to ingest external data streams, but they no longer transform
the received events in time-stamped RDF graphs. Data items remain in their original form,
only the ingestion time is added, postponing the transformation to the moment when
it is required (we name this approach lazy transformation). The processors substitute
the C-SPARQL-based analyzers. The C-SPARQL engine remains as one of the possible
processors, but SLD Revolution can be extended with any continuous processing engine.
The current implementation includes the Complex Event Processor Esper, the SPARQL
Engine Jena-ARQ that operates on time-stamped RDF graphs one at a time, and a custom
JSON path expression evaluation engine built on gson (https://github.com/google/gson).
Translators generalize publishers, which are specific for the Streaming Linked Data
format [
In this Section, we present the experimental settings of this work. As domain we chose Social Network analysis as done by the Linked Data Benchmark Council (LDBC) in the SNBench5. We first explain the type of data we used for our experiments. Then, 5 http://www.ldbcouncil.org/benchmarks/snb we explain how the data were sent to SLD and SLD Revolution. We describe the two continuous processing pipelines that we registered in SLD and in SLD Revolution. Finally, we state which key performance indicators (KPIs) we measure and how.
Input Data. SLD and SLD Revolution receive information in the same way, they both connect to a web socket server and handle JSON-LD files.
4 5 6 7 8 9 10 11 12 13 1 {"@context": { ... }, "@type": "Collection", "totalItems": 1, 2 "prov:wasAssociatedWith": "sr:Twitter", 3 "items":[{ "@type":"Post", "published":"2016-04-26T15:40:03.054+02:00", "actor":{"@type":"Account", "@id":"user:1", "sioc:name":"@streamreasoning"}, "object":{ "@type":"Content", "@id":"post:2", "alias":"http://.../2", "prov:wasAssociatedWith":"sr:Twitter", "sioc:content":"You ARE the #socialmedia!", "dct:language":"en", "tag":[{ "@type":"Tag", "@id":"tag:3", "displayName":"socialmedia"}]} }]} Listing 1. JSON representation of a Twitter micro-post. Due to the lack of space we omitted the context declaration that contains the namespace.
In Listing 1, we propose a JSON-LD serialization of the Activity Stream representation of a tweet as it was injected during the experiments in both systems. The JSON-LD representation of an Activity Stream is a Collection (specified by @type property) composed by one or more social media items. The Collection is described with two properties, i.e., totalItems and prov:wasAssociatedWith, which tell respectively the number of items and the provenance of the items. The collection in the example contains a Post created on 2016-04-26 (published property) by an actor (Lines 11-16) that produce the object (Lines 17-33). The Actor has a unique identifier @id, a displayName, a sioc:name and a alias. The Object has a sioc:content, a dct:language, zero or more tags, and optionally a url and a to to represent, respectively, links to web pages and mentions of other actors. 1 <post:2> a sma:Tweet ; 2 dcterms:created "2016-04-26T15:46:43.346000+02:00"ˆˆxsd:dateTime ; 3 dcterms:language "en"ˆˆxsd:string ; 4 sioc:content "You ARE the #socialmedia!"ˆˆxsd:string ; 5 sioc:has_container "Twitter"ˆˆxsd:string ; 6 sioc:has_creator <user:1> ; 7 sioc:id "2"ˆˆxsd:string ; 8 sioc:link "http://.../status/2"ˆˆxsd:string ; 9 sioc:topic <tag:3> . 10 <tag:3> a sioct:Tag ; 11 rdfs:label "socialmedia"ˆˆxsd:string . 12 <user:1> a sioc:UserAccount ; 13 sioc:account_of "StreamReasoning"ˆˆxsd:string ; 14 sioc:creator_of <post:4> ; 15 sioc:id "1"ˆˆxsd:string ; 16 sioc:name "@streamreasoning"ˆˆxsd:string .
Listing 2. RDF N3 representation of a Twitter micro-post
Listing 2 shows the RDF produced by the SLD adapter in transforming the JSON-LD in Listing 1. The translation operation exploit well known vocabularies, in particular sioc to represent the online community information, prov to track the provenance of an item and dcterms to represents information about the object.
Sending data. A test consists in sending a constant amount of synthetic data using the JSON-LD serialization presented in Listing 1. The data is sent in bunches 3 times per minute (i.e. at the 10th, the 30th and the 50th seconds of the minute). Each bunch contains the same amount of posts. We tested the configuration for different rate: 1500 posts per minute (i.e., 3 bunch of 500 posts), 3000 posts per minute, 6000 posts per minute, 9000 posts per minute, 12000 posts per minute and 18000 posts per minute.
The rates and and the input methodology were chosen based on our experiences on social monitoring (see Section 1). They test a normal situation for SLD (1500 and 3000 posts per minutes) as well as situations that we know to overload SLD (more than 6000 posts per minute).
Pipelines.We tested SLD and SLD revolution with different pipelines: – the area chart pipeline computes the number of tweets observed over time. It uses a 15 minute long window that slides every minute. The results can be continuously computed i) using a generic sliding window operator, which works looking only to the time-stamps of the data items in the generic stream, and ii) accessing with a path expression the totalItems property in the JSON-LD file, i.e., the number of items in the collection. – the bar chart pipeline counts how often hashtags appear in the tweets received in the last 15 minutes. As the area chart pipeline, the window slides every minute. In this second pipeline, RDF streams are adequate and it is convenient to write a C-SPARQL query that counts the number of times each hashtag appears.
The two pipelines are coded in SLD and SLD Revolution in two different ways. SLD performs the transformations of JSON-LD in RDF by default, on all the input data, independently from the task to perform. SLD Revolution keeps the data in its original format as much as possible, i.e., it performs lazy transformations.
SLD Pipelines. Figure 5 presents the two pipelines in SLD. The input data are translated in RDF as soon as they enter the pipelines. The computations for the area chart and for the bar chart (see the part marked with A and B ) are composed by the same type of components and share the new RDF stream translated by the Adapter.
The pipeline A uses two C-SPARQL queries. The first (see Listing 3) applies a tumbling window of 1 minute6 and counts the tweets. 6 A tumbling window is a sliding window that slides for its length
The second aggregates the results from the first query using a 15 minutes time window that slides every minute (see Listing 4).
1 REGISTER STREAM presocialstr AS 2 CONSTRUCT { ?id sma:twitterCount ?twitterC } 3 FROM STREAM <http://.../socialstr> [RANGE 1m STEP 1m] 4 WHERE { SELECT (uuid() AS ?id) ?twitterC 5 WHERE { SELECT (COUNT (DISTINCT ?mp) AS ?twitterC) 6 WHERE { ?mp a sma:Tweet } } }
Listing 3. C-SPARQL pre-query for the area chart that applies a tumbling window of 1 minute and counts the tweets. 1 REGISTER STREAM ac AS 2 CONSTRUCT { ?uid sma:twitterCount ?totTwitter ; sma:created_during ? unixTimeFrame } 3 FROM STREAM <http://.../presocialstr> [RANGE 15m STEP 1m] 4 WHERE { 5 SELECT (uuid() AS ?uid) ?unixTimeFrame (SUM(?twitter) AS ?totTwitter) 6 WHERE { ?id sma:twitterCount ?twitter ; sma:created_during ?timeFrame . 7 ?timeFrame a sma:15mTimeFrame ; sma:inUnixTime ?unixTimeFrame } 8 GROUP BY ?unixTimeFrame }
Listing 4. C-SPARQL query for the area chart that aggregates the results from the query in Listing 3 using a 15 minutes time window that slides every minute.
It is worth to note that the first query is an important optimization in terms of memory consumption. It avoids the engine to keep 15 minutes of tweets to only count them. In SLD we often use this design pattern, we call this first query a pre-query.
Pipeline B also exploits this design; it applies a pre-query to reduce the amount of data and then a query to produce the final result.
It is also worth to note that all the C-SPARQL queries use the form REGISTER STREAM ... AS CONSTRUCT ..., because RDF streams are the only means of communication between SLD components.
The last components of both pipelines are publishers that make the results available to external process outside SLD. In this case, the publisher writes JSON files on disk.
SLD Revolution Pipelines. Figure 6 presents the pipelines in SLD Revolution. As for SLD, the pipeline A is for the area chart, while B is for the bar chart. The first component is no longer an adapter. The data directly enter SLD revolution in JSON-LD format. The first query is a generic 1 minute long tumbling window implemented with the EPL statement in Listing 5. FORCE UPDATE and START EAGER tells the stream processing engine, respectively, to emit also empty reports and to start processing the window as soon as the query is registered (i.e., without waiting for the first time-stamped data item to arrive). It is worth to note that this query exploits the event-based nature of the generic stream it is observing; it does not inspect the payload of the events, it only uses their time-stamps.
select * from GenericEvent.win:time_batch(1 min, "FORCE_UPDATE, START_EAGER") Listing 5. The generic window query in common to both the pipelines in SLD Revolution As explained in Section 3, processors are the central components of SLD Revolution. They can listen to one or more generic streams, compute different operations and push out a generic streams. The type of the input and output streams can be different. The two pipelines uses different processors (e.g. RDF translator, windower and SPARQL).
SLD Revolution maintains the data format as long as possible in order to reduce the overhead of the translations. As already said, SLD Revolution can exploit the tree-based nature of JSON-LD. In pipeline A, it exploits a path expression data to extract totalItems, i.e., the number of items in each collection, from the time-stamped JSON-LD items in the generic stream it listens to. It outputs a tuple htimeframe,counti that is aggregated every minute over a window of 15 minutes using an EPL statement.
The Pipeline B of SLD Revolution translate JSON-LD in RDF in order to extract information about the hashtags. As for the pipeline B of SLD, we use a pre-query design pattern to reduce the amount of data. A SPARQL processor applies the SELECT query in Listing 6 to every data-item in the generic stream it listens to and pushes out a stream of tuples hhashtagLabel,counti. The relational stream is then aggregated with an esper processor with a 15 minute time window that slides every 1 minute (see Listing 7). 1 SELECT ?htlabel (COUNT(DISTINCT(?mpTweet)) AS ?htTweetCount) 2 WHERE { ?mpTweet a sma:Tweet ; sioc:topic ?tweetTopic . 3 ?tweetTopic a sioctypes:Tag ; rdfs:label ?htlabel } 4 GROUP BY ?htlabel 5 ORDER BY desc(?htTweetCount)
Listing 6. SPARQL pre-query for the bar chart select htlabel, sum(count) as sumHt from HTCountEvent.win:time(15 min) group by htlabel output snapshot every 1 min
Listing 7. EPL query for the bar chart
KPIs. As key performance indicators (KPIs), we measure the resources consumption of the two systems and the correctness of the results. For the resource consumption we measure every 10 seconds: i) the CPU load of the system thread in %, ii) the memory consumption of the thread in MB and iii) the memory consumption of the Java Virtual Machine (JVM). For the correctness, we compared the computed results with the expected results. Being the input a constant flows of tweets that only differ for the ID, the area chart is expected to be flat and the bar chart is expected to count exactly the same number of hashtags every minute. 1 Figure 7 offers an overview of the results of the experiments. The full results are reported at the end of this section in Figure 10. On the X axis we plot the median of the CPU load in %, while on the Y axis we plot the memory allocated by the engine thread. The size of the bubble maps the median of the error of the area chart. Bubbles in the lower left corner correspond to the experiment where we sent 1500 tweets per minute.
Increasing the throughput results in more memory consumption and CPU load for both systems. However, not SLD Revolution consumes less memory than SLD and occupies less CPU. Moreover, SLD Revolution presents a linear increment for both these KPIs, while the resource usage for SLD grows exponentially with the throughput. Also the error in the results increases with the throughput: SLD already shows an error greater than 3% in the bar chart at 3000 tweets per minutes and in the area chart at 9000 tweets per minute; SLD Revolution is faster - i.e. it reaches higher maximum input throughput - and more accurate – i.e. it reaches 3% error level only for 18000 tweets per minutes, providing more precise results than SLD.
Figure 8 presents the recorded time-series for CPU load and memory usage in both systems. The memory usage graphs contains two different time series. The blue one represents the memory usage of the system thread, while the orange one shows the total memory usage for the JVM.
The memory usage of the system thread accounts for all the components and data in the pipeline. Notably, when the system under testing is not overloaded, the memory usage is constant over time, while when the system is overloaded it grows until the system crashes. The total memory usage of the JVM shows, instead, the typical pattern of the garbage collector that lets the JVM memory grow before freeing it. Also in this case, when the system it is overloaded, the garbage collector fails to free the memory.
During the experiments the median of the memory used by SLD spans from 115 MB, when loaded with at 1500 posts/min, to 1.6 GB, when loaded with 18000 posts/min. For SLD Revolution, instead, it spans from 44 MB to 511.5 MB in the same load conditions. The experimental results clearly shows that SLD Revolution consumes (in average) three times less memory than its ancestor.
The same considerations can be proposed for the CPU load. The median of the CPU load spans from 2% to 10% for SLD Revolution, while it spans from 10% to 39.5% for SLD. SLD Revolution consumes in average 4 time less CPU time than SLD.
The correctness results are summarized in Figure 9. As explained in Section 4, the percentage of errors is computed by comparing the results for each time interval with the expected ones. The X axis of each plot shows the percentage error; it ranges from 0% to 100%. The Y axis is the percentage of results with that error; it also goes from 0% to 100%. A bar as tall as the Y axis in the left side of the graph means that all results where correct. The smaller that bar is and the greater the number of bars to its right is, the more errors were observed.
In general, the results shows that SLD Revolution is more accurate (the result error is smaller) than SLD. For the area chart the distribution shows that SLD Revolution percentage of error is very low when the input throughput is between 1500 posts/min and 9000 posts/min. When it is higher (i.e., 12000 and 18000 posts/min) also SLD Revolution starts suffering and percentage of errors starts growing. For SLD, errors are present even at lower input rate, the graph shows that the error distribution starts moving to the right at 6000 posts/min. Similar consideration can be proposed for the bar chart error distribution. The degradation of performance of SLD starts a very low rate, a substantial presence of errors around 7% can be seen with 6000 posts/min in input.
Figure 8 and Figure 9 show the deep correlation between resources usage and errors. Clearly, a growing input throughput drives the systems to be less reliable. For both the versions of the SLD framework (SLD Revolution and SLD) the correctness of the results decreases as soon as the machine is overloaded and the resources usage starts rising out of control.
SLD Revolution Min. 1st Qu. Median 3rd Qu. Max.
42 43 44 45 48
45 48 50 52 58
52 61 64 67 78
59 73 80 84 102
66 88 97 107 144
212 392 511,5 597,2 774
1,3% 1,8% 2,0% 2,2% 3,4%
1,3% 2,0% 2,2% 2,5% 3,8%
1,4% 2,1% 2,4% 2,8% 4,5%
1,5% 2,3% 2,7% 3,2% 5,1%
1,3% 2,6% 3,2% 4,2% 8,9%
2,8% 7,2% 10,3% 13,8% 26,2%
0,0% 0,0% 0,0% 0,0% 33,3%
0,0% 0,0% 0,0% 0,0% 33,3%
0,0% 0,0% 0,0% 0,0% 33,3%
0,0% 0,0% 0,0% 0,0% 33,3%
0,0% 0,0% 16,7% 33,3% 33,3%
0,0% 16,9% 32,1% 53,7% 100,0%
0,0% 0,0% 0,0% 0,0% 93,3%
0,0% 0,0% 0,0% 0,1% 2,2%
0,1% 0,1% 2,3% 2,3% 2,3%
1,6% 2,4% 2,4% 2,4% 3,2%
0,2% 1,5% 2,5% 4,7% 6,1%
2,9% 9,4% 12,6% 16,3% 22,4%
In our future work, we intend: i) to empirically demonstrate the value of using SLD
Revolution for all our deployments of socialOmeters, and ii) to investigate if SLD
Revolution can be the target platform for a new generation of Ontology Based Data
Integration [
As for the former, we first intend to stress test SLD Revolution using workloads that resemble reality. Then, we aim at putting it at work in parallel to SLD in real-world deployments. Once we will have collected enough evidence that SLD Revolution is always cheaper faster yet more accurate than SLD, we will start using it for all our deployments.
As for the latter, we aim at further investigating the generic processing model
presented in Section 3. We are defining an algebra able to capture the semantics of
complex stream processing applications that need to integrate a variety of data sources.
The current sketch of this algebra uses S2I and I2S operators from CQL [
In our opinion, the grand challenge is how to fit all those formal elements in a coherent framework that allows a system to automatically decide which is the latest moment for transforming data (i.e., introducing the concept of lazy transformation) and to perform optimization such as introducing the pre-query that we put in all the pipelines illustrated in Section 4.
When the work on this formally defined generic stream processing model will be
completed, we will be able to start investigating how to extend mapping languages
like R2RML7 and, potentially, also ontological languages in order to make them
timeaware [