Distributed, highly-parallel processing frameworks as Hadoop are deemed to be state-of-the-art for handling big data today. But they burden application developers with the task to manually implement program logic using lowlevel batch processing APIs. Thus, a movement can be observed that high-level languages are developed which allow to declaratively model data ows that are automatically optimized and mapped to the batch-processing backends. However, most of these systems are based on programming models as MapReduce that provide elasticity and fault-tolerance in a natural manner since intermediate results are materialized and, therefore, processes can simply be restarted and scaled with partitioning input datasets. For continuous query processing on data streams, these concepts cannot be applied directly since it must be guaranteed that no data is lost when nodes fail. Usually, these long running queries contain operators that maintain state information which depends on the data that has already been processed and hence they cannot be restarted without information loss. This also is an issue when streaming tasks should be scaled. Therefore, integrating elasticity and fault-tolerance in this context is a challenging task which is subject of this paper. We show how common patterns from parallel and distributed algorithms can be applied to tackle these problems and how they are mapped to the Mesos cluster management system.
Processing and analyzing big data is one of todays big challenges. A popular de nition from a Gartner report names the three 'V' s { volume, velocity, and variety as the main characteristics of big data. Among them, velocity refers to the analytics of dynamic data even in (near) realtime.
Several approaches and techniques have been developed in the past to process dynamic data. Data stream management systems (DSMS) like STREAM, Aurora, IBM InfosThis work is partially funded by an IBM PhD Fellowship. phere Streams or our own AnduIN engine provide abstractions to process continuous and possibly in nite streams of data instead of disk-resident datasets. Typically, this includes standard (relational) query operators, window-based operators for computing joins and aggregations as well as more advanced data analytics and data mining operators working on portions of the stream, e.g. windows or synopses of data. Complex Event Processing systems (CEP) particularly support the identi cation of event patterns in (temporal) streams of data such as a sequence of speci c event types within a given time interval. Typically, systems of both classes provide a declarative interface, either in form of SQLlike query languages like CQL for DSMS, event languages like SASE, or in the form of data ow speci cations like SPL in IBM Infosphere Streams.
Recently, several new distributed stream computing platforms have been developed, aiming at providing scalable and fault-tolerant operation in cluster environments. Examples are Apache S4 or Storm. In contrast to DSMS or CEP engines theses platforms do not (yet) provide declarative interfaces and, therefore, require to program applications instead of writing queries. Developers of these systems argue that they provide the same for stream processing what Hadoop did for batch processing { which raises the hope of a similar movement towards higher-level languages as we can see with Pig, Jaql etc. for MapReduce.
However, there are some challenges in scalable and elastic stream processing which are di erent from batch processing with Hadoop. Whereas in Hadoop, input data as well as intermediate results are materialized on disk and, therefore, both, map and reduce tasks can be restarted arbitrarily in case of failures until the entire job is nished, since computation state is saved, the number of nodes assigned to map and reduce tasks can be simply adjusted by partitioning input and intermediate results.
This is more di cult when processing dynamic data { even with platforms as S4 or Storm which to some extent support a reliable and scalable operation. The main di erences are: (1) Partitioning of streams for data-parallel processing is not always easily possible, for example in case of windowor sequence-based operators including CEP operators. Also, elastic operation by adding new nodes at runtime of a query requires at least rerouting of data. (2) Stream queries are typically long running queries which cannot be simply restarted without losing data. Furthermore, because of this, the deployment and resource allocation (placement of queries on nodes, allocating memory and CPUs) are much more critical.
In this paper, we try to answer the question how to bridge the gap between an easy-to-use, high-level declarative interface for data stream analytics and scalable cluster-based stream computing platforms in order to address these challenges. The contribution of this paper is twofold: Based on a basic data ow model for stream queries we describe patterns for fault-tolerant and scalable query processing and discuss constraints of their application. We show the implementation and deployment of these patterns using our distributed stream and CEP engine AnduIN and the Mesos cluster infrastructure by describing techniques supporting exible and elastic deployment. Though, we use the AnduIN system for describing and implementing the concepts, we think the ideas and patterns are applicable to other platforms, too. 2.
The relevant work related to this paper can be classi ed into the main categories: continuous query processing and scalable data ow platforms.
Continuous query processing is usually implemented in data stream management systems (DSMS). Pioneered by systems like STREAM, Borealis, and Telegraph, several approaches and systems have been developed in the last decade including commercial products such as IBM InfoSphere Streams and StreamBase. Typically, these systems provide a SQL-like query language enhanced by features for dealing with continuous queries such as sliding windows.
Partitioning, distributed processing, and fault tolerance
have been studied to some extent, e.g., in Borealis [
Another approach in the form of a programming model
has been proposed in [
An approach addressing load balancing issues by
partitioning while providing fault tolerance for pipelined
data ows is Telegraph's FluX [
Scalable data ow platforms try to extend the
applicability of the MapReduce paradigm for large-scale parallel batch
processing to pipeline processing and continuous query
support. One example is HOP (Hadoop Online Prototype) [
Optimus [
In the following we assume a simple processing model for
data stream queries: a query is represented by a data ow
graph which is a common model in literature [
In such a directed acyclic graph, nodes represent query
operators and edges describe the tuple ow between them.
Query nodes can be arbitrary pipeline operators of a stream
query algebra [
There are two main reasons for distributing query nodes: increasing processing reliability by introducing redundancy, and increasing performance and/or scalability by load distribution. The following patterns support these goals to di erent degrees. In this discussion, we use the term \query task" as the unit of distribution/scheduling for both, elementary algebra operators, and for data ow (sub-)graphs with wellde ned properties (input/output, stateless vs. statefulness). Pattern 1: Simple standby: For a critical query node N a standby node S is maintained on a separate compute unit which is activated if N fails. This requires monitoring of N , e.g. by a combination of heartbeats and cluster coordination service such as ZooKeeper as well as rerouting the input tuple stream to S. Since in case of a failure the state of N is lost, this pattern is applicable only for stateless nodes. Pattern 2: Checkpointing: This pattern is similar to pattern 1 but supports stateful operators. Failover is achieved by periodically checkpointing the state of the critical node N to a shared disk and restarting the standby node S from the checkpoint. Examples of such checkpoints are the content of sliding windows or hash tables for joins and aggregations. Pattern 3: Hot standby: If the failover time of pattern 2 is not acceptable, a hot standby approach can be chosen where redundant query nodes S are kept actively. To achieve this, the input stream has to be sent to all redundant nodes, either by using multicast strategies at the network or at the application level. This pattern works both with stateless and stateful query operators but requires a special node to eliminate duplicate results, e.g., a stream selector node which forwards the input from only one of multiple streams. failover Pattern 4: Stream partitioning: This pattern exploits data parallelism by partitioning the input stream. It can be implemented by a splitter node redirecting each input tuple to one of the query nodes N1 : : : Nk or by a multicast writer with an additional partitioning node P1 : : : Pk for ltering the input stream according to the partitioning scheme. Finally, the results are merged into a single stream. Pattern 5: Stream pipelining: In contrast to pattern 4 this pattern exploits task parallelism by splitting a complex query node N into a sequence of query nodes N1 : : : Nk and placing them on separate compute units.
Usually, multiple patterns will be applied in order to
achieve certain quality of service (QoS) guarantees as fault
tolerance (patterns 1-3) or elasticity for adapting resource
consumption (patterns 4-5) according to the needs of the
applications. Of course this pattern list does not claim
completeness. There are several others that are applicable under
certain circumstances, e.g., parallelization through
aggregation trees for commutative and associative aggregation
operators [
In the following we will describe how these patterns can
be utilized in a data ow framework to dynamically
restructure the physical representation of the graph in a continuous
query context that is executed in a cluster infrastructure.
The restructuring is achieved with a simple set of rewriting
rules that are automatically applied on the graph without
the need to manually code them as in existing approaches
[
1Actually, aggregation trees are just combinations of partitioning and pipelining patterns.
AnduINv2 queries are deployed as separate processes in Mesos which are just-in-time compiled using the system's C++ compiler. This provides an easy mechanism to plug-in user de ned operators and exchange operator implementations. During deployment, processes are interlinked by query channels which simply represent an abstraction over network connections (TCP/UDP sockets, ZeroMQ connections).
Query tasks can be shared among multiple queries when
they share some common (sub)streams or operators as
described in [
Dataflow Graph Rewriting data ows are speci ed in an XML format which can be seen as an intermediate representation, allowing to use different frontends such as CQL or graphical tools. A data ow speci cation consists of stream type de nitions and operator de nitions with name, type, type-speci c parameters as well as input and output channels. These channels are typed and are used to interconnect operators to form a graph. The following example shows a simple data ow speci cation. (We omitted the XML notation for better readability). type name = "aStreamType" f
column name = "x", type = "int" . . . g operator name = "source", type = "reader" f
output name = "aStream", type ="aStreamType" g operator name = "myFilter", type = " lter" f input name = "aStream" param condition = "aStream.x < 42" output name = " lteredStream", type = "aStreamType" g operator name = "sink", type = "writer" f
input name = " lteredStream" g
Task Query executor Mesos slave
Query
Query operator
Query task deploy Query scheduler Query 1: t1, t2, t3, ...
Query 2: ...
Query 3: ...
Mesos master Task Query executor Mesos slave <t1, cores, mem, ports, …> <t2, cores, mem, ports, …>
...
Standby master Task Query executor Mesos slave
Task Query executor Mesos slave s l e n n a h C t u p n
I I~(t)
Parameters (Meta-Query) Query Task
Task State
P~(t) ⌧ i
⌧ p S~(t)
T Input Queue Operator States Output Queue
T T ... future tuples
... processed tuples
O u t p u t C h a n n e l s O~(t) P2 Pk N2
Nk
Note, that apart from stream input and possible output no communication operators have to be speci ed as part of the query. Such operators are added during rewriting if necessary. For formulating rewriting rules we use a simple notation. A data ow as given above is written as sink := writer(f := lter(src := reader)) where writer, lter, and reader are operator types and the optional sink, f, and src names denote operator instances.
During rewriting, graph patterns have to be matched and constraints are checked. For this purpose, the pseudo-type any is used as a placeholder for any possible type, and any* represents a data ow of arbitrary operator types. The following pattern matches a data ow subgraph containing a stateless lter operator (which is the case for any lter): a2 := any(f := lter(a1 := any*))[stateless(f)] To apply the patterns, a rewriting rule can be speci ed: ) a2(stream-selector(failover(writer(f(reader(multicast(a1))))))) | @{pz1 } | @{pz } | @{pz2 }
Besides inserting or replacing operators (such as streamselector, failover, and multicast operators in the previous example), operator nodes are also annotated with placement information where pi; pj with i 6= j denote distinct compute nodes and p denotes an arbitrary number of nodes. 4.2
With these rewriting rules, internal data management nodes and di erent operator implementations can be transparently injected into the query plan without impacting the application. This allows to re-schedule a query task to another node in case of a failure (pattern 1), use an operator implementation that automatically integrates snapshotting (pattern 2), or replicate a task to implement hot standby.
The actual ow of data through query channels during runtime is controlled by special operator parameters { e.g., target IP addresses and ports { that can be adjusted through a concept we call meta queries (cf. Sect. 4.5). To detect and react on failures, query tasks are instrumented with monitoring interfaces that inform the query scheduler about the nodes' health and performance measures as tuple processing rates. The scheduler then triggers a graph rewriting.
When a rewritten graph needs to be deployed, it has to be
guaranteed that no information of the tuple stream is lost.
To analyze the necessary steps, we reduce a query task to a
nite state machine model (cf. Fig. 6) which is common for
implementing CEP operators [
N1
N2
Nk windows) and are updated with each incoming tuple through a state transition function i 2. The meta query extension is represented by special input channels P~ (t) that modify the operator state through p.
To guarantee that a node failure does not lead to an
information loss it is necessary that all results which have not
been consumed by the following target can be reproduced
from the possibly in nite input stream. Therefore, the
operator state needs to be snapshotted after each input tuple,
or { if this is too expensive { the input tuples need to be
persisted in order to reproduce this state with just
'replaying' the input. Which tuples are still required for a possible
replay can be controlled by special tuple messages that are
exchanged between tuple producers and consumers as in the
Borealis system [
In order to implement fault tolerance with transferring
stateful query tasks to other computing nodes, a simple
protocol as presented in [
The same algorithm can be used to replace query tasks
with their rewritten versions that compute the same logical
transformation but use di erent operator implementations
and/or partitioning schemes of the query graph into tasks
for implementing the elasticity patterns 4 and 5.
Rewriting Cost Model: Usually, there are several
possibilities for rewriting data ow graphs. In order to make right
decisions which tasks shall be replaced and how many nodes
should be allocated, a cost model is required taking
possible rewriting bene ts into account as well as costs for the
restructuring, e.g., for transferring states or costs for
additional resources from the cluster infrastructure. Discussing
elaborate decision models is not in the focus of this paper
and is left for future work. We outline a rate-based model
that is suitable to nd hot spots in data ows and is used in
related literature [
Rewritings should be done when a query task is detected
that cannot process its incoming tuples with a rate higher
than the arrival rate, e.g., when the computational
complexity or the memory consumption is too high and therefore the
task accumulates an increasing backlog. Such a task
represents the critical path in the data ow, limiting the overall
throughput. For nding these paths, a rate-based
optimization approach is suitable that scans the data ows starting
at source nodes and detects such bottlenecks based on
monitoring information gathered during the execution [
Inter-operator parallelism (Pattern 4): Usually, it is
better to keep data ow operations on few nodes in order
to avoid costly transfer operations. Hence, initially
compiled queries will comprise few tasks consisting of large
ow (sub)graphs. However, when the computational
complexity exceeds a certain threshold or memory limits for
keeping state information are exceeded, a distributed
processing pipeline will yield better performance. Large graphs
are partitioned, recompiled, and distributed on additional
nodes in the cluster. Moreover, splitting large costly tasks
into smaller distributed ones increases fault-tolerance since
it will become cheaper to recover from node failures [
The most challenging problem in this context is nding suitable partitioning schemes for the operators that shall be deployed in the framework, especially when they are not stateless. In the following, we will present a parallelization scheme in the complex event processing (CEP) context which is prominent in stream processing.
The task of CEP is nding complex patterns in a stream
of base patterns. These patterns are de ned through
certain properties of incoming tuples, usually described through
predicates and additional correlations of their arrival time.
Mostly, sequences and repetitions are used which can be
expressed with regular expressions [
1 partition P
P A matches any incoming tuple B+ matches all following tuples with decreasing value C matches the rst tuple with increasing value after B but with a value less than the previous one of A D+ matches following tuples with increasing values > A For parallelization, we focus on the +-operator since it is challenging for three reasons: First, like a join, it can produce multiple output tuples per input tuple. Second, it needs to store state information for extending existing patterns to longer ones. Since each tuple is possibly multiplying the number of results it is likely that { due to memory constraints { such operators will become critical in the data ow graph. Third, in most cases the behavior of the stream is not predictable, rendering static allocations infeasible.
Fig. 8 illustrates how the operator can be distributed dynamically to multiple nodes with simply applying graph patterns 4 and 5. The P +-operator comprises two parts: a pattern matcher P , and a +-operator which maintains all previously matched patterns as state and concatenates them with subsequent matches. The output of + is the output for the entire operator and serves as input for + again to construct longer matches. On memory over ow, the operator state can be partitioned and distributed to multiple instances where all instances receive the original input stream. Two di erent behaviors of the operator are required to avoid duplicates. The rst instance processes the input directly, i.e., all matching tuples serve as new patterns of length 1. Those matches must not be reproduced by other instances that simply serve as targets for over owing patterns that do not t into the local state but are independent from each other and hence can be processed on separate nodes. When the complexity of the matching algorithm P is critical, elasticity can also be achieved with implementing a pipeline. It exploits the fact that P + can be expressed through P _(P P +), i.e., a pipeline of arbitrary length is constructed for matching incoming tuples in parallel, emitting them as output, and forwarding them to the next stage for extension.
Since such parallelization schemes depend on operator
semantics, the framework provides them to automatically scale
up and down required resources for built-in operators. For
all user de ned functions which are treated as black boxes
by AnduINv2, the parallelization needs to be implemented
by the user as in [
Mesos [
Though, Mesos provides mechanisms to deploy processes, it does not support elastic operation for stream queries. In Hadoop, it is the task of the job tracker to partition the work across a set of map and reduce tasks. In case of data streams the situation is a bit di erent, because we cannot simply stop and continue/restart queries without loosing data. The only way to achieve elasticity is to change query behavior at runtime. Therefore, we introduce the idea of meta queries: in each (adjustable) query task an additional query is running on a control stream consisting of tuples of the form: hquery id; operator id; parameter; valuei
The control stream is produced by the query scheduler which monitors resource utilization and implements strategies for dynamic reallocation. Meta queries are particular useful for implementing the patterns described in Sect. 3. For instance, for failover without publish-subscribe (pattern 1 and 2), the network writer has to be informed about the network address of the newly activated standby node S. For this purpose, the network writer provides a parameter targetaddr for the target address. A control stream tuple like received by the query task triggers sending the tuple stream to the standby node node2. Similarly, for implementing pattern 3, the stream selector node can be informed about switching to the stream produced by query node S.
For partitioning patterns like pattern 4 it is either required to modify the tuple distribution strategy of multicast writers or to adjust partitioning predicates Pi in Fig. 3. Both can be easily implemented by sending appropriate control tuples. 5.
We presented basic concepts how fault-tolerance and elasticity can be achieved in the context of continuous query processing by combining techniques that have proven applicability in other scenarios. These approaches are currently being integrated into AnduINv2, but can be applied in other platforms, too. Our main questions we would like to answer with future experiments are:
While the rst questions intent to pave the way for a streaming-as-a-service infrastructure, answering the last one is needed to keep up with current hardware development trends. We believe that parallel and specialized processors as many-core CPUs, GPUs, or FPGAs will nd their way into future computing centers to provide the most e cient computing platforms for dedicated tasks { an important aspect to tackle the big data challenge.