DEIM Forum 2019 F7-5

CBiX: Incremental Sliding-Window Aggregation For Real-Time
Analytics Over Out-of-Order Data Streams
Savong BOU† , Hiroyuki KITAGAWA† , and Toshiyuki AMAGASA†† Center for Computational Sciences, University of Tsukuba
1-1-1 Tennodai, Tsukuba, Ibaraki 305-8577 Japan
E-mail: savong.bou@kde.cs.tsukuba.ac.jp, {kitagawa,amagasa}@cs.tsukuba.ac.jp
Abstract Stream processing has been extensively applied in wide varieties of fields. In the meantime, stream
aggregating is a popular processing technique of big data. However, it suﬀers from serious setbacks when the ordersof events (e.g., stream elements) occurring are diﬀerent from the orders of events arriving to the systems. We
call such data streams as “non-FIFO steams”. This phenomenon usually happens in distributed environment due
to many factors, such as network disruption, delay, etc. In many analyzing scenarios, eﬃcient processing of such
non-FIFO streams is strongly needed to meet various data processing requirements. This paper proposes an eﬃcient scalable checkpoint-based bidirectional indexing approach, called CBiX, for real-time analysis over non-FIFO
streams. CBiX maintains the partial aggregating results in an on-demand manner per checkpoint. We demonstratethat CBiX requires less time and memory complexity than the state-of-the-art approach. We plan to do extensive
experiments to compare the performance of the proposed approach and the the state-of-the-art approach once wefinish the implementations.
Key words Non-FIFO streams, Aggregation, Sliding Window, Incremental Computation

1.

Introduction

have been proposed. Those algorithms shows promising performance when dealing with the data streams where the or-

The rapid growth of modern technological devices andder of events occurring are the same as the orders of events

SNSs increases the flow of the unbounded transmissions ofarriving to the systems. We call such data streams as “FIFO

record continuously in real time. These unbounded records

streams”. However, in real applications, the arrival orders

are called data streams. Stream processing has been receiv-of data streams are not always ideal. There are many cases

ing a lot of attentions in both academic and industrial fields.

where the occurring and arriving orders of streams are diﬀer-

To date, many data stream management systems (DSMS)

ent, which is called “non-FIFO steams”. This phenomenon

have been developed to process data in various applications

usually happens in distributed environment due to many fac-such as health care, disaster prevention, telecommunication,

tors, such as network disruption, delay, etc.

etc.

Let us consider an example where a Youtuber wants to

To deal with an infinite sequence of streaming records [1–5]arrange an eﬃcient online and oﬄine advertising strategy.

using finite memory (or computational resources), a window

Advertisers/Youtubers want to know the viewing frequen-is recognized as one of the most practical ways to extract

cy/duration or contents/ads of their videos, the viewers’ de-partial data from an infinite stream. Specifically, given an

mographic groups, and the charged/paid advertising fees.

operation over a sequence of records, a window limits the

These aggregating results are needed in real time so that

evaluation scope. Additionally, it is common to slide a win-they can adjust budgets, change campaign strategies, anddow over the stream within a range to apply the operation

plan future directions as soon as possible. Since the Internet

many times. This practice is called a sliding window.

has so greatly expanded within reach worldwide, non-FIFO

It is important to extract abstract or higher-levels of

steams are inevitable. In such scenario, the money is in-knowledge from the raw data. One common technique is ag-

volved, so the aggregation results (e.g., billing fee) should be

gregation. Often aggregation is combined with sliding win-computed as soon as possible. Such unbounded, out-of-order,

dows to extract useful information from streams, which is

and global-scale data streams have become increasingly com-called an aggregate continuous query (ACQ) [6].

mon in daily life, such as Web logs, mobile usage statistics,

It has been extensively studied, and a lot of algorithms

sensor networks, etc.

Flink [7] and Cloud Data Flow [8] take an initiative toaddress this challenging problem. They propose a general
stream processing framework by introducing a flexible setting option to instruct the system how much delay it should
expect for the late-arriving records from streams. For thisFigure 1: An example of non-FIFO stream aggregating given

purpose, they set the timeframe, called watermark, to in-a query with the window size of 18 seconds, the slide size ofstruct the system to recompute the results of the past win-

2 seconds, and Sum aggregation. For simplicity, each record

dows within the watermark timeframe if there are late-arrival

represents both value and timestamp, and the watermark istuples on them.

set to 6 seconds.

An example of these approaches is shown in Fig. 1. Assume
that the current time is at the 22nd second. After next two

2.

Preliminaries

arrival record 19 and two non-late-arrival records 23 and 24

2.1.

Aggregate Continuous Query (ACQ)

arrive to the system. Since the late-arrival record 19 falls

Aggregate Continuous Query (ACQ) consists of three main

into two past windows as shown in the figure, the results ofcomponents: aggregation operation, window, and slide. Ag-

the aﬀected past windows are recomputed. Both Flink [7]gregation operation indicates the necessary kind of compu-

and Cloud Dataflow [8] are the general stream processing

tation. Window denotes the bounded streams on which theframework. The default implementation is very naive andaggregation operation is evaluated. Finally, slide instructs

ineﬃcient. Their main problem is they compute the results

when to execute ACQ. The basic syntax of ACQ is shownof the aﬀected past windows and the current window frombelow:

seconds (at the 24th second), the window slides. One late-scratch non-incrementally every time the window slides.
SELECT A g g r e g a t i o n

Recomputing the results of the aﬀected past windows is

FROM Stream

trivial because the updated results can be obtained by ag-

WINDOW window SLIDE s l i d egregating the existing results with the late-arrival records.
However, computing the result of the current window is chal-

Aggregation represents the aggregation operation, andlenging. Therefore, this paper only focuses on computing theStream represents the data stream.

aggregating result of the current window.

SLIDE represent a window and a slide, respectively. In

W IN DOW and

To the best of authors’ knowledge, there is only one algo-

ACQ, the window and the slide can be expressed by ei-

rithm that can compute the results of the current window

ther the number of records or the time interval. Non-FIFO

incrementally though its performance is very poor. Flat-streams do not happen when the window and slide are ex-

FAT [9] uses binary tree to keep all intermediate results topressed as the number of records, so the window and slideachieve incremental computation over both FIFO and non-are both expressed as time interval throughout this paper.

FIFO streams. If n represents the number of partitions that

Note that ACQ also supports other properties such as

can be derived from the given window W and slide S, and p

W HERE and ORDER BY clauses. Since this work fo-

represents the aﬀected partitions by the late-arrival records,

cuses on sliding-window aggregations, the other properties

the computation complexity of FlatFAT is p ∗ log(n). Thenproposed approach only needs p1 ∗ log( ) + 3 ∗ p2 time comk
plexity (p1 + p2 = p), where k is the number of checkpointsnand∈ [S, n]. The performance improvement is significantkfor both time and space complexities.

are ignored for simplicity.

This paper aims at addressing the above bottleneck byintroducing an eﬃcient approach that can handle the cur-

2.2.

Properties of Aggregation Operations

Aggregation operations are used to summarize the valuesof multiple records. Although there are a lot of aggregation
operations [10], they can be categorized into three groups:•

Distributive aggregation: Suppose S is a sequence ofvalues and x is a new value. Aggregation f () is a distribu-

rent window computation incrementally. The proposed ap-

tive aggregation if f (S.x) = f (f (S).x) where S.x(f (S).x) is

proach use a novel checkpoint-based bidirectional index,

a sequence concatenating S(f (S)) and x. The distributive

called CBiX, to eﬃciently maintain all intermediate results.

operation includes count, sum, min, max, product, sum ofSince the implementation has not finished yet, we plan tosquares, etc.

do extensive experimental evaluations to show the eﬀective-•

Algebraic aggregation: Algebraic aggregation is a com-ness of CBiX over the state-of-the-art approach when dealing

bination of two or more diﬀerent distributive aggregations.

with non-FIFO streams.

For example, the average of a sequence of values can be calculated by combining the sum and the count. Algebraic aggre-gations include the average, standard deviation, geometricgroups: (1) FIFO algorithms [6, 9, 14], and (2) Non-FIFO

mean, range, etc.

algorithms [7–9].

•

Holistic aggregation: All other types of aggregations

3.2. 1.

FIFO algorithms

are included into this category.
This paper focuses on distributive and algebraic aggrega-

There are several algorithms that can achieve O(1) time com-

tions. Holistic aggregation is beyond the scope of this study.

plexity. They can be categorized into: (1) Worst-O(1) algorithms, and (2) Average-O(1) algorithms. Worst-O(1) algo-

3.

Related Workrithm is referred to all algorithm that required at most O(1)time complexity to compute the aggregation when the window slides. Average-O(1) algorithm requires in average O(1)time complexity, but at some slides, it requires at worst O(n)time complexity where n is the total number of slides in thewindow.
DABA [14] and FOA&IOA [14] are the worst-O(1) algo-

Figure 2: An example of Cutty for the ACQ over datarithms. FlatFIT [6] and Two-Stacks [14] are the average-

streams in Fig. 1.

O(1) algorithms. Two-Stacks [14] is the state-of-the-art approach. It uses two stacks to maintain the aggregating val-This section reviews related works of incremental sliding-ues from the previous computations to achieve O(1) time

window aggregations over data streams. Previous studies

complexity on average. Each stack’s element consists of two

can be categorized into two groups: (1) partition aggrega-value: (1) val: is the left aggregating value for one parti-

tion and (2) final aggregation. For simplicity, we use Sumtion, and (2) agg: is the right aggregating value for one orfor example, but the discussion applies to other distributive

more partitions. Maximum size of each stack is equal toand algebraic aggregations as well. In addition, each stream

the number of partitions in the window. When the window

record in this paper represents the value and timestamp, andslides, one element is pushed into stack1, and top element

we only focus on computing the result of the current window.

of stack2 is popped oﬀ. The val of the pushed element is

3.1.

Partition Aggregation

the aggregating value of the coming records, and the agg is

A partition aggregation divides a window into partitions

computed by aggregating the agg of the top element withbased on the given window size and slide size to calculate thethe val of the coming records. The query result is computed

aggregation for the entire window by sub-aggregations forby aggregating the agg values from the top elements of bothpartitions. Such partitioning is important for two reasons.

stacks. When the stack2 is empty, all elements from stack1

First, the eﬀects of all expired records can be eliminated si-

are popped oﬀ and pushed into stack2 one after another.

multaneously without checking each individual record when

Fig. 3 is an example for this approach. Since the win-

Second, the sub-aggregation for eachdow size is 18 seconds, and the slide size is 2 seconds, thepartition, called p − value, can be reused to calculate thenumber of partitions is nine. Fig. 3a shows its status whenquery results of ACQ. Consequently, both CPU and mem-

maintaining all records in the current window. In stack1,

ory cost can be conserved. The following techniques are typ-agg values are computed by aggregating the agg values from

ically used to partition all records in a window: Panes [11],the bottom to the top. Since the size of stack1 is same as

Pairs [12], and Cutty [13].

the number of partitions in the window, which is nine andthe window slides.

Cutty [13] is the most eﬃcient approach for this purpose.

stack2 is empty, all elements from stack1 are popped oﬀ andThis approach creates a new partition only when the window

pushed into stack2 by aggregating the agg value from theslides. Fig. 2 illustrates how Cutty works. Since the windowbottom to the top as shown in Fig. 3b. When the window

size is 18 seconds and slide size is 2 seconds, Cutty partitions

slides, records 23 and 24 arrive (Fig. 3c). The aggregating

the window into nine partitions. Each partition keeps a sub-value of the arriving records is pushed into stack1, and one

aggregation (p − value) for all records within two seconds.

element is popped oﬀ stack2. The query result is computed

When the window slides, the query result is computed byas shown in Fig. 3c. Same process occurs when other records

accumulating p − values of all partitions.

arrive.

3.2.

Final Aggregation

Though, these algorithms show good compromise for both

To improve the performance, final aggregation approaches

speed and memory usage when dealing with FIFO streams,

were proposed on top of the partition aggregation approaches

they cannot handle the out-of-order streams or non-FIFO[6–9, 14].

streams as stated in the respected papers.

These algorithms can be categorized into tworeplace the expired records in the first leaf node (left-mostleaf node). Then, the binary is updated upward to the root
node. Similarly, the late-arrival record (19) falls into the 8thleaf node (counting from left), then the binary tree is also
updated upward to the root node. Then, the query result isthe value of the root node.

(a) Records (5, 6, ..., 21, 22) arein the window.

(b) Popped elements from Stack1
and pushed into Stack2.

(a) FlatFAT for the current win- (b) FlatFAT after the window
dow (at the 22nd second).
(c) Records 23 and 24 arrive.

(d) Records 25 and 26 arrive.

Figure 3: An example of Two-Stacks for the ACQ over dataslides (at the 24th second).

Figure 4: An example of FlatFAT for the ACQ over data
streams in Fig. 1.

streams in Fig. 1. In this example, we assume that there isno late-arrival record 19 coming when the window slides atthe 24thsecond.

3.3.

Discussion

As stated in the respected papers of FIFO algorithms [6,9,14], FIFO algorithms cannot handle non-FIFO streams because they provide inaccurate aggregating results. DABA

3.2. 2.

Non-FIFO algorithms[14], FOA&IOA [14], and FlatFIT [6] rely on man complicated pointers to maintain their many inter-related intermediate results. When the window slides, not only the interme-

Few algorithms [7–9] focus on dealing with non-FIFO
streams.

diate results but also their many pointers need to be updated.

To deal with non-FIFO streams, Flink [7] andExtending these algorithms to support non-FIFO streams is

Cloud Dataflow [8] introduce a “watermark” concept to in-

unrealistic because it is very complicated, and would result

struct the systems how much time the systems should expect

in very poor performance. Therefore, these algorithms are

the arrival of late-arriving records from streams. They set

not included as the comparative approaches to the proposed

the timeframe, called watermark, to instruct the system thatalgorithm.

the past windows within a watermark duration need to be
re-executed if late stream records fall into them.

The work in [14] has proved that Two-Stacks approach is
not only simple to implement but also more eﬃcient thanAn example of these approaches is shown in Fig. 1 as pre-other existing algorithms for FIFO streams. Two-Stacks canviously explained in the Section 1. Both Flink [7] and Cloudbe easily extended to deal with non-FIFO streams. The ex-

Dataflow [8] compute the results of the aﬀected past windows

tending is to recompute all aﬀected intermediate results if

and the current window from scratch non-incrementally ev-

there are late-arrival records. However, it eagerly maintains

ery time the window slides.

many inter-related intermediate results, which is very prone

Recomputing the results of the aﬀected past windows is

to be heavily aﬀected by the late-arrival records.

trivial. However, computing the result of the current win-

For example, in Fig. 3, when the window slides at the 24th

dow is challenging. Therefore, this paper only focuses on

second (Fig. 3c), if one late-arrival tuple 19 together withcomputing the aggregating result of the current window.

non-late-arrival tuples 23, 24 come, all intermediate results

To the best of authors’ knowledge, there is only one algo-

in stack2 are aﬀected. Recompute these aﬀected interme-rithm that can compute the results of the current window in-diate results is very costly but necessary to guarantee thecrementally though its performance is very poor. FlatFAT [9]correct computation of the query result.

uses binary tree to keep all intermediate results to achieve in-

We assume that n represents the number of partitions thatcremental computation. An example of FlatFAT when deal-

can be created from the ACQ with window size W and slide

ing with non-FIFO streams in Fig. 1 is shown in Fig. 4. The

size S, and p partitions are aﬀected by the late-arrival records

current status of FlatFAT is shown in Fig. 4a. When the

per slide.

window slides (Fig. 4b), the non-late-arrival records (23, 24)

For the Two-Stacks, the numbers of elements in both

Fig. 5 is used as an example. The main idea is to group
all records in the window into two parts:•

A small number of oldest records (Blue part in

Fig. 5a): Some records in this part will expire soon, so it isnecessary to eagerly aggregate all intermediate results. The
number of all intermediate results is small, so it is cheap to(a) Current status of CBiX.

update them when some records expire or late-arrival records
fall into this part. The propose approach uses binary tree tomaintain all intermediate results on this oldest part, whereleaf nodes are all p − values, and other nodes maintain intermediate results. The root node maintains one intermediate
result, called b − value, covering all p − values for this part.
•(b) CBiX eﬃciently updates the aﬀected intermediate results overnon-FIFO streams when the window slides.

A large number of records with longer lifespan (Redpart in Fig. 5a): Only one global intermediate result, calledg − value, covering all records in this part, is maintained.

Figure 5: The main idea of CBiX when dealing with the

Even if there are many late-arrival records falling into thisACQ over data streams in Fig. 1.

part when processing future incoming streams, only one intermediate result (g−value) is updated a part from updatingstacks is n.

In the stack1, the only intermediate resultsthat need to be recomputed are the aﬀected vals and thethe aﬀected p − values.
Fig. 5 illustrates the above idea. Current status of CBiX

However, in the stack2, the aﬀected vals

at the 22nd second is shown in Fig. 5a. The binary tree

and all aggs intermediate results upwards need to be recom-

maintains the intermediate results from the last oldest six

puted. Therefore, finding size of stack2 is important. Sizeseconds of records. Other records with longer lifespan areof stack2 shrinks from n to 0, then restarts from n again.

aggregated into one g − value. At the 24th second, the win-

Therefore, getting the average size after one full process-dow slides, records 19, 23, and 24 arrive, and all records

ing of stack2 is a fair comparison. After one full process-in the 1st p − value are expired.

ing of stack2 or after n slides, the average size of stack2 isn ∗ (n + 1)
n + (n − 1) + ... + 2 + 1
n+1
2
=
=
. The latenn2arrival records can fall on the top or bottom elements, whichdates on CBiX are executed: (1) The binary tree is up-

results in small or large number of updates respectively. It(Aresult = b − value + g − value = 34 + 264 = 298).

is fair to assume that the late-arrival records fall into the
n+1
n+1
2
middle element, which requires to update=24elements. Therefore, the average number of updates to bothrecord 19), g − value is the only intermediate result to be

top-most agg.

stacks when there are p partitions aﬀected by the late-arrival
n+1
records per slide is p ∗, which is quite expensive.
4
Similarly, the binary tree in FlatFAT has n leaf nodes.
When the window slides, if there are p leaf nodes aﬀected by
the late arrival records, the cost to maintain the binary tree
per slide is (p + 1) ∗ log(n), which is also very ineﬃcient. Thenproposed approach needs only (p1+1)∗log( )+3∗p2 updateskto CBiX where p1 + p2 = p, which is explained in Section 4.
Table 1 summarizes these computational complexities.

4.

Proposed Approach

4.1.

MotivationDiﬀerent from the existing approaches that eagerly aggregate the intermediate results, the proposed approach maintains the intermediate results in an on-demand manner.

Then, two main up-

dated by purging the first leaf node, and (2) Records 19,
23, and 24 are aggregated into g − value (Fig. 5b). The
result is computed by aggregating b − value and g − value
As can be seen, though there is a late-arrival tuple (e.g.,
updated, which results in very little performance impact.
Note that, for eﬃcient purging of expired records, the latearrival record (e.g., 19) needs to be aggregated into the rightp − value (e.g., the 8th p − value).
4.2.

Structure of CBiX

Given the window and slide defined in the ACQ query, theproposed algorithm divides the window into n partitions.
After partitioning, checkpoint-based circular bidirectionalindex (CBiX) is created. CBiX is formally defined using
Definition 41
Definition 41. CBiX is a checkpoint-based bidirectional indexed array, which consists of:•n elements: Represent all partitions. The aggregating

value of each partition is called p − value.

nk checkpoints: The size of each checkpoint is | | park
titions. Each checkpoint maintains one c − value by aggre•

Table 1: Complexity Analysis
Algorithm

Avg. # of update operations Space
n+1
Two-Stacks p ∗2n4
FlatFAT(p + 1) ∗ log(n)2⌈log(n)⌉+1CBiX

n(p1 + 1) ∗ log(| |) + 3 ∗ p2
kngating all p − values of all | | partitions.
k• After the current checkpoint is entirely processed ornafter every | | slides, the followings are maintained:k– One g − value that is recomputed by aggregating all kc − values.

⌈n ⌉log(| |) +1nk(n − | |) + k + 1 + 2kb − value = updateT ree(pV alue).
•(2)

For each partition index (pIndex), checkpoint index(cIndex), and aggregating value (pValue) of p2 (Line 13),
the following processes are executed:– Update each aﬀected partition by Equation 3 (Line– A binary tree is built on the oldest checkpoint, where
all p − values in the checkpoint are the leaf nodes. The ag-

14):
p − value[pIndex]⊕ = pV alue.

gregating value of the root node is called b − value.

(3)– Update each aﬀected checkpoint by Equation 4 (LineAn example of CBiX is shown in Fig. 6. Setting the opti-

15):

mal number of checkpoints to achieves the best performanceof CBiX is explained in Section 4.4.

c − value[cIndex]⊕ = pV alue.

(4)– Update the g − value by Equations 5 (Line 16):g − value⊕ = pV alue.

(5)•

If the current checkpoint is fully processed or afternevery | | slides (Line 18), the two following processes arekexecuted:– Create a binary tree from all partitions (p − values)in the oldest checkpoint by Equation 6 (Lines 19–21).
Figure 6: A CBiX that is created from a query with windowb − value = createT ree(pV alues).

W time unit and slide S time unit. Based on the slide S,
window W can be divided into n partitions. Suppose that,nk checkpoints with size of | | partitions can be created.
k
4.3.

circular fashion from the 1th(n– Recompute the g−value by aggregating all k c−values
following Equation 7 (Lines 22–24).
g − value = c − value[1] ⊕ ... ⊕ c − value[k]Main Algorithms

CBiX is visited and filled in by the aggregating values in astp − value to the last p − valuethp−value) as new records arrive. When the n

(6)

•

(7)

The result (Aresult ) of the ACQ is computed by Equa-tion 8 (Line 26).

p−value

Aresult = b − value ⊕ g − value

is visited, it restarts the process from the 1st p − value.

(8)

Given a CBiX with n elements, and the aggregation oper-

For simplicity, the functions updateT ree and createT ree

ation ⊕. We assume that every slide aﬀects p1 partitions ofrespectively represent update and create the binary tree by

the oldest checkpoint and p2 partitions of other checkpoints,

returning the updated binary tree and b − value.

the aggregating value of the coming records of each aﬀectedpartition is pV alue, and the oldest p − value to be expired is

Example 41. We explain how CBiX works for the example

expirepV alue. The detail is explained in Algorithm 1. Whenin Fig. 1.

the window slides, the following processes are executed:•

Purge the expired records by removing the oldest

p − value (expirepV alue) from the binary tree (Line 9).
b − value = updateT ree(expirepV alue).
•(1)

For each aggregating value (pValue) of p1 (Line 10),

update the Tree by using Equation 2 (Lines 10–12).

Nine partitions and three checkpoints can be derived fromthe given window and slide. Each checkpoint consists ofthree partitions. Hence, CBiX has nine p − values, threec − values, and one g − value.
The current status of CBiX is shown in Fig. 7a. The green
part represents the binary tree, created on the latest checkpoint, the blue parts represent all checkpoints (c − values),

Algorithm 1 CBiX
Input: An ACQ with window W and slide S, aggregation operation ⊕, and data stream Str
Output: Query Result Aresult
1: Aresult = 0
2: g − value = 0(a) Current records in the window.

3: b − value = 0
4: n: number of partitions from W and S.
5: k: number of checkpoints from n.
6: Array < double > c − value[k]7: Array < double > p − value[n](b) At the 24th second (current time), one late-arrival record 19, and

8: for Window slides as new streams Str arrive do

non-late-arrival records 23 and 24 arrive. Aresult = b − value + g −9:b-value = updateTree(expirepValue)

10:

for pValue in p1 do

11:value = 34 + 264 = 298.

b-value = updateTree(pValue)

12:

end for13:

for ([pIndex, cIndex] → pValue) in p2 do

14:

p − value[pIndex] ⊕= pValue(c) At the 26th second. Records 15, 25 and 26 arrive. Aresult =15:

c − value[cIndex] ⊕= pValueb − value + g − value = 19 + 330 = 349.

16:

g − value ⊕= pValue17:

end for18:

if Current checkpoint is fully processed then19:

for pValue in oldest checkpoint do

20:

b-value = createTree(pValue)

21:

end for22:

for i ∈ {1, . . . , k} do

23:
24:

(d) At the 28th second. Record 21, 27 and 28 arrive. Aresult =b − value + g − value = 0 + 406 = 406.

g − value ⊕= c-value[i]end for25:

end if26:

Aresult = g − value ⊕ b − value(e) The current checkpoint is entirely processed, so new binary tree
on the oldest checkpoint is created, and new g −value is recomputed.

27: end for

Figure 7: Running example of CBiX for the ACQ over data
and the red part represents the of g − value. Note that, thestreams in Fig. 1.

1st c − value is currently empty.
At the 24th second, one late-arrival record 19, and non-computed by aggregating all c − values because the current

late-arrival records 23 and 24 arrive. Based on the times-g − value will become invalid due to the expiration of thep − value, andrecords in the 5th p − value. Same procedures are done whentamp, record 19 is aggregated to the 8threcords 23 and 24 are aggregated and replace the value ofthe 1st p − value (Fig. 7b) because the current value of theother records subsequently arrive.
4.4.

Optimal Number of Checkpoints

p − value are expired. Consequently, the binary tree is

The performance improvement of CBiX relies on setting

updated by purging the expired p − value. Then, the record

the right number of checkpoints. The optimal number of

19 is aggregated into the 3rd c − value (The 8th p − value

checkpoint is defined in Theorem 41.

st

1

rd

is in the 3c − value), and the g − value. Similarly, therecords 23 and 24 are aggregated into the 1st c − value (The
1st p − value is in the 1st c − value), and the g − value. The
query result is computed by aggregating b − value (34) andg − value (264), which is 298.
Same procedures are done when the window slides at the26th second (Fig. 7c), and 28th second (Fig. 7d).
After the window slides at the 28th second (Fig. 7d), thecurrent checkpoint (1st checkpoint) is entirely processed.
Therefore, the new binary tree is created on the oldest checkpoint (Fig. 7e). At the same time, new g − value is re-

Theorem 41. The optimal number of checkpoints in CBiX√n ∗ (p1 + 2).
is2 ∗ ln(10)Proof. The optimal number of checkpoints is obtained if thenumber of updates to CBiX is minimum. The number of updates to CBiX can be varied, so it is important to find theaverage one per checkpoint.
n• After every | | slides, the following updates are exek
cuted:– Construct a binary tree on the oldest checkpoint,nnwhich requires | | ∗ log(| |).
kk– Recompute the g − value by aggregating all c − value,
which requires k.

[2]• Purge the expired records at the oldest c − value fromnthe binary tree, which requires log(| |).
k• Every slide aﬀects p1 partitions of the oldest checkn
points, so it requires p1 ∗ log(| |) updates to the binary tree.
k• Every slides aﬀects p2 partitions of other checkpoints,

[3]

[4]

so it is required to update p−values, c−values, and g−value
p2 times, which requires 3 ∗ p2.
Therefore, the average number of updates to CBiX per[5]

slide is:nn
| | ∗ log(| |) + knnkkf (k) =+ log(| |) + p1 ∗ log(| |) + 3 ∗ p2
n
k
k
| |
k
2
k
= (p1 + 2) ∗ log(n) − (p1 + 2) ∗ log(k) ++ 3p2.
n[6]

Mathematically, the value of f (k) is minimum at k = k0 if

[7]

•

′

k0 is the solution of the derivative of f (k) (f (k0 ) = 0),and•

The double derivative of f (k) at k = k0 is positive[8](f ′′ (k0 ) > 0)
We have
p1 + 22∗k+k ∗ ln(10)n√n ∗ (p1 + 2)
f ′ (k) = 0, then k0 =2 ∗ ln(10)f ′ (k) = −f ′′ (k) = (−) ∗ (−) ∗[9]p1 + 212∗+
log(10) k2
np1 + 22+
k2 ∗ log(10)nReplace the value of k0 to f ′′ (k). Then, we get f ′′ (k0 ) > 0.

[10]

=

The theorem is proved.
[11]

5.

Conclusion

We have proposed a novel checkpoint-based bidirectional
index (CBiX) for incremental sliding-window aggregation[12]

over out-of-order data streams. The proposed algorithm efficiently compute all intermediate results in an on-demandmanner. All intermediate results are categorized into two[13]

main groups: (1) b − value and (2) g − value with less impact from the late-arrival records.
For future work, we are going to finish the implementations. Then, the experiments will be performed.

Acknowledgment
This work has been partly supported by the NICT BigClouT project.
References[1] S. Bou, T. Amagasa, and H. Kitagawa, “Scalable keyword[14]

search over relational data streams by aggressive candidate
network consolidation,” in Information Systems - ELSEVIER, ser. C, vol. 81, 2019, pp. 117–135.
——, “Filtering xml streams by xpath and keywords,” inProceedings of the 16th International Conference on Information Integration and Web-based Applications & Services(iiWAS 2014), 2014, pp. 410–419.
——, “Path-based keyword search over xml streams,” in INTERNATIONAL JOURNAL OF WEB INFORMATIONSYSTEMS (IJWIS), ser. 3, vol. 11, 2015, pp. 347–369.
——, “An improved method of keyword search over relational data streams by aggressive candidate network consolidation,” in In: Hartmann S., Ma H. (eds) Database andExpert Systems Applications. DEXA 2016. Lecture Notes
in Computer Science, vol. 9827, 2016, pp. 336–351.
——, “Scalable keyword search over relational data streams
by aggressive candidate network consolidation,” in Information Systems (2018), 2018, https://doi.org/10.1016/j.is.2018.12.004.
A. U. Shein, P. K. Chrysanthis, and A. Labrinidis, “Flatfit: Accelerated incremental sliding-window aggregation forreal-time analytics,” in In Proc. International Conference
on Scientific and Statistical Database Management, ser. Article No. 5, 2017, pp. 1–12.
P. Carbone, A. Katsifodimos, S. Ewen, V. Markl, S. Haridi,
and K. Tzoumas, “Apache flink: Stream and batch processing in a single engine,” in IEEE Data Eng. Bull, ser. 4,
vol. 38, 2015, pp. 28–38, https://flink.apache.org/.
T. Akidau, R. Bradshaw, C. Chambers, S. Chernyak,
R. J. Fernandez-Moctezuma, R. Lax, S. McVeety, D. Mills,
F. Perry, E. Schmidt, and S. Whittle, “The dataflow model:a practical approach to balancing correctness, latency, andcost in massive-scale, unbounded, out-of-order data processing,” in Proceedings of the VLDB Endowment - Proceedingsof the 41st International Conference on Very Large Data
Bases, ser. 12, vol. 8, Kohala Coast, Hawaii, pp. 1792–1803.
K. Tangwongsan, M. Hirzel, S. Schneider, and K.-L. Wu,“General incremental sliding-window aggregation,” in In
Proc. the VLDB Endowment, ser. 7, vol. 8, 2015, pp. 702–
713.
J. Gray, S. Chaudhuri, A. Bosworth, A. Layman, D. Reichart, M. Venkatrao, F. Pellow, and H. Pirahesh, “Datacube: A relational aggregation operator generalizing groupby, cross-tab, and sub totals,” in Data Mining and Knowledge Discovery, ser. 1, vol. 1, 1997, pp. 29–53.
D. M. Jin Li, K. Tufte, V. Papadimos, and P. A. Tucker,“No pane, no gain: eﬃcient evaluation of sliding-window aggregates over data streams,” in In ACM SIGMOD Record,ser. 1, vol. 34, 2005, pp. 39–44.
S. Krishnamurthy, C. Wu, and M. J. Franklin, “On-thefly sharing for streamed aggregation,” in In Proc. the ACM
SIGMOD international conference on Management of data,
2006, pp. 623–634.
P. Carbone, J. Traub, A. Katsifodimos, and V. M.
Seif Haridi, “Cutty: Aggregate sharing for user-defined windows,” in In Proc. the ACM International on Conference on
Information and Knowledge Management, 2016, pp. 1201–
1210.
B. Moon, I. F. V. Lopez, and V. Immanuel, “Low-latency
sliding-window aggregation in worst-case constant time,” in
In Proceedings of the 11th ACM International Conference
on Distributed and Event-based Systems, 20017, pp. 66–77.