As of today, active replication is used in complex event processing systems to enable near zero latency take over in case of host failures. Moreover, elastic complex event processing systems adapt their resource consumption to the actual system load. However, active replication is a coarse-grained approach demanding the duplication of all used resources. Therefore, we envision a system adopting adaptive ne-grained replication strategies allowing to trade o availability and used resources.
Switzerland
The dissemination of high frequency event sources raises
the demand to extract information from high velocity data
streams. Prevalent application domains comprise automatic
stock trading, credit card fraud detection, automated
homecare, as well as scienti c experiments, logistics, and
telecommunication. To process high velocity data (> 10.000 events
per second) with low latency (< 100 ms), a new class of
applications enabling the e cient analysis of data in real-time
has ermerged. Prominent examples of such complex event
processing systems comprise Borealis [
Distributed complex event processing systems spanning
thousands of hosts cope with very high data rates and
extensive computations. However, the error probability increases
with the number of components in a system. Because in data
centers the probability for a single host to fail at least once
a year is between 4 and 8 percent [
Various fault tolerance techniques like active replication [
The remainder of this paper is structured as follows: Section 2 introduces the assumed system model. Section 3 presents an approach leveraging spare resources to improve the availability of queries. In Section 4 we propose an optimization component to minimize the resource consumption for a certain availability target. The challenges arising from an application of the approaches in an elastic system are outlined in Section 5. Related research is discussed in Section 6. Finally, Section 7 concludes the paper. 2. 2.1
Queries in the system are continuous queries that can be added and removed at any point in time. As opposed to one-shot queries that are sent to the system and then produce a result once, continuous queries remain in the system for a certain period of time and produce results continuously. Let Q = fq1; q2; : : : ; qlg be the set of queries in the system.
Queries are speci ed by the user in an event processing language (EPL). In the system queries are represented as directed acyclic graphs (DAGs). Nodes represent operators that are connected by unidirectional edges. A query compiler transforms queries speci ed in EPL into a query graph.
Each operator has one or two inputs and multiple outputs. Operators with two inputs are referred to as binary operators. Each operator has a type de ning its basic behaviour: source, projection, lter, aggregation, sequence, join, and destination. Besides a type all operators, except sources and sinks, have a predicate re ning its functionality. For example in case of a lter operator the predicate speci es the lter condition.
The following example query calculates the average price of the SAP stock over the last 10 minutes.
INSERT INTO outStream
SELECT compName, avg(tick) FROM tickStream WITHIN 600 seconds GROUP BY compName
WHERE compName = "SAP";
The corresponding query graph, which is depicted in Figure 2, contains a source, lter, aggregation, and destination operator.
To minimize the number of operators in the system, all
queries are optimized via a query optimization component,
which analyses the query graph of already running queries
with respect to reusable operators leveraging incremental
multi query optimization (MQO) techniques [
S1 S1
F1 F1
A1 merge A2
D1 D2
Running
Query New Query
To ensure near zero latency takeover, we assume active replication in the system. Each operator can have multiple replicas exhibiting exact the same behaviour as the primary operator. However, this approach results in the consumption of additional resources.
Query operators and replicas are deployed independently on available hosts. The current placement of operators on hosts is described via a placement function plc : O ! H, where O = fo1; o2; : : : ; omg is the set of operators for all queries in the system and H = fh1; h2; : : : ; hng denotes the set of used hosts. 2.3
According to the query model the system comprises several
operators (timed processes) that are executed on a set of
hosts. We assume that each host has a probability p to fail
with a crash stop failure. Crash stop failures result in an
immediate crash of all operators placed on the speci c host.
Moreover, we assume that Byzantine (value) errors caused
by erroneous executions can be transformed into crash stop
failures, e. g. by means of Software Encoded Processing [
A query is considered broken, i for at least one of its operators neither a primary operator nor an operator replica is executed anymore. 2.4
We assume a load model, which is based on work in the
context of the data streaming system Borealis [
For the example query depicted in Figure 2 the load of the di erent operators can be speci ed as load(S1) = r1 c(S1), load(F1) = r1 c(F1), load(A1) = r2 c(A1), load(D1) = r3 c(D1). Since each operator is associated with a selectivity value, input event rates of operators can be calculated as linear combination of source stream rates and operator selectivity of predecessor operators. For example the input event rate r3 of operator D1 can be expressed as r3 = sel(A1) r2 = sel(A1) (sel(F1) r1).
S1 r1
F1 r2
A1 r3
D1
Moreover, each operator has incoming (netin) and outgoing (netout) network tra c. The incoming tra c netin(op) is calculated as product of input rate and input tuple size of the operator, whereas the outgoing tra c netout(op) is calculated by the output rate multiplied with the output tuple size.
For the sake of convenience in the remainder of this paper the term load is used interchangeably with CPU resources. 3.
Complex event processing systems are exposed to a varying workload caused by di erent event rates as well as the addition and removal of queries. Figure 3 sketches a ctitious workload of a system processing queries for tra c monitoring. Peak loads indicate rush hour tra c. To deliver results for the queries in real-time, the system has to be equipped with enough resources to handle those peak loads. However, since the peak load can only be estimated, phases of underprovisioning may occur, which are indicated by the dark gray areas. Moreover, this leads to phases of overprovisioning indicated by the dotted areas.
70 60 ts50 s o40 H ed30 s U20 10 0
Overprovisioning
Time
Underprovisioning
Heuristic replication targets the usage of such spare resources in distributed complex event processing systems to increase the availability of queries. Assume a system that comprises a set of n hosts H = fh1; h2; : : : ; hng with an overall CPU capacity of loadcap = n hostcap. Query operators are placed according to an operator to host mapping plc. The unused CPU resources loadrem can be calculated as loadrem = loadcap
X load(op) op2O Since queries can be added to and removed from the system at runtime the amount of remaining resources usable to improve the availability of queries in the system changes over time. If the remaining load is smaller than the load of all queries in the system, not all operators can be replicated and, thus, a subset of operators has to be selected for replication. Thus, we propose a three step approach. First, all operators are rated according to their potential to improve query availability by means of an operator importance heuristic (see Section 3.1). Second, the set of operators as well as the replication level, i. e. the number of replica instances, is determined for each operator using an operator selection algorithm (see Section 3.2). Third, the replica instances are deployed on hosts according to an operator placement strategy (see Section 3.3). Finally, in Section 3.4 we present an evaluation. 3.1
The potential of operators to improve the availability of queries is rated using a normalized importance function impheur : O ! [0; 1] that associates each operator in the system with a normalized importance value. Depending on the source of information used for calculating this importance value, static and dynamic heuristics can be distinguished. 3.1.1
Static heuristics calculate the importance of operators based on properties observable in the global query graph. Due to the multi query optimization operators can be reused by multiple queries. Hence, if a reused operator fails, all depending queries crash. One possible heuristic, referred to as query out degree heuristic, is based on this observation and rates the importance of operators that are reused by multiple queries more important. The calculation of the query out degree heuristic can be expressed as impQOD(op) = deg(op) jQj where deg(op) is the number of queries leveraging the operator op, and jQj is the number of queries running in the system. 3.1.2
Dynamic heuristics assess runtime properties of operators and are, thus, dependent on models providing estimations if an operator is deployed for the rst time. If for example the required resources are considered, the heuristics are dependent on the load model of the system (see Section 2.4). One example for a dynamic heuristic is the low utilization heuristic that tries to replicate as much operators as possible. Therefore, operators with a small load are preferred. The calculation of the heuristic can be expressed as follows: impLU (op) = 1
load(op) max(load(O)) where max(load(O)) is the maximum operator load among all operators currently running in the system. 3.1.3
The heuristics introduced so far operate on an operator level, i. e. operators are treated independent of each other. Hence, operator-level heuristics might cause the selection of only most but not all operators for a query. Not selected operators are weak spots and decrease the achievable availability for the query tremendously. Therefore, we propose to combine operator importance values into query importance values, which are expressed using a normalized importance function for queries impQheur : Q ! [0; 1]. Operator selection algorithms operating on query level heuristics will ensure that once a query was selected for replication all operators of that query are replicated. Only if not enough resources are available to replicate a full query, single operators would be selected. 3.2
Operator selection strategies take care of the actual operator selection based on the operator importance and the remaining load loadrem in the system. The calculated operator selection can be described as function sel : O ! Z, which associates each operator with a number of replicas referred to as replication level rep(op). 3.2.1
A simple operator selection strategy sorts all operators descending based on their relative importance de ned via impheur. Afterwards operators are selected for replication until all spare resources loadrem are consumed. Selecting operators for replication includes the determination of the replication level, i. e. the number of replicas to create for a single operator. Di erent selection procedures can be established: 1. All operators can be selected for replication at least once if enough remaining resources are available. If afterwards spare resources are still available, the replication level for the already selected operators can be increased stepwise. This procedure ensures that each operator is replicated at least once, if enough resources are available. 2. A replication level larger than one might be set directly for an operator. Hence, some operators might be excluded from replication, if already all resources are consumed. 3.2.2
The simple operator selection strategy selects operators
for replication based on either a static or dynamic heuristic.
Thus, only one type of available information is incorporated
at a time and the resulting selection of operators can lead to
a non-optimal availability of queries. Therefore, we propose
to augment the selection process based on a static heuristic
with runtime information by modeling the operator selection
problem as 0-1 knapsack problem [
Each operator opi 2 O has a value vi = impheur(opi) and a weight wi = load(opi). The maximum weight of the bag equals the remaining resources loadrem in the system.
Because the 0-1 knapsack problem is NP-hard and load values of operators and, thus, the optimal solution changes continuously, we suggest an approximation by means of a combination of static and dynamic heuristics. The optimization is based on the intuition that operators with the greatest pro t per weight unit have to be selected rst. Thus, we propose to leverage a product of the query out degree and low utilization heuristic
impQOD LU (op) = impQOD(op) impLU (op) as basis for the operator selection. 3.3
Operators selected for replication via the operator selection algorithm have to be deployed on hosts. However, depending on the chosen operator placement for the same set of selected operators di erent availabilities can result. 3.3.1
A bin packing algorithm [
Each host is modeled as bin, where the available CPU resources form the capacity. Replicas are rst sorted in decreasing order according to their normalized importance and then assigned using their load as weight. Moreover, two constraints are ensured: (i) the network capacity of a host should not be exceeded, (ii) two replicas of the same operator are never placed on the same host. To reduce the consumed network bandwidth, possible hosts are ordered according to a neighboring factor representing the amount of predecessor and successor operators deployed on the same host. 3.3.2
Even though the simple bin packing algorithm is replicaaware due to the additional placement constraint, the system availability is not maximized because as many replicas of 100% Query Out Low Utilization (50%) Low Utilization (50%) Query Out (50%)
Random (50%) 0.0 0.5 1.0 2.0
Host Crash Probability (%) ● 3.0 di erent operators as possible are placed on one host. Thus, multiple replicas crash in case of a host failure.
Therefore, we propose to use a bin packing algorithm with
color constraints [
We have performed a preliminary simulation-based evaluation of the proposed approaches. We used a query generator to generate queries based on six query templates that di er in structure and operator parameters. Operators were deployed on simulated hosts with a CPU capacity of loadcap = 1. We assume a system comprising 100 hosts. Moreover, operators were replicated at most once. To guarantee statistical correctness, 1000 runs were conducted for each experiment and values have been averaged.
Figure 4 depicts the percentage of remaining running queries after a time period of one day assuming di erent host failure probabilities p = f0:0025; 0:005; 0:01; 0:02; 0:03g and di erent heuristics for rating the importance of operators. Because a constant system load load(O) = 100 was generated, the stated percentages of resources available for replication are equal to the actual remaining load loadrem in the system. Leveraging either the low utilization heuristic or the query out degree heuristic to select operators for replication improves the percentage of remaining running queries by approximately 1.7 percentage points compared to a random operator selection assuming a host failure probability of p = 0:02 and loadrem = 50. Using a combination of both heuristics improves the query availability further by 0.4 percentage points, resulting in 99.2 % remaining running queries.
Figure 5 shows the availability increase as a function of available resources for replication. A combination of query out degree and low utilization heuristic results in 99.2 % remaining running queries, if a host failure probability of p = 0:02 and loadrem = 50 is assumed. Moreover, the achievable percentage of running queries is only diminished )100 ● ● ● % ( ise98 r e u96 Q g in94 n n u R92 100% 90% 50% 10% 1% ● 0% ● ● ● 3.0 0.0 0.5 1.0 2.0
Host Crash Probability (%) by approximately one percentage point compared to full replication, if 50 % less spare resources are used for replication and the host failure probability is p = 0:03. This result can be explained by the fact that with loadrem = 50 available for replication still 80 % of the query operators are replicated. Moreover, those operators are reused to a large extent or have small cost.
Service level agreements are important in scenarios where a certain availability is required. However, achieving a higher availability by replicating more operators requires more resources. Therefore, the solicited availability and the resulting resource consumption should be traded o . Replication of operators in a system can be optimized with respect to two di erent optimization goals: 1. For a given resource limit costmax, the availability of queries A(Q) can be maximized. 2. For a given availability of queries A(Q), the resource consumption cost(Q) can be minimized.
The achieved availability of queries is in uenced by the set of operators selected for replication, the number of replicas that is created for each of the selected operators, as well as the placement of replicas on hosts. All those in uential parameters are re ected in the placement plc and, thus, the placement can serve as predictor for the achievable availability. To optimize the replication decisions according to the two speci ed optimization goals, additional models have to be incorporated. To estimate the costs for a placement, the load model for query operators can be leveraged. Moreover, an availability model is required that enables the estimation of the query availability resulting from a certain placement.
Assuming the availability of those models, the optimization problems can be formulated as follows: maxfA(plc) j plc 2 P; cost(plc) costmaxg and minfcost(plc) j plc 2 P; A(plc) A(Q)g where P is the set of all possible placements resulting from di erent operator selections, di erent replication levels, as well as di erent placement strategies.
Those optimization problems can be solved using
wellknown optimization algorithms like genetic search [
Elastic data stream processing systems are able to cope with varying query as well as event load. Mechanisms to dynamically allocate and release hosts depending on the actual demand prevent costly overprovisioning and performance barriers due to underprovisioning. Ideally the system can scale out inde nitely to serve high event rates and scale in to lower the used resources in case of low utilization.
Processing queries in an elastic environment poses various challenges. Once new queries are submitted to the system, the encompassed operators have to be distributed to the running hosts. Because of operator reuse the load of already deployed operators changes. To not impair the performance of the system two reactive actions are taken: (i) overloaded hosts might be relieved by automatically migrating operators from one host to another, (ii) overload situations that cannot be resolved by moving operators from one host to another are resolved by splitting the operator into several operator instances that handle only a portion of the overall load and can be deployed independently.
However, handling replicas in a dynamic system is demanding:
The complexity of the recon guration caused by the exoneration of overloaded hosts is increased since additional communication channels for replicas have to be maintained.
If operators are split into several instances, all replicas have to be split too in order to maintain a consistent system state.
Systems that are used in conjunction with heuristic replication (see Section 3), have to decide in case of spare resources whether to release hosts or to create additional replicas.
Recon guration routines have to ensure that the new placement does not violate existing service level constraints for the queries (see Section 4). 6.
Fault tolerance techniques for data stream processing
systems like active replication [
Moreover, operator placement algorithms are leveraged,
which have been studies extensively by various authors. A
survey is available in [
Optimization in the area of data stream processing systems
is done for example to achieve an optimal overall
utilization [
An approach related to that in this paper has been studied
by Fernandez et al. in [
As of today, many data stream processing systems use replication to ensure high availability in case of host failures. However, to replicate a system completely, at least twice the resources have to be allocated, which is costly.
We proposed a heuristic replication approach enabling the use of remaining system resources to increase the availability of queries. An operator selection algorithm is used to determine a subset of operators for replication that are then placed via an operator placement algorithm. Moreover, we suggested a replication optimizer which allows users to guide the replication while trading o cost and availability. Finally, the challenges arising from a combination of these techniques with elastic data stream processing systems were discussed. Using heuristic replication as well as the replication optimizer with a system reacting on changes, e. g. in event rate and selectivities, demands the adaptation of the placement routines. However, the normalized importance functions as well as the optimization routines might be reused unchanged.
To validate the approaches we simulated a heuristic replication approach comprising operator selection as well as operator placement strategies. Given a set of remaining resources, the fault tolerance of complex event processing systems is improved. Choosing a proper heuristic can improve the availability two percentage points compared to a random operator selection. Compared to full replication only one percentage point is lost spending 50 % less resources for replication.
In the future, we plan to implement the proposed approaches with a state-of-the-art streaming system.