Data Stream Processing applications can process large data volumes in near real-time. In order to face varying workloads in a scalable and cost-effective manner, it is critical to adjust the application parallelism at run-time. We formulate the elasticity problem as a Markov Decision Process (MDP). As the MDP resolution requires full knowledge of the system dynamics, which is rarely available, we rely on model based Reinforcement Learning to improve the scaling policy at run-time. We show promising results even for a decentralized approach, compared to the optimal MDP solution.
New emerging application scenarios (e.g., social analytics, fraud detection, Smart City) leverage Data Stream Processing (DSP) to process data streams in near real-time. A DSP application is usually represented as a directed acyclic graph (DAG), with data sources and operators as vertices, and streams as edges [ ]. Each operator continuously receives data (e.g, tuples), applies a transformation, and generates new outgoing streams.
A commonly adopted DSP optimization is data parallelism, which consists of scaling-in/out the parallel instances of the operators, so that each processes a portion of the incoming data (at the cost of using more resources) [ ]. Due to the unpredictable and variable rate at which the sources produce the streams, a key feature for DSP systems is the capability of elastically adjusting the parallelism at run-time. Most of the existing DSP frameworks allow to allocate more than one replica per operator, but their support for the run-time reconfiguration is quite limited, as regards both the mechanisms and the policies.
In this paper, we focus on the auto-scaling policies. We formalize the elasticity problem for a DSP application as a Markov Decision Process (MDP), presenting both centralized and decentralized formulations. Unfortunately, in practice the optimal MDP policy cannot be determined, because several system dynamics may be unknown. To cope with the model uncertainty, we rely on Reinforcement Learning (RL) approaches, which learn the optimal MDP policy on-line by interacting with the system. Specifically, we present a model based RL solution that leverages the partial knowledge of the model to speedup the learning process.
Elasticity for DSP is attracting many research efforts [ ], with most approaches relying on heuristics to determine the scaling decisions. An optimization model that also considers the operator placement problem has been presented in [ ], but it cannot be easily solved in a decentralized manner. Here we describe a simpler model, for which we can derive a decentralized formulation. The application of RL techniques to DSP elasticity is quite limited. Heinze et al. [ ] propose a simple RL approach to control the system utilization, but they focus on infrastructure-level elasticity. Lombardi et al. [ ] exploit RL in their elasticity framework as well, but the learning algorithm is only used for thresholds tuning. In [ ] different RL algorithms have been compared for solving the elasticity problem for a single DSP operator in isolation, while in this work we consider whole applications.
In the rest of this paper, we first formulate the elasticity problem as an MDP in Sect. , presenting in Sect. the RL based algorithm for learning the scaling policy; we evaluate the proposed solutions in Sect. , and conclude in Sect. .
In this paper, we consider the elasticity problem for a DSP application composed of operators. Each operator is possibly replicated into a number of instances and, without lack of generality, we assume even distribution of the incoming data among the parallel instances. For each operator, an Operator Manager monitors the operator functionality, while an Application Manager supervises the whole application. The number of parallel instances used by each operator is adjusted either by its Operator Manager (decentralized adaptation) or by the Application Manager (centralized adaptation).
At each decision step, for each operator we can add an instance, terminate one, or keep the current parallelism. Following a scaling decision, the operator is subject to a reconfiguration process; as the integrity of the streams and the operator internal state must be preserved, the whole application is usually paused during the process, leading to downtime [ ]. Our goal is to take reconfiguration decisions as to minimize a long-term cost function which accounts for the downtime and for the monetary cost to run the application. The latter comprises (i) the cost of the instances allocated for the next decision period, and (ii) a penalty in case of a Service Level Agreement (SLA) violation. In particular, we consider a constraint on the application response time , so that a penalty is paid every time the response time exceeds a given threshold.
In order to keep the system model simple, we consider a deployment scenario with (i) homogeneous computing resources on which the operator instances are executed, and (ii) negligible communication latency between them. We defer to future work the extension of the model for a more realistic distributed setting.
We define the response time as the maximal source-sink total processing latency over the application DAG.
Model In the considered system, reconfiguration decisions are taken periodically. Therefore, we consider a slotted time system with fixed-length intervals of length
, with the -th time slot corresponding to the time interval [, ( + 1) ]. We denote by ,
∈ [1, at the beginning of slot for operator , and by , its average input rate measured during the previous slot. Additionally, we use to denote the overall application input rate (i.e., the total data sources emission rate). At the beginning of slot , a decision is made on whether reconfiguring each operator.
We first consider a centralized model, in which the reconfiguration decisions are taken by the Application Manager; then, at the end of the section, we consider the case in which the responsibility of making scaling decisions is decentralized, and each Operator Manager acts as an independent agent. In both the cases, we formalize the resulting problem as a discrete-time Markov Decision ] the number of parallel instances
Centralized Elasticity Problem An MDP is defined by a where is a finite set of states, ( ) a finite set of actions for each state , ( ′| , ) are the state transition probabilities, (, ) is the cost when action is executed in state , and ∈ [0, 1] is a future cost discounting factor.
We define the state of the system at time as = ( , 1, , 2, , . . . , , ). For the sake of analysis, we discretize the arrival rate by assuming that 0, ,¯. . . , ∈ { is ( ) = 1( ) × · · · × ¯} where ¯ is a suitable quantum. For each state , the action set ( ), where, for each operator , ( ) = { +1, − 1, 0} (except for the boundary cases with minimum or maximum replication).
System state transitions occur as a consequence of scaling decisions and arrival rate variations. It is easy to realize that the system dynamic comprises a stochastic component due to the exogenous rate variation, and a deterministic component due to the fact that, given action and the current number of instances, we can readily determine the next number of instances. An example of a system state -tuple ⟨ , , , , ⟩ transition is illustrated in Fig. .
To each state pair we associate a cost (, ) that captures the cost of operating the system in state and carrying out action , including: . the resource cost (, ), required for running ( + ) instances for each operator , assuming a fixed cost per instance; assuming a constant reconfiguration penalty; . the reconfiguration cost
( ), which accounts for the application downtime, . the SLA violation cost
(, ), which captures the penalty incurred whenever the response time (, ) violates the threshold .
We define the cost function (, ) as the weighted sum of the normalized terms: (, ) = where , ︀∑ =1
and + ,
Decentralized Elasticity Problem In the decentralized adaptation scenario, we assume that each Operator Manager independently acts on its associated operator, having only a local view of the system. We again rely on MDP to formalize the cost minimization problem for each agent (i.e., the Operator Managers). Omitting the reference to the specific operator, we define the state at time as the pair = ( , ), where is discretized using a suitable quantum for each operator. The action set is simply ( ) = { +1, − 1, 0} (except for the boundary cases with minimum or maximum replication).
Because the agents have not a global view of the application, they can only optimize local metrics, and thus we have to formulate a new local cost function ′(, ). We replace the SLA violation penalty with one based on the operator utilization (, ) and a target utilization ¯ . We get: ′(, ) = A policy is a function that associates each state with the action to choose. For a given policy , let ( ) be the value function, i.e., the expected infinite-horizon discounted cost starting from . It is also convenient to define the action-value function : ×
→ ℜ taking action in state and then following the policy :
which is the expected discounted cost achieved by (, ) = (, ) + ∑︁
( ′| , ) ( ′), ∀ ∈ ′∈
It is easy to realize that the value function and the function are closely related in that ( ′) = min ∈ ( ) ( ′, ), ∀ ∈ knowledge of the Q function is fundamental in that it directly provides the associated policy: for a given function , the corresponding policy is ( ) = . More importantly, the arg min ∈ ( ) (, ), ∀ ∈ satisfies the Bellman optimality equation:
. We search for the optimal MDP policy *, which *( ) = ∈ min ( ) ︃{ (, ) + ∑︁ ′∈ ( ′| , ) * ( ′) , ∀ ∈ ︃}
In the ideal situation, we have full knowledge of the system, and we can directly compute * using the Value Iteration algorithm [ ]. In more realistic cases, we have only partial knowledge of the underlying system model (e.g., the workload distribution is usually unknown). We can resort to Reinforcement Learning (RL) approaches, which are characterized by the basic principle of learning the optimal policy by direct interaction with the system. In particular, we consider a model based RL algorithm that, at each time step, improves its estimates of the unknown system parameters, and performs an iteration of the
( ) ( )
RL based Elastic Control Algorithm : Initialize the action-value function : loop : : : : : : : :
end for : end loop choose an action (based on current estimates of Q) observe the next state +1 and the incurred cost update the unknown system parameters estimates for all ∈
do end for for all ∈ ( ) do (, ) ← ^ (, ) + ∑︀ ′∈
^( ′| , ) min ′∈ ( ′) − 1( ′, ′)
We first consider the case in which the operator response time model is known, and let the algorithm learn the state transition probabilities.. In order to estimate ( ′| , ), it suffices to estimate the input rate transition probabilities [ +1 = ′| = ] , since the dynamics related to the number of instances are known and deterministic. Hereafter, since takes value in a discrete set, we will write , ′ = [ +1 = ′¯| = ¯], , ′ ∈ { number of times the arrival rate changes from state ¯ to ′¯, in the interval { 1, . . . , } , , ′ ∈ { 1, . . . , } . At time the transition probabilities estimates are 0, . . . , } for short. Let , ′ be the ︂̂ , ′ =
, ′ ︀∑ =0 ,
If we remove the assumption on the known response time model, we have to estimate the cost (, ) as well, because we cannot predict the SLA/utilization violation any more. So, we split (, ) and ′(, ), respectively defined in ( ) and ( ), into known and unknown terms: the known term (, ) accounts for the reconfiguration cost and the resources cost, whereas the unknown cost (, ) represents the SLA (or utilization) violation penalty. We use a simple exponential weighted average for estimating the unknown cost: ^, (, ) ←
(1 − )^, − 1(, ) + , where , = (or
) if a violation occured a time and 0 otherwise.
As regards the complexity of the algorithm, the size of the state-action space is critical, since each learning iteration requires ( that in the centralized model | and | | 2 | operators , whereas they are not influenced by in the decentralized model. 2) operations. We observe grow exponentially with the number of
We evaluate by simulation the presented models, and compare the policies learned through RL to the MDP optimal one. In order to explicitly solve the MDP, we need a state transition probability matrix, which is not available in practical scenarios. Thus, for evaluation, we consider a dataset made available by Chris
For simplicity, we consider a pipeline application, composed of a data source and up to operators. Each operator runs at most = 5 instances, each behaving as a M/D/ queue with service rate . For evaluation, we consider a scenario with slightly different service rates, and set 1 = 3.7, 2 = 4 = 3.3, and 3 = 2.7 tuple/s. Because of space limitation, we defer the evaluation of real world topologies to future work. We consider = 1 min, and aggregate the events with ¯ = 20 tuple/min. For the cost function, we set in the dataset over one minute windows. We assume = , , ∀ , discretized = 0.4, = = = 0.2,
= 650 ms, and ¯ ∈ { 0.6, 0.7, 0.8} . As regards the learning To simplify notation, we simply use to denote the input rate. In the centralized model, we use the same estimates for the total application input rate . http://chriswhong.com/open-data/foil_nyc_taxi/ ( ) ( ) algorithm, we set = 0.99, and = 0.1. We compare the results obtained by solving the MDP to those achieved by the centralized RL algorithm (with and without the known response time model) and by the decentralized solution.
In Table we report the results for a -operators topology. As expected, the minimum average cost is achieved solving the MDP; interestingly, the centralized RL solution incurs almost negligible performance degradation, and the gap with the decentralized approach is not significant as well. However, we note that the performance of the decentralized solution depends on the target utilization ¯ , which has still to be set manually in our work. Setting a too high (or too low) value results in a different trade-off between SLA violations and used instances, with negative effects on the overall cost as well. The decentralized solution shows a higher number of reconfigurations, due to the lack of coordination between the agents. As illustrated in Fig. a, the convergence velocity of the different solutions is similar, except for the centralized RL algorithm. In absence of the response time model, the algorithm is indeed significantly slower to learn than the other solutions. When the response time model is known, the algorithm converges much faster, despite the large state-action space.
We also compare the decentralized approach to the MDP varying the number of operators in the application. As shown in Fig. b, the cost gap between the two solutions slightly increases as the application gets more complex. However, we observe that the decentralized algorithm has not scalability issues as the number of operators increases, while solving a centralized problem gets easily impractical.
In this paper we have formalized the elasticity problem for DSP applications as a Markov Decision Process, and proposed a Reinforcement Learning based solution to cope with the limited knowledge of the system model. Our numerical evaluation shows promising results even for a fully decentralized solution which, leveraging the available knowledge about the system, does not suffer from the extremely slow convergence of model-free RL algorithms. In practical scenarios, we could also combine the proposed solution with a simple threshold-based policy 0.5 0.4 0.2 0.1 1
MDP Centr. RL+
Centr. RL Decentr. RL (U=0.7) 7 (a) 0.18 0.16 (b) 4 to be used at the beginning, while the agents learn a good policy to be adopted in the following.
For future work, our goal is twofold. We plan to improve the decentralized learning algorithm exploring RL techniques specifically targeted to multi-agent systems. At the same time, we will extend the model to cope with a more complex and realistic scenario, considering, e.g., resource heterogeneity and network latency in distributed deployments.