Stream query processing is becoming increasingly important as more time-oriented data is produced and analyzed nowadays. Examples for stream-based applications include sensor networks and infrastructure monitoring, electronic trading on Wall Street, or social networks. In the last decade several technologies have emerged to address the challenges of processing such high-volume and realtime data streams, which do not take the form of persistent database relations, but rather arrive in continuous, rapid and time-varying data objects. In this paper we focus on the new problem of preference-based query processing to analyze data streams. Preferences allow us to filter out only relevant information of the continuous data flow. In addition we provide a database approach that stores only the most important information in the database w.r.t. the user's preference. For this we introduce novel preference-based integrity constraints to keep the data right and consistent.
Many modern applications such as network monitoring, financial analysis, infrastructure
manufacturing, sensor networks, meteorological observations, or social networks require
query processing over data streams, e.g., [
Such stream data can be simple numbers or complex data, depending on the kind
of stream provider. Through collecting and analyzing this data new ways of doing
business emerge, since having such comprehensive data allows to act faster and better on
certain situations. However, several aspects of data management need to be considered
in the presence of data streams, offering a new research direction for the database
community, cp., [
In this paper, which is work in progress, we present an approach to analyze data
streams with the support of user preferences. The goal is to find the most interesting
information w.r.t. to one’s preference among the incredible number of streamed objects.
Preferences help us to separate the relevant data from the pointless information and
make it accessible to the user either as human readable data or by storing it in a database.
Another advantage of preference queries is that they never lead to an empty result. In
addition, we provide a language for connecting and querying streams in Preference SQL
[
As running example we use the online social networking service Twitter as a
realworld stream application. Twitter is a very fast medium of information dissemination.
The amount of tweets, that are posted every day is very large. That is 500 million short
messages daily according to data from the year 2015 [
The remainder of this paper is organized as follows: Section 2 highlights related work. Section 3 recapitulates essential concepts of the used preference model. In Section 4 we present the framework for analyzing data streams with the help of preferences, and in Section 5 we introduce a language for preference-based stream analysis. Finally, we conclude in Section 6. 2
Data streams can be found in every area of life: stock market, social networks, network monitoring, or meteorological observations, just to name a few. Hence, it is not amazing that many scientists all over the world try to process and to analyze these streams to extract important information from such continuous data flows.
Babu and Widom [
Since we refer to Twitter in our prototype implementation, we also present some
current work on this social network. In [
All previous work rely on analyzing the content of a stream or describe stream processing systems. In this paper we filter data by a preference based approach, which never leads to an empty result and only extracts the most important information w.r.t. the user’s preference. 3
Preference queries in database systems have been in focus for some time, leading to
diverse approaches, e.g., [
To specify a preference, the framework follows an inductive approach of base preference constructors that can be combined to complex preference constructors. Subsequently, we present some selected constructors which are often applied by users.
Preferences on single attributes are called base preferences. There are base preference
constructors for continuous (numerical), discrete (categorical), temporal, and spatial
domains. For example, the categorical preference POS(A; POS-set) expresses that a
user has a set of preferred values, the POS-set. The BETWEEN(A; [low; up]) preference
expresses the wish for a value between a lower and an upper bound. The LOWEST(A)
constructor and the HIGHEST(A) constructor prefer the minimum and maximum, resp.
Note that the dominance criterion of the base preferences is based on a SCORE function
+
f : dom(A) ! R0 , where f (v) describes how far the domain value v is away from the
optimal value. Dominated tuples have higher function values, i.e., x <P y () f (x) >
f (y). For more details we refer to [
Complex preferences determine the relative importance of preferences and
combine base or again complex preference constructors. Intuitively, people speak of “this
preference is more important to me than that one” or “these preferences are all equally
important”. In the Pareto preference P := P1 P2 = (A1 A2; <P ) all preferences are
of equal importance, while in a Prioritization preference P := P1 & P2 the preference
P1 = (A1; <P1 ) is more important than P2 = (A2; <P2 ). For a more formal definition
and more detailed information we refer to [
Example 1. Consider the UEFA European Championship which takes place in France in 2016. The tournaments take place in various cities (location). Since many people post interesting information about the tournaments on Twitter, we want to get all tweets which are posted by an author who stays in Lyon or Marseilles. This preference can be expressed by a POS constructor: P := POS(location; fLyon, Marseillesg). If there is no tweet posted in Lyon or Marseilles, the preference selection operator would find all tweets from other cities. 3.2
The Preference SQL query language is a declarative extension of SQL by strict partial order preferences, behaving like soft constraints under the BMO query model as described above. Syntactically, Preference SQL extends the SELECT statement of SQL by an optional PREFERRING clause, cp. Figure 1.
The keywords SELECT, FROM and WHERE are treated as the standard SQL keywords.
The PREFERRING clause specifies a preference which is evaluated after the WHERE
condition. If TOP-k is specified, only the k best tuples according to the preference order
are returned [
Example 2. The POS preference in Example 1 can be expressed in Preference SQL as follows, where Tournament is the database which contains information about the football competitions in France.
SELECT * FROM Tournament
In the above example IN expresses a POS preference. The keyword AND would state a Pareto preference, and PRIOR TO leads to a Prioritization (not shown in this example).
The current University prototype of the Preference SQL database system adopts a
(Java 1.7-based) middleware approach as depicted in Figure 2. For a seamless application
integration we have extended standard JDBC technology towards a Preference SQL /
JDBC package. This enables the client application to submit Preference SQL queries
through familiar SQL clients. The Preference SQL middleware parses the query and
performs preference query optimization. The preference selection operator implemented
by several evaluation algorithms computes the final result. This approach has proven its
flexibility and performance in various prototype applications conducted so far, see [
SQL Client
Preference SQL
JDBC
Preference SQL
Parser Query Optimization
Algorithms
Data
Our goal is to analyze data streams with Preference SQL, which needs well structured data. Hence, we need a framework that transforms the data from a stream into a processable format. For this goal we use Apache Flink, an open source platform for distributed stream and batch data processing1. The core of Apache Flink is a data flow engine created in Java and Scala. Flink includes several APIs2, e.g., for Data-Streams, DataSets, Tables, etc. These data streams can be created from various sources (e.g., message queues, socket streams, files) and converted to other (structured) formats.
For our prototype we decided to use Twitter as an example data source, because it has an open API and free accessible data (tweets). In addition, Apache Flink provides a Twitter Streaming API connector for direct connection to this social network. 4.1
When processing data streams with Preference SQL, two fundamental tasks must be carried out: 1. The data must have a Preference SQL processable format, because Preference SQL works on attribute based data, but streams are often encoded in various formats, e.g., as JSON3-objects as in Twitter. 1 https://flink.apache.org/ 2 Application Programming Interface 3 http://www.json.org/ 2. The result computation must be adapted to stream properties, since the dataflow is continuous. There is no “final” result after some data of the stream is processed. Hence, the result must be calculated and adjusted as soon as new data arrive.
Before we discuss these tasks in detail, we provide a brief overview of the whole system (see Figure 3). For more details we refer to the following sections. The incoming data stream is processed by Apache Flink, where our StreamProcessor transforms the data into a Preference SQL readable format. Our DataAccumulator builds chunks of objects, which can be processed by Preference SQL. Preference SQL evaluates the user preference on these data chunks and presents the result to the user. For this we suggest two possibilities: (1) we present the result to the user on his mobile phone, tablet or PC directly after the computation, or (2) we store it in a database for later usage. The first possibility is intended for users which are in front of their internet compatible device and want to find specific information rapidly. On the other hand we propose a method to store the result in a database for further processing to which we refer in Section 5.3.
Stream of objects
ETL Process in Apache Flink Data Stream
StreamProcessor
DataAccumulator
Stream of processed data
Stream of data chunks
Database Preference
SQL
Best matching
objects
Live output to cell phone, tablet or PC The individual objects delivered by a data stream are encoded and not compatible with Preference SQL, which works on attribute-based data. That means the data must be structured and needs an attribute based format like attributeName = attributeValue.
The transformation of the stream objects to a list of single attribute values occurs in the StreamProcessor, cp. Figure 3. For this one has to implement the mapping of the object in the stream to a table structured format. The data types of the attributes can be extracted from the stream objects, e.g., in the case of JSON-encoded tweets with the help of the Twitter API.
Example 3. The twitter messages (tweets) are JSON-encoded as shown in Figure 4. For example, this tweet was posted on June 22, 2016 at 12:49:31 UTC (created at:”Wed Jun 22 12:49:31 +0000 2016”) and has a specific identifier (id:745599599151353856). There are several different fields which describe a tweet in detail, e.g., name, description (the user-defined string describing the account), created at (date of account creation), followers count, status count (number of tweets of the user), lang (language the user speaks) and many more. Our StreamProcessor converts such objects into more structured data, e.g., into created_at = ‘‘Wed Jun 22 ...’’ and defines the data type “VARCHAR”. f” c r e a t e d a t ” : ”Wed Jun 22 12 : 4 9 : 3 1 +0000 2016 ” , ” i d ” :745599599151353856 , ” i d s t r ” : ” 745599599151353856 ” , ” t e x t ” : ” #SVK and #NIR have a l r e a d y q u a l i f i e d b u t who w i l l j o i n them ?nnnnFind o u t how t h e l a s t 16 i s s h a p i n g u p : h t t p s : / / t . co / Wu1kXiTt7L #EURO2016” , ” s o u r c e ” : ”<a h r e f =n” h t t p : / / t w i t t e r . comn” r e l =n” n o f o l l o w n”>T w i t t e r Web C l i e n t <n/a>” , ” t r u n c a t e d ” : f a l s e , ” i n r e p l y t o s t a t u s i d ” :745581100681019392 , ” i n r e p l y t o s t a t u s i d s t r ” : ” 745581100681019392 ” , ” i n r e p l y t o u s e r i d ” :1469402426 , ” i n r e p l y t o u s e r i d s t r ” : ” 1469402426 ” , ” i n r e p l y t o s c r e e n n a m e ” : ”UEFAEURO” , ” u s e r ” : f” i d ” :1469402426 , ” i d s t r ” : ” 1469402426 ” , ” name ” : ”UEFA EURO 2016 ” , ” s c r e e n n a m e ” : ”UEFAEURO” , ” l o c a t i o n ” : n u l l , ” u r l ” : ” h t t p : / / euro2016 . t i c k e t s . u e f a . com” , ” d e s c r i p t i o n ” : ” The o f f i c i a l home o f #EURO2016 on T w i t t e r . ” , . . . g
After the object preparation the endless stream must be split into finite parts to
be processed in Preference SQL. This is done by grouping objects into chunks. This
concept is similar to moving window, described in [
In the last step we analyze the data chunks within Preference SQL. The input is a stream of chunks, the output contains the most preferred objects w.r.t. the preference specified by a user. Preference SQL computes the result based on the current chunk provided by the ETL processor (Figure 3). But this temporary BMO-set is not the final result, because when new data arrives it may occur that the next chunk contains better objects w.r.t. the user preference. Hence, the current temporary BMO-set must be compared to the objects from the next chunks in the ETL process, and so on.
More detailed: Let c1; c2; ::: be the chunks provided by the ETL processor. The
Preference SQL system evaluates the user preference P on the first chunk, i.e., [P ](c1).
Since c2 could contain better objects we also have to compare the new objects from c2
with the current BMO-set, i.e., we compute [P ]( [P ](c1) [ c2), and so on. However,
this leads to a computational overhead if c2 is large. Therefore we apply a pre-filter to
c2, i.e., we first compute [P ](c2) and afterwards apply the preference selection to the
union of [P ](c1) and [P ](c2), since the following holds [
In this section we suggest a language for preference-based stream analysis. For this we first show how to connect streams within Preference SQL and afterwards we present a language for analyzing streams based on preference constructors. In addition, we present preference integrity constraints for table statements.
We want do develop an approach, that allows to analyze any data stream with Preference
SQL. For this, we have to connect the data stream to the Preference SQL system, in
particular to the Apache Flink engine as depicted in Figure 3. Note that there are also
other approaches to stream analysis and querying, e.g., [
Since we want to connect different streams, we present a general approach to create a stream connection within the Preference SQL system. For this we provide an interface which must be implemented for each individual stream. In our current University prototype we provide a TwitterStream connector and we plan to implement connections for stock market and other well-known stream provider. After implementing the stream connector, a user can connect to a stream by a CREATE STREAM statement. Figure 5 presents our syntax of our proposed language for stream creation.
CREATE
STREAM streamName
AS stream_connector stream_connector ::=
TwitterStream(consumerKey, consumerSecret, token, secret) StockStream(…)
…
The keywords CREATE STREAM introduce a stream connection with the user defined name streamName. TwitterStream corresponds to the name of the pre-imple-mented Twitter connector which gets the user-specific login credentials.
Example 4. Figure 6 shows an example for the connection statement for Twitter. The cryptic character sequences represent the Twitter account login data.
(’NQEk0KbszVbaAcjCWLksbodkN’, ’HDKxlp2REOHvuq59oKrZZsdFovItwG6upOGJuSN4btr6npp2c3’, ’2400192752-DQSedtepr68SerQVyjHLzpHhMitcwJQfbvwxnLi’, ’BAk0krCYq77W4p45UwyuAuNnpR3nrv9WofO9PNL46YFch’);
After the connection to a stream the data can be queried with Preference SQL. For this we provide a preference-based stream language, whose syntax is shown in Figure 7.
After SELECT STREAM one can specify a list of columns column_list, where each column corresponds to a field in a stream object. streamName corresponds to the user defined name given in the CREATE STREAM statement (cp. Section 5.1). The kind and the quantity of the chunks, that are build from the endless stream (cp. Section 4.2), can be defined by the user. Thereby, the chunks can be size or time based. When specifying a time (in seconds), the DataAccumulator in the ETL process (cp. Figure 3) builds chunks based on the data retrieved within the given time slot. By specifying a size the ETL processor groups size chunks together. PREFERRING introduces the preference condition which will be evaluated on the data stream.
SELECT
STREAM column_list
FROM streamName CHUNK
SIZE TIME = number
PREFERRING preference_condition
Example 5. Let us imagine that a user wants to read the tweets about the semifinals in the European Football Championship. The desired hashtag is #EURO2016 or #em2016 and the location should be the same as in Example 2. In addition, equally important (Pareto preference) to the previous preferences is the author of a tweet, in this example UEFA Euro 2016 or EM 2016 Frankreich. The chunks should be constructed time based, here 60 seconds. The corresponding preference stream query is shown in Figure 8.
SELECT STREAM *
CHUNK TIME = 60
5.3
In this section we suggest an extension of SQL constraints [
In this sense, the idea is to define preference based integrity constraints which apply whenever a tuple should be inserted into the table. If the new tuple passes the preference condition it will be stored in the table. Otherwise, if it is dominated by any other tuple already in the table, the new tuple can be discarded or moved to another table which holds dominated objects. Note that when considering streams, the current content of a database table does not necessarily represent the perfect matches, because they can be dominated by better candidates arriving later in the stream. Therefore it is important to specify what happens with these dominated tuples. Figure 9 shows the syntax diagram which describes our preference integrity constraints.
TABLE CHECK cp. SQL-92 … cp. SQL-92
… WITH
DOMINATED PREFERRING ( preference_condition
) DELETE PUSH
INTO
The preference constraint follows the standard CREATE TABLE syntax [
Holding already dominating tuples could be useful, e.g., if one wants to get the
Top-k elements of a preference query and the result w.r.t. the preference does not provide
enough objects (cp. [
Example 6. Reconsider Example 5, but now we want to specify a table constraint instead of analyzing the streams directly within Preference SQL. Figure 10 shows the CREATE
Note that constructing a cascade of worse objects means to specify preference based integrity constraints for the table worseTweetsTable and move the dominated objects to a third relation, and so on. 6
In this paper we proposed a preference based approach for stream analysis. We suggested a language for connecting and querying streams in Preference SQL. In addition, we presented the possibility to specify integrity constraints based on preference constructors. We have implemented our approach based on Twitter data, because of its open API. However, the same approach can be used for other data streams, such as stock exchanges or Facebook posts.
Our work is in progress and thus, we plan to implement further stream connectors and to extend the preference stream query language. A comprehensive experimental evaluation as well as a comparison to other existing stream analysis frameworks is also future work.