Stream processing is a popular paradigm to process huge amounts of data. During processing, the actual characteristics of the analyzed data streams may vary, e.g., in terms of volume or velocity. To provide a steady quality of the analysis results, runtime adaptation of the data processing is desirable. While several techniques for changing data stream processing at runtime do exist, one speci c challenge is to minimize the impact of runtime adaptation on the data processing, in particular for real-time data analytics. In this paper, we focus on the runtime switching among alternative distributed algorithms as a means for adapting complex data stream processing tasks. We present an approach, which combines stream re-routing with bu ering and stream synchronization to reduce the impact on the data streams. Finally, we analyze and discuss our approach in terms of a quantitative evaluation. Data stream processing; runtime adaptation; impact-minimizing adaptation enactment; algorithm switching
Big data applications aim at processing huge or complex
data sets, which usually cannot be handled by traditional
approaches. Distributed stream processing [
While approaches such as Borealis [
One speci c way of adapting stream processing is to switch
among di erent algorithms, which provide similar
functionality but operate at di erent runtime characteristics [
While adaptation for algorithms running on single nodes can be achieved with existing techniques, the research question in this paper is How to realize an impact-minimizing runtime switching among alternative distributed (real-time) data stream processing algorithms? At a glance, switching an algorithm at runtime may appear rather similar to adapting the query plan in known approaches, but, in contrast, we aim at arbitrary, distributed, potentially stateful data analysis algorithms. Our contribution is an approach for runtime switching among distributed data analysis algorithms, which aims at minimizing the impact on the data streams that adaptation can cause, such as missing, repeated, disordered or (massively) delayed data items. Therefore, our approach considers the memory state of the involved algorithms and synchronizes the (common) output stream. We show that our approach can perform a runtime switch in 60110 ms depending on the setting and that the output data stream is not seriously a ected. In this paper, we focus on switching techniques and, therefore, making the actual decision for adapting the data processing is out of the scope and subject to future work.
The work we presented in this paper has been performed in the EU-funded project QualiMaster3, which aims at building a real-time adaptive data processing infrastructure. For demonstration, the project creates algorithms and applica1http://storm.apache.org/ 2http://spark.apache.org/ 3http://qualimaster.eu tions for analyzing systemic market risks. In this context, we apply our approach to enable the setting described above, i.e., to dynamically switch from a software-based execution to a hardware co-processor wrapped by a distributed software algorithm, e.g., to cope with dynamic load changes.
The remainder of this paper is structured as follows: In Section 2, we discuss related work. In Section 3, we introduce our approach in terms of two variants, a simple one as basis for later analysis and an advanced runtime switching including state maintenance and stream synchronization. We evaluate both variants empirically in Section 4. Finally, in Section 5 we conclude and provide outlook on future work.
In this section we discuss related work in terms of adaptive stream processing engines and mechanisms for adapting the processing. For minimizing the impact, it is important that the target algorithm takes over the processing as soon as possible and that the produced output stream does not have duplicated or missing items (in our context even the sequence of output items is not changed). Therefore, we also discuss work on migrating the memory state of algorithms at runtime and on synchronizing (output) data streams.
Several adaptive stream processing engines are
described in literature, for example Borealis [
Di erent mechanisms to adapt stream processing
are proposed in the literature. We now provide an overview
of the most closely related mechanisms. Data admission
(also called load shedding) [
When switching among stateful algorithms, it is
important to enable the target algorithm to take over processing
at a certain point in time. Therefore, the memory state
must either be built up or it must already be up-to-date.
One type of strategies is based on running operators or new
query plans on demand in parallel to build up the required
state, e.g, the parallel track strategy in [
When the actual runtime switch happens, in a short time
frame data items may be duplicated or items may even be
missing without further consideration. This can be
mitigated by explicitly synchronizing the output stream.
Some synchronization approaches are discussed in literature,
typically relying on some notion of a synchronized (wall)
time. For example, in media stream processing, Rothermel
et al. [
In summary, to our very best knowledge there are no approaches on dynamically switching among arbitrary data stream processing algorithms, which explicitly aim at minimizing the impact on the involved data streams. 3.
In this section, we describe our approach to the runtime switching among distributed data stream analysis algorithms at low impact. More speci cally, we introduce rst some terms and discuss requirements characterizing the potential impact on the data streams. Then, we present two approaches: In Section 3.1, we discuss simple algorithm switching, which we use as a basis for the evaluation in Section 4. In Section 3.2 we present an advanced approach considering memory state and stream synchronization.
Dynamic algorithm switching performs at runtime a switch
among alternative algorithms, e.g., from an algorithm
family of functional similar algorithms having di erent runtime
trade-o s [
R2. Transparency: Although stream processing systems
shall be resilient against stream imperfections [
R3. Minimizing the switching time: Switching algorithms at runtime shall happen as fast as possible to reduce the time for causing disturbances to the streams. R4. Minimizing e ects on stream characteristics: In addition to R3, also the e ect on further (application-) relevant stream characteristics such as latency, throughput or volatility shall be minimized.
In this paper, we focus on throughput as one particular stream characteristic (R4). Other characteristics as well as non-functional aspects such as resource consumption are out of scope of this paper due to space limitations. From a technical point of view, we require that alternative algorithms share the same input and output (item) types, respectively. Further, we assume that the processing environment is robust, i.e., tuples are processed without failures. 3.1
We introduce now a simple approach for switching among alternative distributed algorithms, which basically relies on re-routing the input stream at runtime. We use this approach as a baseline in our analysis in Section 4.
Figure 1 illustrates the approach in terms of a data ow diagram, i.e., nodes represent (distributable) data processors and edges the data ow. Without loss of generality, we illustrate our approaches using just two alternative algorithms. A setting with multiple alternatives can be constructed similarly. In Figure 1, the processors P1,1 to P1,n constitute Algorithm1 (akin P2,1 to P2,m for Algorithm2), whereby the data ow within the respective algorithm is not relevant for our discussion. In addition, two guarding processors control input (Switching Element ) and output (Join Element ) streams. Actually, these guards can be realized as part of proceeding or succeeding processors to save resources. 1 switch Switching Element √ 2 2
Let us assume that Algorithm1 is the currently active algorithm. A signal, i.e., an asynchronous event sent to a processor, indicates the actual need to switch the active algorithm to another target algorithm 1 . In the simple approach, the switch signal causes an immediate re-routing of the data streams 2 , i.e., the stream to Algorithm1 is disabled and the alternative stream to Algorithm2 is enabled at the same time. No action is needed in Join Element .
Actually, the simple approach does not consider queuing e ects in the active algorithm, which can cause the algorithm to continue processing even after the switch. Thus, this approach can signi cantly increase the switching time and lead to inconsistent results as the processing of both algorithms can overlap in an uncontrolled fashion. In summary, the simple approach likely fails regarding R1 - R4. 3.2
In this section, we discuss an advanced approach, which aims at minimizing the impact on the underlying data streams through realizing the requirements R1 - R4. The core idea of this approach is to combine four techniques into a single approach, namely:
Output stream control to disable the output stream of the actual algorithm upon switch (R3, R4) and to avoid accidental data overlap (R1).
Acknowledgements and bu er transfer to track and stop processing in the actual algorithm during the switch (R1 - R3). Therefore, we (conceptually) send an acknowledgment signal for each item processed by an algorithm to its input bu er, so that incompletely processed items can be identi ed and transferred to the target algorithm. As the transfer time depends on the amount of bu ered items, we aim at an improved mechanism in future work.
State maintenance to prepare the state of the target
algorithm so that it can safely take over processing at
a distinct point in time (R3, R4). Currently, we rely
on a variant of the parallel track strategy [
Output synchronization based on unique item identi ers to ensure the output item sequence and, thus, transparency according to R2. This is important, as the involved algorithms may process data with a different latency.
Combining these four techniques in a distributed processing environment requires the exchange of various types of signals. Figure 2 a) illustrates the design of the advanced approach. Basically, the advanced approach is an extension of the simple approach in Section 3.1, and also includes two alternative algorithms and two guarding processors. For enabling acknowledgment and bu er transfer, we equip both algorithms with an entrance queue. Conceptually, we represent this as two individual processors (Intermediary1 and 1 switch Switching Element 2 √ 3 passivate 9 synchronized 7 X
6 5 transferred headId lastProcessedId
a) Intermediary 1 ack emit P2,1 ack ... 8 ... Intermediary2) so that the queues can be maintained independently. A speci c implementation may also be based on the queues of P1,1 or P2,1, respectively. The queues in the intermediary processors remove an item only when it is fully processed by the last processing node of the respective algorithm. This is indicated by an acknowledgment signal (ack in Figure 2) sent by P1,n or P2,m, i.e., the last node of the respective algorithm. Thus, the items in the queues are either pending to be processed or emitted to the respective algorithm but not fully processed yet. Moreover, we utilize a further queue in the Switching Element to control the overall stream ingestion.
Let again Algorithm1 be the currently active algorithm and let t0 be the point in time when Switching Element receives the switch signal 1 . In addition to the overall design depicted in Figure 2 a), we illustrate the individual phases of the approach in the timing diagram in Figure 2 b). The switch signal initiates the runtime algorithm switching. First we warm-up Algorithm2, i.e., we run Algorithm2 in parallel to build up its state. Therefore, we activate the stream to Algorithm2 by duplicating the input items in Switching Element 2 . At the end, Join Element passes only the items of the active algorithm, i.e., it discards the output of the passive algorithms, in particular the output of Algorithm2 in the warm-up phase. Now, both algorithms process the input stream in parallel for tm.
The actual switch happens at t0+ tm as indicated in Figure 2 b) by performing the output synchronization, i.e., Intermediary1 negotiates with Intermediary2 the last processed item in Algorithm1 as a basis for the queue transfer. During switching, each item is queued in the intermediary processors along with a sequential identi er indicating the arrival order of the items. Let lastProcessedId be the identier of the last item emitted to Algorithm1. However, ongoing data processing during the synchronization may invalidate lastProcessedId . Therefore, we rst passivate 3 both algorithms during the synchronization and disable 4 also the output of results in P1,n to avoid that acknowledgment signals disturb the synchronization. P1,n con rms the passivation (not shown on Figure 2) so that Intermediary1 can now send lastProcessedId to Intermediary2 5 . We denote the time needed for synchronization as ts.
Let headId be the identi er of the head of the queue in Intermediary2. As the involved algorithms may operate at a di erent speed, we must consider three cases for the synchronization (illustrated as queues for the intermediary processors in Figure 3): a) If headId = lastProcessedId , both algorithms are running at the same speed. No items must be transferred and Algorithm2 can immediately take over the processing from Algorithm1 as shown in Figure 3 a). b) If headId < lastProcessedId then Algorithm1 is faster than Algorithm2. In this case, no items must be transferred, but the items [headId , lastProcessedId ] must be skipped as they would cause duplicated results. For example, in Figure 3 b) items [499, 500] have been processed and must to be skipped. c) If headId > lastProcessedId , items (lastProcessedId , headId ) must be transferred to Intermediary2. Therefore, Intermediary2 sends the headId to Intermediary1 6 and initiates the queue transfer, i.e., Intermediary1 sends unprocessed items via network to Intermediary2. Let tt be the transfer time. At tt = t0+ tm+ ts+ tt, Intermediary2 is noti ed about the end of the queue transfer 7 to prepare for regular items from Switching Element . In the example in Figure 3 c), the lower queue is processed faster than the upper queue. So the items (500, 503) must be transferred to avoid a gap.
Due to the ack signals from P2,m, Intermediary2 can track whether all data items for warm-up have been processed. As soon as synchronized items are passed to Algorithm2, Intermediary2 sends an emit signal 8 to P2,m enabling the lastProcessedId
lastProcessedId a) output of processed data to Join Element . To minimize the switching time, Algorithm2 processes synchronized items in parallel to the queue transfer, i.e., it starts processing real data already during queue transfer. Finally, P2,m con rms the activation of the output stream (not shown in Figure 2). In turn, Intermediary2 noti es Switching Element about the end of the synchronization 9 as well as that Algorithm2 took over the processing and Algorithm1 is discarded.
As discussed, the advanced approach uses a combination of signals, queue transfer, state warm-up and output synchronization to reduce the impact of switching among distributed algorithms at runtime. In this approach, the critical period is ts when synchronization happens and no processing takes place. During queue transfer, Algorithm2 can already take over processing. Due to the warm-up phase, the overall switching time is at least of tm+ ts, but the queue transfer may cause a peak load of items to be processed. However, during the warm-up phase, we utilize the processing resources for two algorithms in parallel. As mentioned already above, optimizing the resource consumption is out of the scope of this paper.
Realizing such a complex signal-based approach requires adequate support for the data analyst, e.g., a framework or code generation as we do in the QualiMaster project.
We present now an evaluation of the approaches discussed in Section 3. The goal of the evaluation is to analyze and validate the actual impact of the proposed approaches with respect to requirements R1 - R4. Therefore, we evaluate in this section both, the simple approach as a baseline as well as the advanced approach. In particular, we focus on the impact on the data streams, i.e., the time for performing an algorithm switch (R3) and, as mentioned above, the throughput as a measure of (R4). In addition, we analyze the realization of R1 and R2. In this section, we discuss rst the environment, the implementation of the approaches and the settings we used for conducting the evaluation. Then, in Section 4.1 we evaluate the simple approach and the advanced approach in Section 4.2.
Environment. We conduct our experiments on a Storm (version 0.9.5) cluster composed of one Nimbus machine for managing topologies, 6 worker machines for distributed processing and three Zookeeper instances (version 3.4.6) for managing the workers. Each machine is equipped with an Intel(R) Core(TM)2, 1.86GHz CPU and runs Ubuntu 12.04.4 LTS as well as Java JDK version 1.7.0 71. All machines in the cluster are connected over a dedicated network switch via Gigabit-Ethernet at a transfer rate of around 100MByte/s. In addition, we synchronize the system clocks of all machines using the Network Time Protocol (NTP) in order to enable comparisons of the timestamp-based logs written during the experiments. Due to time synchronization, the actual time di erence among the machines is less than 3 ms.
Implementation. We implemented both approaches discussed in Section 3 as of Storm topologies consisting of processors called Bolts and data sources called Spouts. The Storm mechanisms for guaranteeing the processing of each item already provide us with an implementation of the acknowledgment signals. We realized the other signals in terms of Zookeeper change noti cations, i.e., when data is written by the sender into a dedicated Zookeeper node, the receiver is noti ed by the Zookeepers about the change and can react on the signal data. All processors discussed in Section 3 except for the intermediary ones are implemented as Bolts. As Bolts work item-wise and Spouts operate asynchronously and their internal queues are not accessible, we realize the intermediary processors in the advanced approach as Spouts with explicit queues. In more detail, we use a dedicated input queue receiving input items and an output queue temporarily storing sent and pending items, removing them upon acknowledgement signals. The connection between the Switching Element and the intermediate processors are realized as network connections. For the experiments, a single Spout realizes the stream ingestion and produces sequential integer items at a con gurable rate. For achieving repeatable experiments, we use simple distributed algorithms, which consist of a processor that just passes the received input data. To simulate performance di erences, the algorithms can be con gured with a certain latency (delay) per item. For obtaining experimental data, a processor records the individual arrival time per item in a processorspeci c log le. Running such a topology without delays in the algorithms and no switching shows that logging does not signi cantly in uence the nominal throughput.
Setup. In our experiments, we use a constant ingestion frequency of 1000 items per second. Algorithm1 is the active algorithm at the beginning of each experiment. For both approaches, we perform two experiments, one with delayfree (near-ideal real-time) algorithms and one to analyze the behavior at a certain load. More speci cally, in the load situations Algorithm1 acts as a slow algorithm with a delay of 30ms per item. Based on the window time of a speci c scenario in our project context, we use 30s as warm-up time in this evaluation. The experiments are implemented as Java programs managing the topologies and issuing the switch signal so that the experiments become repeatable.
We discuss now the results of evaluating the simple and the advanced approach. For each experiment, we repeatedly run the respective experiment implementation at least ve times in order to detect deviations. While we identi ed and xed deviations in pre-experiments, there were no signi cant deviations in the nal experiments, so that we report results of typical executions in this paper. Although the log les written during the experiments are recoded in terms of milliseconds and the analyses were done based on these logs, we illustrate below the results by overview gures in terms of seconds time unit. 4.1
In this section, we discuss the evaluation of the simple approach presented in Section 3.1. As the best case, both algorithm runs without delay, i.e., they do not queue data. In this case, our experiments show no negative impact on the throughput. However, the more interesting case includes an algorithm that processes at a certain latency so that queuing e ects can occur.
Figure 4 illustrates the throughput, i.e., the number of data items per second logged in the Join Element . In the beginning, the data processing is running on the slower Algorithm1. At 1 , we send the switch signal and cause a rerouting of the data streams from Algorithm1 to Algorithm2. As depicted in Figure 4, the throughput after the switch uctuates for around 15 seconds. To explain this e ect, we analyzed the logs written by our implementation of the processors. By comparing the timestamps of the items, we realized that the increasing throughput at 2 is due to the queuing of items in the preceding Switching Element caused by the slow algorithm. Instead of coping with the queued items immediately, in this experiment, the processing of the queued items is deferred by 3 seconds. Further, the reduced throughput at 3 is caused by the queuing of items in the (internal queues of) Algorithm1 due to its con gured latency. The queuing in Algorithm1 keeps its processors still active as indicated by the timing bars in Figure 4 although the data stream is already processed by Algorithm2. Thus, the simple approach signi cantly increases the switching time until all items queued in the Algorithm1 are processed. In this experiment, this leads to inconsistent results as both algorithms overlap processing after the switch. The logs of Algorithm1 and Algorithm2 indicate that 1427 out-of-order items are caused by the ongoing processing in Algorithm1, which a ects the overall output accuracy. As a result, we conclude that the simple approach does not ful ll R1 - R4. 4.2
In this section, we evaluate the advanced approach presented in Section 3.2 in terms of both, switching among the same speed algorithms as well as from a slower algorithm.
Figure 5 depicts the throughput while switching among algorithms operating at no delay. After the actual switch signal arrives at 1 , both algorithms run in parallel for the warm-up phase until 2 . As items are processed by both algorithms at the same speed, no items are transferred (the rst case in Section 3.2) as indicated by the experiment logs. active algorithm 1 2
At 2 , Algorithm1 is terminated and Algorithm2 takes over the processing. Based on the experiment logs, we also found that actually no items are synchronized. Further, on the output steam neither items are missing nor duplicated (R1). Moreover, the output item sequence corresponds rst to the sequence produced by Algorithm1 and, after the switch, by Algorithm2 (R2). The actual impact is a short reduction of the throughput to 950 items, which is caused by the output synchronization. In this experiment, we identi ed the actual switching time as 60ms from the logs. This is reasonable as 6 signals are exchanged and in our cluster the time between sending and receiving an event is in average around 10 ms.
Figure 6 illustrates the throughput while switching from a slower algorithm at 30 ms delay to a faster one. The actual switch signal arrives at 1 and again both algorithms run in parallel until 2 , the end of the warm-up phase of 30s. Please note that during the warm-up phase the output is still produced by the slower Algorithm1. The output synchronization starts at 2 . Switching from a slower algorithm, i.e., the third synchronization case discussed in Section 3.2, leads to a queue transfer from Algorithm1 to Algorithm2. In this experiment, 2889 items are transferred and the total transfer takes tt=820 ms as indicated by the experiment logs. As discussed in Section 3.2, Algorithm2 starts processing data in parallel to the queue transfer so that the switching time is not dominated by tt. However, several transferred items arrive in short time at Algorithm2 and lead to a throughput peak 3 . As we analyzed from the logs, in this experiment the actual switching time takes 106 ms, mostly due to the 9 signals (each taking around 10 ms) sent during the switch. 5.
Distributed data stream processing is a popular approach to realize Big Data Applications. While processing data streams, the actual stream characteristics may change dramatically, e.g., when sentiments change in the Social web or when stock markets become hectic. To cope with such changes during processing and to provide a steady output quality, data processing must adapt to the actual context.
In this paper, we discussed the runtime switching among (distributed) data processing algorithms as one speci c form of realizing adaptive stream processing. We introduced an advanced switching approach, which takes queuing e ects into account, maintains the output sequence, and, to reduce the switching time, utilizes parallel track processing to warm-up the target algorithm. We implemented both approaches on Apache Storm, performed experiments and analyzed the results. As expected, plain stream re-routing su ers from queuing e ects, item duplication and requires an overall switching time of more than 15 s. In contrast, on our cluster, the advanced approach reduces the switching time to 60 ms (algorithms with no latency) or less then 110 ms (at 30 ms latency) without disturbing the output sequence. While e ects regarding timeliness occur due to the processing of queued items, we showed the e ectiveness of our approach in reducing the impact on the data streams.
Currently, we integrate the advanced approach into the model-based generation of topologies developed in the QualiMaster project to unburden the data analyst from manually implementing e cient adaptation that require queuing and complex signal interaction. Ultimately, this will include the generation of transparent integrations of hardware co-processors. Here, initial results in combination with dynamic switching among algorithms are promising.
In the future, we plan to improve our approach, in particular to speed up the events, to limit the queue transfer time and to research state transfer mechanisms that do not need a parallel warm-up, in particular for switching between software-based processing and hardware co-processors. Furthermore, we aim at taking further stream characteristics into account as well as improving the overall resource usage, e.g., by allocating resources only for active algorithms and, if needed, during the warm-up phase. Moreover, we will consider the impact of switching on the result quality through quality measures and consider how processing failues can be handled to guarantee robustness.
This work was partially supported by the European Commission in the 7th framework programme through the QualiMaster project (grant 619525).