We propose a minor extension of the Graph Store Protocol and the SPARQL syntax that should be su cient to enable the application of rules to some kinds of RDF Streams (as de ned by the the RDF Stream Processing W3C Community Group) to generate new RDF streams. The IRI identifying an RDF stream is extended with a query string eld-value pair de ning a sequence of windows of the stream. The extended IRI is passed to a data concentrator, which recasts the stream as a dynamic RDF dataset. A SPARQL query may then be applied to this dynamic dataset whenever it is updated, i.e. when the content of the next window as been fully received. The SPARQL query uses the CONSTRUCT form to generate a new element of the resultant RDF stream. The approach is illustrated with a prototypical example from the healthcare domain.
RDF Graph Stores are increasingly used in the private and public sector to manage knowledge. In practice, although these Graph Stores can be dynamic, most individual transactions involve either complete replacement or small changes. When transactions are made by many users simultaneously or through streaming from data sources, the rate of change can be quite fast, relative to the timescale of query processing. E ectively delivering an up-to-date view (i.e. results of a stored query) of a dynamic Graph Store in real time requires a di erent approach to query processing than that implemented in most RDF query engines currently available.
The RDF standards have not addressed dynamics to a signi cant extent,
other than to informally propose a name: \We informally use the term RDF
source to refer to a persistent yet mutable source or container of RDF graphs.
An RDF source is a resource that may be said to have a state that can change
over time. A snapshot of the state can be expressed as an RDF graph." [
The RDF Stream Processing Community Group3 is taking on a specialized case of dynamic RDF called an RDF Stream, corresponding to the case of a 3 https://www.w3.org/community/rsp/ sequence of RDF datasets called \timestamped RDF graphs", where a particular triple in the default graph has been designated as the timestamp triple. A sequence of overlapping windows (contiguous subsequences) can, in general, be considered as snapshots of a dynamic Graph Store4 such that whenever the window shifts to a new position, a few stream elements expire from and a few new stream elements are added to the dynamic Graph Store. 1.1
Business Drivers
Autonomous, distributed sources will eventually yield data in quantities limited
only by the aggregate bandwidth of wireless transmission media. It has been
estimated, the market for devices and services in support of this processing will
reach a value in excess of twelve digits by 2020 [
The task to purposefully process this data promises to be formidable. A service which leverages the advantages of RDF data processing, to provide a mechanism for rule based integration, validation and evaluation of diverse streaming data sources, will o er signi cant advantage.
Particular drivers in this setting are the following:
{ develop a mechanism to augment assertions during query processing of
dynamic RDF Graph Stores, at basically the complexity of SPARQL construct
{ create means to express actions that should happen when a speci ed integrity
constraint is violated by a dynamic RDF Graph Store
{ implement real-time processing of such features with good scaling properties
in size and rate of change of the Graph Store
{ enable linked data to full l compliance requirements [
RDFS and OWL are the traditional technologies for expressing conceptual models for RDF data, and can in theory be used to augment assertions during query processing, e.g. SPARQL under the RDFS- or OWL-entailment regime. 4 Because blank nodes are shared between elements of an RDF stream, then the dynamic Graph Store must also share blank nodes between versions if this correspondence is to hold.
Due to the Open World Assumption, RDFS and OWL models cannot be used for validation of RDF according to a data model, except for the extreme case that results in inconsistency.
SPARQL queries express validation and integrity constraints well, but the queries for these tasks are not intuitive, and thus are di cult to properly construct and maintain. It is imperative that extensions to incorporate rules and process data streams avoid compounding these di culties.
The challenge is to develop a language with the optimum satisfaction of the
following characteristics:
{ syntax that is both intuitive for the audience (close to RDF) and su ciently
expressive
{ has well-de ned execution semantics
{ can be integrated into remote service control and data ows.
{ permits real-time performance, which requires e cient target data-set
generation (whether update di s or inherent stream structure) and an activation
network implementation. [
Numerous alternatives (see Table 1) have been proposed to query dynamic RDF data. These include proposals for both streaming and versioned data. In order to provide a service which supports queries of adequate complexity, we expect we will need to provide the following combinations and variations: { treat queries as autonomous views in order to promote reuse; { separate query expressions from temporal attributes in order to permit combinations; { permit temporal attributes to be computed as an aspect of query processing; { remain compatible with standard SPARQL in order to permit sharing and simplify development { remain compatible with standard RDF data models { permit named graphs in order to ful l application requirements and permit compatibility with standard encodings { restrict the processing model to just essential components in order to limit the deployment and management complexity
This perspective eliminates those proposals which either reify statements in the data model or add a temporal term. It also argues against a processing model which involves query re-writing and delegation. It also eliminates language extensions which encode temporal attributes as static values in special clauses. It leads to the proposal for Dydra, to combine REVISION and WINDOW speci cations for temporal properties to designate the target dataset, and to make the temporal state available through the function THEN (e.g. returns the time when the processor had received all the elements in a stream window) and permit the window or revision expression to contain variables, but otherwise to conform to
4
SPARQL Inferencing Notation (SPIN), also called SPARQL Rules, is a W3C Member Submission5 as well as a \defacto industry standard"6 implemented by Top Braid. The Member Submission de nes the SPIN syntax via a schema, but does not provide semantics. In particular, although the SPIN submission is claimed to be an alternative syntax for SPARQL, the transformation from SPIN to SPARQL has not been published. Further, although SPIN uses an RDF-based syntax, this alone is not su cient to achieve rule semantics; a su ciently expressive entailment regime must be de ned (e.g. based on the as-yet unpublished transformation to SPARQL) that asserts the constructed source jointly with the queried sources.
Shape Expressions Language (ShEx) is speci ed in a W3C member
submission7. It is intended to ll the same role for RDF graphs that the schema
language Relax NG lls for XML, and its design is inspired by Relax NG,
particular in its foundation on regular expressions. ShEx has both a compact
and an RDF-based syntax, in the same way that Relax NG has a compact
and an XML-based syntax. The execution semantics of ShEx is well-de ned,
but is of limited expressivity. In particular, ShEx does not support named
graphs or RDF Datasets. ShEx precedents are strictly targeted REST [
Shapes Constraint Language (SHACL) is intended as an successor and extension of ShEx developed by the RDF Shapes Working Group.8. SHACL does not support named graphs or RDF Datasets directly, but has an advanced syntax that allows the use of SPARQL. SHACL does not support modularization; it is not obvious how to make this extension. Like ShEx, SHACL has no constructive capability, and so can only perform validation. Semantic Web Rule Language (SWRL) is a W3C Member Submission9 with well-de ned semantics having a logical expressivity that exceeds SPARQL. 5 SPIN - Overview and Motivation https://www.w3.org/Submission/2011/
SUBM-spin-overview-20110222/ 6 http://spinrdf.org/ 7 Shape Expressions 1.0 De nition: https://www.w3.org/Submission/shex-defn/ 8 https://www.w3.org/2014/data-shapes/wiki/Main_Page 9 https://www.w3.org/Submission/SWRL/ [[56]] GraphPatternNotTriples ::=
OptionalGraphPattern | GroupOrUnionGraphPattern | MinusGraphPattern | GraphGraphPattern |
RevisionGraphPattern | WindowsGraphPattern | ServiceGraphPattern [[60 a ]] RevisionGraphPattern ::=
'REVISION ' ( Var | Revision | String ) GroupGraphPattern [[60 b ]] WindowsGraphPattern ::=
'WINDOWS ' WindowRef GroupGraphPattern [[60 c ]] WindowRef ::= ( 'R ' Integer ? '/' ) ? WindowRelation [[60 d ]] WindowRelation ::=
Revision ( '/' ( Revision | XPathDuration )
( '/' ( XPathDuration | Integer ) )? )?
[[60 e ]] Revision ::= UUID | / HEAD (~[
Rule Interchange Format [
Each of the existing languages described above has drawbacks that make it unsatisfactory relative to the evaluation criteria of the business case. Instead, we consider a minor extension to SPARQL and the Graph Store protocol. Figure 1 describes the grammar extensions to support elementary dataset revisions, dataset interval revisions, and repeated interval datasets (aka windows of streams).
Figures 3 and 4 illustrates a query which combines static base data with a temporal clause - in this case in the \validity time" domain, with a clause for which the dataset constitutes a stream.
Figure 2 depicts how the generated solution propagation graph is amenable to a static temporality analysis, based upon which the process control is tailored to the dataset form.
In Figure 3, we show a query supporting the \connected patient" scenario
[
https://www.w3.org/TR/rif-rdf-owl/#RDF_ 10 RIF-RDF
Compatibility
James
Anderson et
al. the pro cessor, a data concentrator comp onent, to start the windowing op eration as so on as p ossible.
The text
PT05M/ sp eci es the duration of the temp oral interval of the time-based window function, while PT01M/ sp eci es the duration of the interval b etween start p oints of adjacent window functions.
written in the syntax of the prop osed SPARQL extension, constructs the content of the SMS message.
datatyp e xsd:dateTimeStamp, and no two elements in the stream have the same timestamp, giving a total order to the stream elements.
p ermits a unique interpretation of
recent window of the stream.
For simplicity, in the concatenation ab ove, the IRI of the RDF stream is assumed to not already have a query string. It is straight-forward to accommo date this p ossibility in
the query. 11
RDF
Distinct
Time-Series pro le http://streamreasoning.github. io/RSP- QL/Abstract%20Syntax%20and%20Semantics%20Document/ #distinct- time- series- profile
The proposed approach, to express temporal speci cations in a SPARQL dataset designator, has its origin in the unique architecture of Dydra12, Datagraph's SPARQL query service. This comprises two independent components, the RDF data store and the SPARQL query processor. These communicate through a limited interface, which allows the processor just to associate a speci c dataset revision with an access transaction as the target and to use statement patterns to perform count, match and scan operations on this target.
The following changes will su ce to support streams based on this architecture: { Extend the RDF store SPARQL Graph Store Protocol implementation to interpret intra-request state to indicate transaction boundaries. This can be HTTP chunking boundaries, RDF graph boundaries, or some other marker. Based on current import rates, the store should be able to accept 105 statements per second per stream with the e ective rate for sensor readings dependent on the number of statements per reading. { Extend the dataset designators to designate the dataset state which corresponds to some speci c set of transactions. The speci cation will permit combinations of temporal and cardinality constraints. { Extend the query processor control model from one which performs a single reduction of an algebra graph to one in which individual branches can be reiterated according to their temporal attributes, while unchanged intermediate results are cached and re-used. { Extend the response encoding logic to permit alternative stream packaging options, among them, repeated http header/content encoding and http chunking with and without boundary signi cance. 6
We have presented the grammar for a SPARQL extension and described a related Graph Store protocol extension that enables a sliding window operation on an RDF stream to be interpreted as a dynamic dataset. We illustrated the usage of this extension to describe a continuous query, with rules given in the SPARQL CONSTRUCT/WHERE form, on an RDF Stream in the distinct time-series pro le with a sample query and a data propagation diagram.
Future work includes generalizing the SPARQL and Graph Store protocol extensions to cover a larger class of RDF streams, and demonstrating the feasibility and performance of the extension with a prototype implementation.
This work has been partially supported by the \InnoPro le-Transfer Corporate Smart Content" project funded by the German Federal Ministry of Education 12 http://dydra.com and Research (BMBF) and the BMBF Innovation Initiative for the New German Lander - Entrepreneurial Regions. The authors wish to thank Davide Sottara of Arizona State University for assistance developing the connected patient query, and the reviewers for helpful suggestions. temporal literal, R a revision identi er and G a graph name. Temporality indicates whether temporal properties are speci ed through variables bound and constrained in a BGP and/or through a dataset clause.
XXXXXXXXX
Feature Proposal
Model
temporal RDF
[
O)
T )
BGP
TriplePattern ::= (Subject Predicate Object ) : [T]
R&W Base
[
((S
Datatset ::= FROM VersionIri
R34ples
C-SPARQL
CQELS
[
Window ::= ( RANGE Number TimeUnit ( ( STEP Number TimeUnit ) | TUMBLING ) ) | ( TRIPLES Number ) ((S ((S
P P
O) O)
T ) +
T )
GraphPatternNotTriples ::= ... | StreamGraphPattern
StreamGraphPattern ::= RANGE [ Window ] VarOrIRIReference TriplesTemplate
Window ::= ( RANGE Number TimeUnit ( SLIDE Number TimeUnit )? )
| ( TRIPLES Number ) | NOW | ALL
SPARQL Stream [
P
O)
T )
Datatset ::= FROM STREAM Iri Window ? Window ::= FROM NOW ( - Number TimeUnit )? TO NOW ( - Number TimeUnit )? ( STEP Number TimeUnit ) ?
"virtual", graph
R no, no, graph graph
R
T
yes
yes
yes
yes
no
yes
EP-SPARQL
[
P
O)
T
T ) BGP GraphPatternNotTriples ::= ... | SeqGraphPattern | EqualsGraphPattern | OptionalSeqGraphPattern | OptionalEqualsGraphPattern
NullOperator ::= ... | getDURATION | getENDTIME | getSTARTTIME
INSTANS
Verb ::= ... | :hasWindow
DIACHRON
[
O
G)
BGP O)
R) federated metadata
SourceClause ::= ... | ( FROM DATASET URI ( AT VERSION URI )? ) | ( FROM CHANGES URI ( BEFORE VERSION URI )? )
| ( AFTER VERSION URI ) | ( BETWEEN VERSIONS URI URI )
SourcePattern ::= ... | (DATASET URI ( AT VERSION URI )? ) | ( CHANGES URI ( BEFORE VERSION URI )? )
| ( AFTER VERSION URI ) | ( BETWEEN VERSIONS URI URI )
Dydra
[
O
G) RevisionGraphPattern ::= 'REVISION' ( Var | Revision | String ) GroupGraphPattern WindowsGraphPattern ::= 'WINDOWS' WindowRef GroupGraphPattern
BGP Dataset algebra reduction yes O)
R)
BGP O) O)
R) T )
BGP
P P P P P P P
Model algebra reduction rewrite, generate dataset, delegate rewrite, generate dataset, algebra algebra reduction rewrite (SQL), delegate rewrite (DSMS), delegate rewrite (PROLOG), delegate no yes no no no yes no no PREFIX fhir : <http :// hl7 / org / fhir /> PREFIX obs : <http :// hl7 / org / fhir / Observation .> PREFIX sct : <http :// snomed . info / id /> PREFIX prov : <http :// www . w3 . org / ns / prov #> PREFIX lr : <http :// localhost / local - records #> CONSTRUCT {
GRAPH ? elementID { ? obsID a <http :// hl7 / org / fhir / Observation >; obs : subject ? patient ; obs : effectiveDateTime ? heartRate_time ; obs : code <http :// localhost / local - records # AbnormallyHighHeartRate > ; obs : valueCodeableConcept [ fhir : CodeableConcept . text 'yes ' ] . } ? elementID prov : generatedAtTime ? generatedTime ; # this would likely be provided with the request as a protocol parameter lr : stream <urn : uuid : b4f166f6 -387c -11 e6 -83 de -180373 ae0412 > . } # determine heart rate exceeding some limit # based on history , activity and institutional factors WHERE { # this would likely be provided with the request as a protocol parameter BIND ( < patient / @pat2 > AS ? patient ) ? patient lr : isConnectedTo ? heartRateSensor . ? heartRateSensor lr : hasStreamID ? heartRateStreamID ;
a lr : HeartRateSensor . ? patient lr : isConnectedTo ? activitySensor . ? activitySensor lr : hasStreamID ? activityStreamID ;
a lr : ActivitySensor . } } ? patient fhir : Patient . birthDate ? birthdate .
BIND ( ( YEAR ( NOW ()) - ? birthdate ) AS ? age )
VALUES ( ? ageLevel ? ageMin ? ageMax ? ageFactor ) { ( lr : elderly 70 120 0.7 ) ( lr : midaged 30 70 1.0 ) ( lr : young 0 30 1.3 ) } FILTER ( ? age >= ? ageMin && ? age < ? ageMax ) VALUES (? activityLevel ? energyExpMin ? energyExpMax ? activityFactor ) { ( lr : high 25 100 2.0) ( lr : medium 11 25 1.0) ( lr : low 0 10 0.7 ) } FILTER ( ? maxEnergyExp >= ? energyExpMin * ? ageFactor &&
? maxEnergyExp < ? energyExpMax * ? ageFactor ) BIND ( ( (1 + ? patient_heartRate_std ) * ? activityFactor )
AS ? relHeartRate_delta_threshold ) BIND ( (? heartRate_base * (1 + ? relHeartRate_delta_threshold ))
AS ? heartRate_threshold ) # determine current pulse averages from sensors BIND ( IRI ( CONCAT ( STR (? heartRateStreamID ), '? WINDOWS =R/ HEAD / PT05M / PT30S '))
AS ? service_heartRate_ST )
SERVICE ? service_heartRate_ST { SELECT ( AVG (? heartRate ) AS ? heartRate_avgST ) # if the transaction timestamp , instead # ( THEN () AS ? heartRate_time ) ( MAX (? heartRateTime ) AS ? heartRate_time ) WHERE { ?g prov : generatedAtTime ? gen . { GRAPH ?g { [] a <http :// hl7 . org / fhir / Observation > ; obs : effectiveDateTime ? heartRateTime ; obs : code <http :// snomed . info / sct :36407505 > ; # heart rate obs : valueQuantity [ fhir : Quantity . value ? heartRate ;
fhir : Quantity . unit 'bpm ' ] . } } } } # test limit FILTER ( ? heartRate_avgST > ? heartRate_threshold ) # iff it gets this far , it will generate a result observation BIND ( UUID () AS ? obsID ) BIND ( UUID () AS ? elementID ) BIND ( NOW () AS ? generatedTime )
Fig. 4. Combined base-data, temporal and streaming query, continued