DEIM Forum 2014 D3-2

An Efficient Execution Scheme for Designated Event-based Stream Processing
Yan Wang†

andHiroyuki Kitagawa‡†Graduate School of Systems and Information Engineering, University of Tsukuba‡Faculty of Engineering, Information and Systems, University of Tsukuba
1-1-1 Tennodai, Tsukuba 305-8573, Japan
E-mail: †wangyan@kde.cs.tsukuba.ac.jp,‡kitagawa@ kde.cs.tsukuba.ac.jp

Abstract With the increase of streaming information sources, stream processing has been an important research issue. In
this paper, we introduce the designated event-based stream processing scheme in the stream processing engine. Different fromthe traditional stream processing scheme, the query results are supposed to be generated only when tuples come fromdesignated streams specified by the user. After introducing the proposed scheme, we consider its implementation. We discuss
some important implementation internals of the stream processing engine and show an efficient designated event-based
execution scheme. We also consider multi-queries which is a common situation in the stream processing engine. The
experiment shows the advantages of the proposed execution scheme.
Keyword Stream processing engine，Designated event-based processing

1. Introduction

processing. In this paper, we introduce the designatedRecent years, the amount of unbounded generated data is

event-based processing scheme in the stream processing

growing rapidly like network packets, stock trades andengine and propose an efficient execution scheme for it.

sensor data. Such unbounded generated data is called

This paper is organized as follows. In section 2, we talkstreams. Many stream processing engines have been

about related works. In section 3, we show the traditional

developed to deal with the streams like STREAM [1] , S4 [ 2] ,stream processing model. Section 4 describes designatedBorealis[ 3], D-stream[ 4]

and Storm[ 5]. Users can register

event-based

processing

scheme.

Some

basic

continuous queries on these systems and a continuous

implementation internals are explained in section 5. A

query is often involved with multiple stream sources. The

naive execution scheme is explained in section 6 and a

traditional stream processing scheme is that whenever data

smart execution scheme is proposed in section 7. How tocomes from any stream sources, relevant continuous

deal with multiple queries is explained in section 8.

queries are supposed to be evaluated and the query results

Section 9 shows the experimental evaluation and section

would be generated. However, such a processing scheme is

10 gives the future work and conclusion.

not always what users want. There is another kind of
requirement - sometimes, the result is supposed to be

2. Related Works

generated onl y when data comes from particular streams

STREAM [ 1] is a traditional stream processing engine. It

specified by the user. For example, suppose there are two

accepts CQL [ 6] queries and translates a quer y into an

stream sources: one is Twitter stream, and the other is

execution plan containing leaf operators to get tuples from

News stream. Users may want to get the related Tweets in

input streams. No matter which leaf operator accepts an

30 minutes on the arrival of a piece of news. Then the

input tuple, the whole executi on plan tree is executed andusers are able to analyze the query result s to see wanta query result is generated. It does not support thepeople are thinking about the latest news. In this case, thedesignated event-based processing.

query results are to be triggered by the News stream only

Discretizedstream [ 4]isadistributed

real

time

on the arrival of a piece of news. On the arrival of a Tweet,

computation system. It processes streams in small time

the query is not supposed to be evaluated and no queryintervals. The input tuples in each interval a re processed

result would be generated. We call such a processing

in a batch and then a query result is generated. However, it

scheme as designated event-based stream processing

does not support the d esignated event -based processing.

scheme which implies the query results are triggered by

Storm [ 5] is also a distributed real time computation

the designated streams. The traditional stream processing

system. Topology can be registered in the system. The

engine dose not support the designated event -based

topology contains spout nodes getting input tuples from

stream sources. Whenever a spout node gets input tuples,

supposed to be executed whenever new tuples arrive at thethe whole topology is evaluated and a quer y result issystem. It is eas y to know that whenever a new tuple

generated.

arrives at the system, the window operator responsible for
accepting it would generate a new relation which possibly
has a large overlapping with the previous one. It is the

3. Traditional Stream Processing Model
streamsame with the relation -to-relation operators. It is not aprocessing, we introduce the traditional stream processing

wise way to always output the query results which have amodel first.

large overlapping. Usually, we

Beforeexplaining

designatedevent -based

want to output the

How to perform queries on streams which contains

difference between the current output relation and theunbounded data? We are familiar with the relational

previous output one and that is what relation-to-stream

databases which contain a finite number of tuples in each

operators do. An Istream operator always outputs the newtable. Some relational operators, su ch as join andtuples in the current output relation.

aggregation, must work on the finite number of tuples.
When performing the relational operators on a stream, the4. Designated Event-based Stream Processing

stream must first be ‘cut’ into a relation containing a finite
number of tuples, then the relational operators can workon the relation. In the stream processing engine, window
operators are used to generate a relation from a stream.
Type
Name
Stream-to-relation
Tuple-based window
Time-based window
Relation-to-relation Selection
Projection
Join
Groupb y_aggregation
Relation-to-stream
Istream
Rstream
Dstream
Table 1 Operator

Figure 1 Query Example
In this section, we explain the designated event -based
stream processing scheme. We first define a discrete,ordered time domain T. A general time instant t is any
value from T. In the traditional stream processing engine,a tuple may arrive at the system at any general time instant.
The query result may be generated at any general time
instant. As for our designated stream processing, the query
is supposed to be evaluated onl y when tuples come fromparticular streams specified by the user. We call themThree classes of operators are defined over streams andrelations.

They

relation-to-relationarestream-to-relation

operators

andoperators,

relation -to-stream

operators. A stream-to-relation operator takes a stream as
an input and produces a relation as an output. A
relation-to-relation operator takes a relation as an input
and produces a relation as an output. A relation-to-stream
operator takes a relation as an input and produces a stream
as an output.
Some basic operators are shown in Table 1.

A

tuple-based window operator is a stream -to-relation

master streams. The timestamps of tuples coming frommaster streams form a discrete, ordered time domain T’⊆T, and a master time instant t’ is any value from T’. The
query is supposed to be evaluated and generate query
result only at t’.
Now we introduce the designated event -based strea mprocessing scheme. In a query plan tree, the innermost
operators

arestream-to-relation

outermostoperators

isaoperators

relation -to-stream

and

the

operatorproducing a stream. The query result is supposed to begenerated only when a new tuple comes from a master

operator and it always has a window size, then it generates

stream at the master time instant t’. We denote the result

a relation containing the number of tuples equal to theof the query at t’ as S(t’). We emphasize that thewindow size. For example, a tuple-based window operator
with a window size of 2 0 always generates a relation
containing the latest 20 tuples. After generating relations
from streams, the relation -to-relation operators can takesuch relations as inputs and perform relational logics suchas selection, projection, join and so on. The outputs ofthese operators are also relations.
In the traditional stream processing engine, a query is

traditional stream processing scheme is a special case of
our scheme. Just specify all streams as master streams,
then the results of our scheme are the same as the
traditional one.
We give an example of the proposed processing scheme.
Let us consider the query in Figure 1. This query intents
for a join operation on the latest two tuples from Stream1

Figure 2 Example input

Figure 4 Quer y plan tree

Figure 3 Quer y result
and the latest two tuples from Stream2 on the x attribute .

Figure 5 Window operator

Stream1 is a master stream. The example input tuples areoperators output the update of the relation instead of theshown in Figure 2. The tuples arrive at the stream

whole relation. Some operators like join, aggregation andprocessing engine in a timestamp order from stream1 andwindow operators, need to know the whole relation for

stream2. Each tuple coming from stream1 contains an x

execution.

attribute and a y attribute. Each t uple coming fromthemselves to save the current state. For example, astream2 contains just an x attribute. Because str eam2 is a

row-based window operator with a window size of 2

master stream and tuples arrive at the stream processing

contains a synopsis always saving the latest 2 tuples.

engine at time instants 2 and 6, the output tuples are

In order to make the behaviour of operator s and synopsis

generated at time instants 2 and 6 as shown in the Figure

more clearly, we give an example. The query in Figure 1 is

3.

translated into a query plan tree in Figure 4. We use the

These

operators

maintainsynopsesin

input tuples in Figure 2. On the arrival of the tuple at time

5. Incremental Execution Scheme

instant 4 from Stream1, the internal state of the window

Before discussing the implementation of the designatedoperator w1 changes as what we show in Figure 5. Its

stream processing scheme, we look into some important

window size is 2 and the window s ynopsis always saves

implementation internals of the stream processing engine.

the latest 2 tuples. On the arrival of the tuple <4,4,6,1>,

A window operator generates a relation from a stream.

this tuple is inserted into the window synopsis, the number

Whenever a new tuple comes to the window operator fromof tuples in the synopsis exceeds the window size , and thea stream, the output relation of the wi ndow operator would

oldest one should be deleted from the synopsis. Then the

be changed - the new coming tuple is inserted in the

output of the window operator are tuple <4,1,-,5,2> andrelation and some tuples may go outside of the window

tuple <4,4,+,6,1>. The status of the window operator after

and are deleted from the relation. As for implementation,

the procssing is shown in Figure 5 on the right.

itisinefficient

for

the

window

operatororrelation each time. Since a relation is changing in a

6. Naive execution scheme
6.1. Execution modeltimestamp order including newly arriving tuples andWeomitted obsolete tuples, it is more efficient to do

implementing the designated event -based processing. The

incremental computation. We append each tuple with abasic idea is very simple – everything works the same withplus tag or a minus tag as well as a tuple identifier. A plus

the traditional stream processing engine except for the

tag represents for a newly arriving tuple and a minus tag

relation-to-stream operator which is the outer mostrepresents for an obsolete tuple. The tuple identifier isoperator in the query plan tree. Our relation-to-stream

used to uniquely identify a tuple. Now an element is

operators generate output only on the arrival of tuples

represented by <t,id,T,s> consisting of a timestamp,

originated by the master streams. The whole query plan

identifier, tag and tuple.

tree is evaluated on the arrival of tuples coming from any

relation-to-relation operators to work on the whole

Now stream-to-relation and relation -to-relation

firstpresent

stream source.

a

naive

execution

schemeforshown in Figure 7 on the left. The synopsis is empty and6.2. Master markthe input tuple is a plus tagged tuple with a true masterIn order to know which tuples are triggered by mastermark. Then this tuple is inserted into the synopsis first andstreams, each tuple is given a master mark to tell whether

then all the tuples in the synopsis is output and thethis tuple is originated b y a master stream or not. The

synopsis is cleaned up as shown in the Figure 7 on thevalue of master mark is true or false. If a tuple has a true

right. We can see the tuples at time instant 2 and 6 havemaster mark, it means this tuple is originated by mastertrue master marks and the Istream operator generates

streams. Now an element is represented by <t,id,T,M,s>output tuples when receiving them. The query result iswhere M represents for a master mark.

exactly what we show in Figure 5.

Then we can know whether a tuple is originated by amaster stream by looking into the master mark. But how to

6.4. Shortcoming

make sure all tuples originated by master streams have

We can see that the whole query is evaluated on the arrival

true master marks? The approach is ver y simple. At first,of tuples coming from any stream sources. On the arrival

we give a true master mark to each tuple coming fromof a tuple coming from a master stream, some tuples frommaster streams and false master marks to other tuples.

non-master streams ma y have gone outside the given

Then when an operator accepts an input tuple, all output

window size and obsolete. They do not contribute to any

tuples generated based on the input tuple are given thequery result, but they are processed by the whole query

same master mark and timestamp as the input tuple.

plan tree and many useless intermediate tuples may begenerated.
For example, the tuple at time instant 3 from stream 1 in

6.3. Istream Operator

the Figure 3 does not contribute to any query result.
However, such non-master stream tuples are processed by
the whole query and generate many useless intermediate

Figure 6 Istream Operator Input Sequencetuples because they are pushed to the following operators
as soon as they comes to the window operator. Instead ofthat, we can change the behavior of the window operator
to delete useless tuples from the beginning and reduce theprocessing cost. We call it smart approach.

7. Smart execution scheme
Figure 7 Istream Operator Execution Example

7.1. Execution modelNow we introduce our Istream operator whose behavior is

One query has two execution modes, waiting mode andchanged. Our Istream operator always outputs the new

eager mode. At first, it is in the waiting mode and s ystemtuples triggered by arrivals of tuples with master marks .

gets input tuples from the information sources. If a tuple

So it should maintain a synopsis saving the input relation.

comes from the non-master stream, the query remains in

When an Istream operator accepts an in put tuple, we first

the waiting mode. If a tuple comes from a master stream,

see whether it is a plus or minus tagged tuple. If it is athe query’s state changes to the eager mode and allplus tagged tuple, the tuple would be inserted into theoperators are triggered.

synopsis,

operators, the query’s state comes back to the waiting

and if it

is

a

minus

tagged tuple,

thecorresponding tuple is deleted from the s ynopsis. Next, it

After the execution of allmode.

sees whether it is a true master mark tuple or a false

The behavior of the window operator is changed. We

master mark tuple. If it is a true master mark tuple, all of

introduce two kinds of window operators: one is masterthe tuples in the synopsis are output as the query results

window operator, which is for master streams, and theand the synopsis is cleaned up. If it is a false master markother one is smart window operator, which is fortuple, it does nothing. With the input tuples shown in

non-master streams. The behavior of master window

Figure 3, the input tuple sequence of the Istream operator

operators remains the same as we describe d in Section 4.3,

is shown in Figure 6. The internal state of the Istream

while the behavior of smart window operators is different.

operator on the arrival of the tuple at time instant 6 is

7.2. Smart Window Operator

in Figure 9. Because less intermediate tuples are generated,
we can see its length is shorter than the sequence in Figure
6. It generates the same results.

7.3. Analysis
On the arrival of tuples from non -master streams, thecoming tuples are buffered in the smart window operators.
If the number of the buffered tuples exceeds the window
size, the overflowing tuples are just thrown away. Suchtuples are not processed by the whole query plan tree, thus
Figure 8 Smart Window Operator

the processing cost will be reduced.

8. Multiple query
Figure 9 Istream Operator Input Sequence

There are always more than one query registered in a

When the query is in the waiting mode, a smart window

stream processing engine. We think about how to support

operator accepts all input tuples. Then, it buffers thedesignated event -based processing dealing with multiple

tuples in the window synopsis and does not output them.

queries.

When the query is in the eager mode, a smart window

multiple queries, a direct thought is sharing the window

operator outputs all buffered tuples.

operators for these common streams. However, though a

The synopsis of the smart window operator is divided

stream

If somemaybecommon streams

involvedinare involved in

multiplequeries,theinto two parts to specify which tuples are buffered. One iscorresponding window operator may behaves differently

output part saving the tuples that have been output and theforother is non-output part saving the tuples that have nottime-based, master or smart. Even for the tuple -based

been output. In the waiting mode, the incoming tuples arewindow operator, the window sizes ma y also be different.

inserted in the non -output part. In the eager mode, theFurthermore, different queries have different execution

tuples in the non-output part are output and move to themodes. A smart window operator acts differently in the

output part.

eager mode and waiting mode. How to deal with thedifferent

queries.

For

example,tuple-basedorWhen a tuple is inserted in the smart window operator

situation that a shared window operator runs in the eager

in the waiting mode, the number of tuples in the synopsis

mode for one query and in the waiting mode for anothermay exceeds the window size and the oldest tuple should

query?

be deleted from the s ynopsis. If the oldest tuple is in the

Because a shared-window operator works differently

output part, it means this tuple has been process ed by thedepending on the query, it is better to abstract a virtual

whole query plan tree and the window operator should

window operator and let different queries register their

output the corresponding minus tuple. If t he oldest tuple isown strategies in it. The strategy contains (window type,in the non-output part, it means this tuple has not been

window size, output queue, quer y id) . Window type’s

processed by the whole plan tree and it can be deleted

value ma y be tuple smart, tuple master, time smart or timefrom the synopsis dire ctly without generating any plus or

master. The memory space of the s ynopsis belonging tominus tuples.

the virtual window operator is shared among multiple

Again we look back to our example of input tuples inqueries. The overlapping tuples belonging to different

Figure 2. The tuple arrives at time instant 5 from stream 1.

queries are also shared in the synopsis. During the

The state of the smart window operator is shown in theexecution of a virtual window operator, it executes all

Figure 8. Two tuples have already saved in the non-output

registered strategies. It can get the execution mode bypart of the window synopsis. When the coming tuple isquery id, execute its logic, and push output tuples to theinserted in the synopsis, the oldest tuple should be del eted.

corresponding output queue. Different strategies also

Because the oldest tuple which has a tuple identifier of 4

should maintain their own pointers to save the position

is in the non-output part, it can be deleted directly. Then

information of tuples in the s ynopsis.

this tuple is only processed by the smart window operator
instead of the whole query plan tree in the naive approach.
The input tuple sequence of the Istream operator is shownWe show experimental results using two streams :stream1 is not a master stream and stream2 is a masterstream. The query is the one presented in Figure 1. The
incoming rate of stream1 is 5 thousand tuple/s. We
Figure 10 Multi-query example

changed the incoming rate of stream2 from 1 thousand
tuple/s to 5 thousand tuple/s. We executed the query for 5
minutes and observed the time spent on processing thequery as well as the total number of tuples processed by
the whole query plan tree for both t he naive approach andthe smart approach.

Figure 11 Query plan tree for multi -quer y

The time spent is shown in Figure 12 and the number of

We give an example of multi -queries in Figure 10 whichtuples processed is shown in Figure 13. We can see whenis translated into a query plan tree in Figure 11. As

the input rate of master stream2 is smaller than stream1 ,window operator w1 is shared by multiple queries, i t is a

the time of the smart approach is less than the naive

virtual window operator and is registered with strategies

approach because many intermediate useless tuples are not(row master, 50, q1, Query1) and (row smart, 100, q2,

generated. When we increase the input rate of master

Query2). On the arrival of a tuple from stream1, query1 isstream2, it approaches to the naive one because lessin the eager mode, while query2 is in the waiting mode.

buffered tuples are deleted from the window operator

During the execution of the window operator w1, it

directly.

processes its processing logic for each strategy andgenerates the output tuples. We can see that the memory

10. Conclusion and Future Workspace of the latest 50 tuples are shared between query1

Weand query2 thus the memory cost is reduced.

processing and proposed its efficient execution scheme.
We

9. Experiment

have
have

proposeddeveloped

designatedastreamevent -based
processing

streamengin eimplementing the proposed execution scheme, and shown
its advantages by experiments.
Future research issues include more elaborated analysisof the proposal and the parallel execution of multiple
queries in the designated event -based stream processing.

Reference[1] A. Arasu, et al. STREAM: The Stanford Data Stream
Management System, 2004.
[2] L. Neume yer, et al. S4: distributed stream computing
platform. In KDCloud, 2010.
[3] D. J. Abadi, et al. The design of the Borealis stream
processing engine. In Proceedings of the Conferenceon Innovative Data Systems Research, CIDR, 2005.
[4] M. Zaharia, et al. Resilient Distributed Datasets: A
fault-tolerant abstraction for in -memory clustercomputing. In 9 th USENIX NSDI, Apr. 2012.
[5] http://stor m-project.net/[6] A. Arasu, et al. CQL: A Language for Continuous
Queries over Streams and Relations. In Proc. of theNinth Intl. Conf. on Database Programmin g
Languages, September 2003.

Figure 12 Query evaluation time

Figure 13 Processed tuple number
We have implemented both the naive and the smart
execution
processing.

schemesofthedesignatedevent -based