Keywords: Stream processing, Adaptive SPS, Predictive algorithm, Reactive algorithm, Google Cloud Platform. vi Flux de données, Traitement en temps réel, Algorithme predictive, Algorithme reactive

The amount of data produced by today's web-based systems and applications increases rapidly, due to the many interactions with users (e.g. real-time stock market transactions, multiplayer games, streaming data produced by Twitter, etc.). As a result, there is a growing demand, particularly in the fields of commerce, security and research, for systems capable of processing this data in real time and providing useful information in a short space of time. Stream processing systems (SPS) meet these needs and have been widely used for this purpose. The aim of SPSs is to process large volumes of data in real time by housing a set of operators in applications based on Directed acyclic graphs (DAG).

Most existing SPSs, such as Flink or Storm, are configured prior to deployment, usually defining the DAG and the number of operator replicas in advance. Overestimating the number of replicas can lead to a waste of allocated resources. On the other hand, depending on interaction with the environment, the rate of input data can fluctuate dynamically and, as a result, operators can become overloaded, leading to a degradation in system performance. These SPSs are not capable of dynamically adapting to operator workload and input rate variations. One solution to this problem is to dynamically increase the number of resources, physical or logical, allocated to the SPS when the processing demand of one or more operators increases.

This thesis presents two SPSs, RA-SPS and PA-SPS, reactive and predictive approach respectively, for dynamically modifying the number of operator replicas. The reactive approach relies on the current state of operators computed on multiple metrics, while the predictive model is based on input rate variation, operator execution time, and queued events. The two SPSs extend Storm SPS to dynamically reconfigure the number of copies without having to downtime the application. They also implement a load balancer that distributes incoming events fairly among operator replicas.

Experiments on the Google Cloud Platform (GCP) were carried out with applications that process Twitter data, DNS traffic, or logs traces. Performance was evaluated with different configurations and the results were compared with those of running the same applications on the original Storm as well as with state-of-the-art work such as SPS DABS-Storm, which also adapt the number of replicas. The comparison v shows that both RA-SPS and PA-SPS can significantly improve the number of events processed, while reducing costs.

The first thing I would like to say is: Gracias a la vida, que me ha dado tanto. I am thankful to have closed this cycle of my life, from which I have learned both professionally and personally. I am thankful to have been surrounded by wonderful people, who gave me the energy to move forward. I am thankful to have lived long enough to be here and to be able to look at the sky. I thank you from the bottom of my heart.

I would like to thank my teachers, I feel very fortunate to have worked with you. Pierre, I thank you for all the confidence you have had in my project, my ideas, and my goals. Without your support I could not have come this far, and it is partly thanks to you. Merci beaucoup. Luciana, I thank you for all the love and dedication you have shown me, I will never forget all those details. Without your welcome my Latin heart would be colder, and with you I have felt this warmth that makes me feel at home. Muito obrigado. I would also like to thank Nicolás, because you have been an important person in this journey. Without you, this cycle could not have started, and likewise, it could not have ended. So many conversations we had, which gave me the strength to keep going when I felt stuck. Muchas gracias.

I would like to thank everyone who welcomed me at the Sorbonne Université, because each one contributed to the progress of this journey. So many people who trusted me to start on my teaching path, who gave me the wings to begin a great challenge. Thank you so much. I would like to thank everyone who has welcomed me in Paris. Although many people come and go, each one will have a part of my heart, and I am grateful for the love they have given me. Without these people, everything would be more difficult.

I would like to thank my family. Alba, quién sabe en qué momento de tu vida estés leyendo esto o probablemente ya te lo habré leído, pero quiero que sepas lo agradecido y orgulloso que estoy de ser tu papá, has sido una hija maravillosa, quién me ha brindado su amor incondicional. Cami, te quiero agradecer por haber emprendido este viaje a mi lado, has sido una de las grandes artífices de esta aventura, la cual ha culminado gracias al cariño y apoyo que me has dado, sin dudas todo sería distinto sin ti. Papá, te quiero agradecer todo el apoyo y confianza que iii has tenido conmigo, desde que emprendí este viaje siento que hemos pasado un período maravilloso juntos. Sofi, Amandi, Iñaki y Victor, gracias por todo el cariño vertido todas las veces que los visito en nuestra querida patria. Mamá, ya no puedo agradecerte, lamentablemente ya no estás acá, pero te dedico estas palabras de agradecimiento por haberme hecho quién soy. Gracias totales, les amo con todo mi corazón.

Thanks to the Chilean government. This thesis was funded by the National Agency for Research and Development National Agency for Research and Development (ANID) / Scholarship Program / DOCTORADO BECAS CHILE / 2018 -72190551. I want to thank the project ANID FONDECYT Nº11190314, Chile. This material is based upon work supported by the Google Cloud Research Credits program with the award GCP19980904.

1

The tremendous volume of information in the Internet started with Web 2.0 development. Web 2.0 established a paradigm shift where users actively participate and share data through applications such as blogs, social networks, or other web applications [START_REF] Newman | Web 2.0-The past and the future[END_REF]. Since the amount of exchanged information grows everyday faster, data processing becomes even more challenging. Data analysis for extracting information may be defiant. Such a task is even more complex if the analysis should take place in real-time. Under such a scenario, traditional processing systems based on the Batch processing paradigm like MapReduce [START_REF] Dean | MapReduce: simplified data processing on large clusters[END_REF][START_REF] Casado | Emerging trends and technologies in big data processing[END_REF] are not suitable for carrying out the analysis. Figure 1.2 shows a data pipeline using Batch processing, where events are read to be processed periodically according to a time window, and, therefore, report are not generated online.

Sustained on the need to process interactions in an online manner, different processing systems have been proposed in the literature. These systems are capable of dealing with unbounded sequences of events and deliver low-latency data. Nowa- days, most users require quick and updated information online as a support for decision-making [START_REF] Özal | Real-time route planning with stream processing systems: a case study for the city of Lucerne[END_REF].

Recorded Events

A good example is social network analysis under post-disaster scenarios, where events are continuously generated, processing them as close to real-time as possible is necessary to obtain timely information to mitigate the disaster effects [START_REF] Wladdimiro | Disaster management platform to support real-time analytics[END_REF].

With such information, areas may be prioritized, resource distribution can be improved, people searches may be more efficient, among others.

Another application is stock exchange prediction [START_REF] Ding | Stock Technical Analysis System Based on Real-Time Stream Processing[END_REF]. In this case, processing systems may analyze data and create mathematical models to predict the next day's market behavior.

In network security scenarios it is possible to monitor the network activity [START_REF] Zhao | A high throughput distributed log stream processing system for network security analysis[END_REF].

As the data is processed in real-time, it may help to detect 'on the fly' malicious behaviors. A similar approach is followed on logs analysis. Online data processing makes it possible to detect bugs or errors, as well as to see if there is any intruder or system policy violation.

Stream Processing Systems (SPS) were specially designed to fulfill these needs [START_REF] Leibiusky | Getting Started with Storm -Continuous Streaming Computation with Twitter's Cluster Technology[END_REF]. The goal of a SPS is to process unbounded continuous flows of events and to provide a scalable and efficient tool to process data close to real-time. Figure 1.3 presents an example of a stream processing pipeline. Events are processed on the fly and aggregated results are stored in a database or presented on a dashboard.

SPSs are based on directed acyclic graphs (DAG) whose vertices and unidirectional edges correspond to operators and event data flows respectively [START_REF] Chakravarthy | Stream Data Processing: A Quality of Service Perspective -Modeling, Scheduling, Load Shedding, and Complex Event Processing[END_REF]. An external source continuously provides the data that the system consumes. Light programming tasks (like filters, counters, storage, etc.) are handled by operators that quickly and in pipeline-style process the data (events). In a processing infrastructure (e.g. clusters, clouds, etc.), resources (e.g., VMs) are allocated to execute the operators which are frequently replicated for performance reasons. Figure 1. 4 shows an example of a SPS application, whose pipeline is composed of one input data and four operators. In this example, each operator processes the events according to the pipeline flow, provided by the input data. SPS frameworks, such as Storm [START_REF] Leibiusky | Getting Started with Storm -Continuous Streaming Computation with Twitter's Cluster Technology[END_REF] or Flink [START_REF] Carbone | Apache Flink™: Stream and Batch Processing in a Single Engine[END_REF], are configured before their deployment, usually defining both the DAG and the number of operator replicas beforehand. There are works that adapt the number of replicas dynamically based on analysis of input fluctuation [START_REF] Cardellini | Decentralized self-adaptation for elastic Data Stream Processing[END_REF][START_REF] Hamidreza Arkian | Modelbased Stream Processing Auto-scaling in Geo-Distributed Environments[END_REF][START_REF] Kotto Kombi | DABS-Storm: A Data-Aware Approach for Elastic Stream Processing[END_REF]. However, replica reconfiguration usually requires stopping the application and restating it. For adapting the number of replicas, there exist basically two approaches. The reactive approach considers the state of the system, while the predictive analyses the history of the states.

Split

Another critical feature of SPSs is the grouping strategy which is responsible for sharing the input load among operators' replicas. For instance, the round-robin policy is one of the most common approaches to implementing shuffle grouping. However, the latter does not consider the input load of each replica, i.e., current queued events waiting to be processed.

The behaviour of data streams can be quite dynamic. Traffic behaviour may suddenly increase/decrease, and event distributions may be unbalanced, among others. Such a dynamics may create struggle operators, increasing end-to-end latency and message loss. To solve this problem it is necessary to dynamically adapt the processing logic of the SPS, thereby increasing or decreasing the number of replicas of an operator, so a good adaptation decision may minimize the aforementioned problem: the latency and message loss can be reduced reaching higher accuracy for the data analysed.

For this reason, one solution is to propose an adaptive SPS, which characterize system utilization requirements and automatically increase/decrease the number of replicas of critical operators to keep latency restrictions and reduce event loss.

Contributions

This thesis proposes two adaptive SPSs (RA-SPS and PA-SPS) based on Storm [START_REF] Toshniwal | Storm@ Twitter[END_REF]. Its aim is to adapt the number of replicas of the operators according to the sparks in the data stream. Both adaptive SPS use the MAPE model [START_REF]An Architectural Blueprint for Autonomic Computing[END_REF], which consists of a control loop composed of four modules: Monitoring, Analysis, Planning, and Execution.

RA-SPS and PA-SPS exploit a reactive and predictive approach respectively. RA-SPS bases its decision on the state of an operator at runtime, according to the analysis of traffic peaks in short periods of time. PA-SPS bases its decision on the behaviour of the input data and the number of operator active replicas required for processing the input data, finding patterns in the traffic to anticipate possible overloads, or underloads in the SPS.

In order to cope with stream fluctuation both SPS, use a pool of operator replica, created beforehand. These replicas may be either active or inactive. Active replicas are deployed by the Storm scheduler. On the other hand, inactive replicas are idle, so they do not consume CPU resources but can be allocated as needed. Consequently, a pool of replicas can be dynamically activated (resp., deactivated) when the system detects the need for increasing (resp., decreasing) the number of active replicas of an operator. Note that under this model, the stream processing systems can adapt without stopping the SPS.

Another feature introduced in the proposed SPSs is the load-aware grouping that partitions the stream among replicas based on their respective current load.

RA-SPS dynamically modify the number of replicas per operator based on a multimetric. The multi-metric is composed of system statistics such as the queue size, the average execution time of an event, and the utilisation rate of an operator. It determines the state of the operators during a time interval, and depending on the state of the operator and conditions of the SPS, it is decided whether or not to modify the number of replicas.

PA-SPS dynamically allocates the number of replicas per operator necessary to process the input stream, defining the events that each operator O should process within a time interval. In order to predict the input stream and analyse the system's potential future behaviour, a predictive model is proposed, integrated to PA-SPS. By considering both (1) the events sent to an O by its direct operator predecessors as well as those from earlier time intervals that O could not process at the time, therefore, kept in a queue and (2) event execution time, the ideal number of O's replicas for processing these events at each interval is deduced. As a result, the number of O's replicas changes over time, depending on the input rate.

This proposal is evaluated over the infrastructure of a well-known cloud provider, the Google Cloud Platform (GCP). Performance was an exhaustive evaluation in terms of classical metrics such as end-to-end latency, resource utilization, the number of processed events, and the error of our estimations. This work presents, analyses, and discusses the results also comparing them with state of the art solutions.

Publications

Two articles in international conferences, two articles in French conferences, and one journal article submission under review. Chapter 4 presents the contributions of each article.

International Conferences

Organization

The remaining of this theses is organised as follows:

Chapter 2 gives the necessary concepts for the understanding of SPSs, the architecture of the SPS frameworks, and our contribution.

Chapter 3 presents related work about adaptive SPS. Existing solutions use a manual approach, which depends on a user for modifying the SPS, or a automatic approach, where the SPS is in charge of the adaptation.

Chapter 4 presents the proposed SPSs(RA-SPS and PA-SPS) as well as the pool of replicas and Load-Aware grouping approach.

Chapter 5 presents evaluation results related to RA-SPS and PA-SPS experiments conducted on GCP.

Finally, Chapter 6 presents the conclusion and some future work.

Background 2

This section introduces some concepts used in this work. First, we define what is stream processing. This is important since it characterizes the behaviour of events collected from data sources under this paradigm. Then, we introduce the Stream Processing Systems (SPS), both the system and its components, and some frameworks used to implement them.

Stream processing

Streaming refers to a continuous flow of data that is generated, transmitted, and processed in real time over a network [START_REF] Menin | The Streaming Media Handbook[END_REF]. It allows users to access and consume its content as it is under transmission, so there is no need to store all the content before using it. It consists of an external entity sending data to a data processing system. If the service is busy, the data is queued. Generally, streaming is used by web interaction, such as social networking or online playback of multimedia content.

Streaming is largely used in processing information in real-time, when the temporality of the data is relevant, such as online playback of multimedia material. For instance, Streaming API provided by Twitter, can be used to study the most relevant tweets, trending topics or hashtags for specific cases such as election campaigns or natural disasters. Streaming is processed by SPSs.

Figure 2.1 shows a server emitting a data stream, which is received by different clients. Each client is in charge of processing the received data, if the client is busy, data is buffered. Otherwise the client will process the data according to the service policy.

Stream processing is a computing paradigm and data processing approach that involves the continuous processing of data as it is generated and ingested in real-time. This paradigm focuses on programming applications that can process information on the fly as close to real-time as possible, using system resources in a parallel or distributed manner to meet their objective. It is typically used for real-time applications that close to real time responses, such as monitoring, fraud detection, and predictive maintenance. Figure 2.2 presents a Stream processing scenario. The input stream corresponds to the input stream of external data, which is delivered by an external source such as sensors, log records, or database transactions. Each data is a basic unit, which is processed to obtain relevant information. The stream process is the component in charge of processing the input stream. These processes can be stateful, so if necessary, the processing status is stored in an external database. It is also possible to process several data in parallel, either using more logical or physical resources. The output stream is the processed data, which can be used as a data stream to be processed or stored.

Processing Buffer

Input stream

Output stream For the correct functioning of these applications, some requirements should be satisfied. The work [START_REF] Andrade | Fundamentals of Stream Processing: Application Design, Systems, and Analytics[END_REF] defines them, which are classified as follows:

Stream process

• Processing large quantities of distributed data: One of the main goals of Stream processing is the processing of large volumes of data, which are sent in a distributed manner by external sources. By processing these data, the system can monitor, analyse, or control them in real-time, since the data processing is performed as the data stream continues.

• Addressing stringent latency and bandwidth constraints: This point refers to a stable connection between components over the network, where bandwidth or latency is not a limiting factor for data processing. Such a requirement is important since an application that considers data in real time is useless if it presents a high latency. Low latency should always be maintained, so that the data is as close to real time as possible.

• Processing heterogeneous data: A standard both in the data structure and its format should be application in the raw data from external sources that are used in the Stream processing. In this way, the processing will be homogeneous for all data, avoiding problems associated with the data structure.

• Providing long-term high availability: SPS operators can fail which inducing the decrease of data throughput. Thus, it is important to have a fault tolerance mechanism to reduce the loss of information. In the order, provide a constant processing of data, which is stable and persistent over time. Otherwise, information can be lost, compromising the accuracy of the results and requiring more time to collect the lost information or reach a similar state.

Stream Processing Systems

A Stream Processing System (SPS) is a software or framework designed to process and analyze data streams in real-time, which is based on the concepts of Stream processing. The main goal of an SPS is to process high volumes of data in a distributed way and in real-time [START_REF] Kleppmann | Making Sense of Stream Processing[END_REF]. In contrast to the traditional batch processing model, which stores the data to later process it offline [START_REF] Hawwash | From Tweets to Stories: Using Stream-Dashboard to weave the twitter data stream into dynamic cluster models[END_REF]. SPS shift involves the analysis of data without requiring storage.

The paradigm used by SPSs is based on a directed acyclic graph (DAG) as shown in Figure 2.3. The operators correspond to the vertices of the graph, such as analyzers, word filters or some particular algorithm, while the edges correspond to the dataflow between operators [START_REF] Shahrivari | Beyond Batch Processing: Towards Real-Time and Streaming Big Data[END_REF]. In addition, the input data (the source) is originated by an external entity, such as streaming from social networks, statistics from the 2.2 Stream Processing Systems monitoring of a system, or transactions in the stock exchange. The first operator consumes data from the source [START_REF] Appel | Eventlets: Components for the integration of event streams with SOA[END_REF]. In most graphs, the terminal operators are in charge of storing the results of the data processing in a database. It should be noted that SPSs are distributed, i.e., each vertex is hosted in a physical node available in the environment where the system is hosted, either a cluster, a grid or a cloud. To achieve communication between the operators, systems specially designed for this type of task are used, such as Apache ZooKeeper [START_REF] Hunt | ZooKeeper: Wait-free Coordination for Internet-scale Systems[END_REF]. The latter consists of a centralized orchestration that maintains configuration and synchronization information of the distributed applications. To this end, each processing node must register in the system, and it is the orchestrator that is subsequently in charge of synchronizing the nodes available for processing and distributing the events among them.

Most applications running on SPS, manage large amount of data, which must be processed to obtain information or statistics, as , for instance, fraud detection, collection of information in case of disasters or analysis of interaction in social networks. To perform real-time processing of the data, [START_REF] Stonebraker | The 8 requirements of real-time stream processing[END_REF] establishes the following requirements:

• Keep the Data Moving: To ensure low latency in the processing of high volumes of data, it is necessary to preclude the storage of the data when processing it, since the storing data adds unnecessary latency to the system. Thus, the aim is to process data "in-stream", i.e., as the data is received, it is processed.

• Query using SQL on Streams: To reduce the development and programming time of projects, it is important to provide an abstraction in the operations performed by the program, as done by a high-level language such as SQL. In this way, there is a set of default functions, which we can be use to query, group, join, or modify, accepted by most popular DBMS. Thus, providing a support to StreamSQL, a variant of SQL designed for SPS, ends up being an important tool for the SPS ecosystem.

• Handle Stream Imperfections: Because data are processed in-stream, it is necessary to provide a provision model for handling data that is delayed, missing, or out-of-order. Hence, if the SPS is performing an operation that considers a calculation in a time window, a mechanism must be provided to handle delayed or out-of-order events. On the other hand, if the system is waiting for a missing data, which might never arrive, then a timeout mechanism should avoid that the system blocks forever.

• Generate Predictable Outcomes: SPS can predictably process data, always given the same results. This means that the process must be deterministic and repeatable. Such a behaviour reflects the fault tolerance provided by the SPS, given that in the case of data loss, data recovery is possible and the result remains the same.

• Integrate Stored and Streaming Data: Some applications may be composed of stateful operations. For instance, the word-counting operator requires variables that store the statistics of the incoming stream. Therefore, the SPS must provide a system for storing, accessing and modifying the states used by the application. These states can also be added to the data in the stream, so it is important to have a uniform language to deal with both components.

• Guarantee data security and availability: SPS must provide data fault tolerance mechanism to ensure data integrity and provide integrity to the processing of critical data information. In this way, the system must provide checkpoint management for both data integrity and state. Thus, in case of data processing failure, the system will be able to reprocess data.

• Partition and Scale Applications Automatically: The distribution of an SPS is important both in terms of scalability and cost. By using the system on a single machine, resource constraints are likely to happen. Likewise, the costs associated with distributing a system across multiple machines are higher. Therefore, the SPS should ideally provide a transparent and automatic distribution of operators on the available machines, providing scalability in its processing.

• Process and Respond Instantaneously: When considering the use of SPS, a system is required to respond close to real-time. Therefore, the SPS must provide a solution that copes with operator overloads, which affect system performance. The solution should present low overhead, i.e., small implementation cost or resources requirements, increasing, thus, the efficiency and performance of the system.

Stream

Stream is a continuous flow of data records or events that are processed and analysed in real-time. It represents a sequence of data elements that arrive continuously over time. Stream is the fundamental building blocks in SPS corresponding to the input and output channels for data processing. They can represent various types of data, such as sensor readings, log entries, financial transactions, social media updates, or any other type of event-based data.

Streams in SPS typically exhibit the following characteristics [AS+13]:

• Continuous flow: Streams are continuous and persistent, with data flowing continuously over time. Data arrived are generated, transmitted, and added to stream, creating a dynamic traffic.

• High data arrival rate: In Streams usually data arrives at a high rate or frequency, requiring SPSs to handle and process data in real-time or near real-time to keep up with the incoming data.

• Potentially unbounded: Streams can be unbounded, meaning that there is no predefined endpoint or limit. They can continue indefinitely, and the processing system needs to handle the continuous arrival of data without assuming an end.

• Ordered or unordered: Streams can be ordered, where the order of events is meaningful and needs to be preserved during processing. On the other hand, if the order of events is not important, processing can occur independently on each event.

• Potentially partitioned: In distributed stream processing systems, streams can be partitioned into multiple partitions or shards. Each partition contains a subset of the data, and, thus, for parallel processing across multiple nodes or processing units can take place.

Streams are the primary input to SPS, and then they are ingested, processed, and transformed by various operators or computations. The processed results or derived streams can also be emitted as outputs for further processing, storage, visualization, or integration in other systems.

DAG

A directed acyclic graph (DAG) defines the processing logic of the SPS where each vertex represents an operator, and unidirectional edges (arcs) represent the dataflow. The DAG is defined as G = (V, A), where V is the set of vertices and A is the arcs group. Likewise, n represents the number of vertices in V and m represents the number of arcs in A. Each vertex corresponds to an operator, and each arc corresponds to a stream in the SPS application. For example, Figure 2.4 shows a DAG, where

G = (V, A) with V = {v 1 , v 2 , v 3 , v 4 } and A = {a 1 , a 2 , a 3 , a 4 , a 5 }, n = 4
and m = 5. In this case, the initial vertex is v 1 and the terminal vertex is v 4 . A path v is a sequence of vertices and arcs defined as v = v i a k v j ...v p a l v q , where v i , v j , v n , v m ∈ V for i, j, p, q ∈ {1, ..., n} and a k = {v i , v j }, a l = {v p , v q } ∈ A. Therefore, since in DAGs there are no cycles, the first and last vertices in the path are not the same, i.e., v i ̸ = v q . For example, in the Figure 2.4 it is possible to trace a path between v 1 and v 4 , defined by

a 1 a 2 v 1 a 3 v 2 a 4 v 3 v 4 a 5
v 1 = v 1 a 1 v 2 a 3 v 4 , v 2 = v 1 a 2 v 3 a 4 v 4 or v 3 = v 1 a 5 v 4 , without having cycles.
For a DAG G = (V, A), it is possible to associate the relation ≤ defined by: for any pair of vertices (v i , v j ) ∈ v for i, j ∈ {1, ..., n} , v i ≤ v j if there exists a path v from v i to v j in G. In this way, a topological ordering of G is defined as a list (v i , ..., v j ) of vertices of G for i, j ∈ {1, ..., n}, such that if i ≤ j, then there is no path between v j to v i . For example, Figure 2.4 has the list (1, 2, 3, 4) or (1, 3, 2, 4). The ordering is unique, only in the case that a path connects all the vertices, and corresponds to the order in which the vertices appear.

Paths can start from different vertices. Thus, depending on its origin, the set of vertices of different paths is not the same. there are two initial vertices (v 1 and v 5 ) and two terminal vertices (v 4 and v 7 ). In the case of choosing v 1 as the initial vertex, it is possible to end at v 4 or v 7 , but not in the case of v 5 , since it can only end at v 7 . By taking different paths, the application relies on different operators, so processed data also varies.

Tuples

The SPS must process structured input data. Therefore, initial operators are used to structure the input data, which can be structured, semi-structured or unstructured. Thus, each structured data created by an external data source is called tuple. A tuple is a basic unit of data that is defined by (key, value), a set of attributes or fields, which flow through the application graph. Each tuple is a single event in the data stream, which is often used to represent an event such as sensor reading, user action, or a financial transaction in real-time applications such as monitoring, alerting, or fraud detection. In general, a tuple has the timestamp of its creation, as well as the timestamp of its predecessor vertices in the DAG. The number and type of attributes in a tuple depend on the need of the application, as well as the path taken by the tuple.

Figure 2.6 presents the flow of a tuple through SPS application, which is defined by four components: input data, two operators and a database. Operator v 1 counts the number of words in the stream, and operator v 2 stores the tuples in a database. The start of the flow begins with the input data, structured in key-value. After this, the tuple is sent to operator v 1 , which determines that the number of words is 2. Therefore, it adds a new field count to the tuple and the timestamp record, and sends it to operator v 2 . Finally, the v 2 operator stores the tuple in a database, adding the timestamp record.

Operator

An operator is a processing unit that receives data from one or more input streams, performs some computation on the data, and produces one or more output streams. An operator is represented by a vertex in the DAG. The majority of operators are designed to perform light tasks, nevertheless, they can also implement complex tasks, depending on the design of the SPS application.

Depending on the DAG, operators can be replicated and the income data is processed in parallel. Also, each operator has a level of parallelism, which is related to the SPS implementation. Same solutions propose to associate parallel unit to logical resources (i.e. processes or thread), and others to physical resources (i.e. containers or VM). Figure 2.7a shows the DAG for an SPS application. The DAG is composed of three operators: v 1 , v 2 and v 3 . Due to the complexity of operator v 2 , the application requires increasing the parallelism of the operator. Figure 2.7b shows the parallelization Data is represented by tuples, so operators process tuples. Depending on the operator's task, tuples are sent without or with modification. It is also possible to discard them, create new tuples, or on the other hand, to perform some analysis on them. Futhermore, depending on whether or not the operator maintains the state or context of the data it processes, operators are classified into: stateless and stateful.

Stateless operators do not maintain the internal state or context of the data. They are independent tasks that depend only on the input tuples, without the need to know the past processing or global state of the application. These types of operators are used to perform operator-specific tasks. Stateless operators are associated with mapping, filtering, and data transformation tasks in the SPS.

Figure 2.8 presents an operator v 1 that performs a filtering task, which analyses if the text is in english written. If it is not, the event will not continue in the pipeline. In most cases, stateless operators are easily parallelizable, because it is not necessary to have consistency in the state of replicated operators, so they have a high scalability, which is useful for processing large amounts of data. When these operators are restarted, they have no impact in the overall processing of the data. For stateful operators, the output depends on the internal state of the operator. Therefore, they require past knowledge of the events or global state of the SPS to perform the task effectively. Such tasks are associated with windowed aggregations, pattern detection, or any operation that requires continuous analysis of data over some time. Figure 2.9 presents operator v 2 which has a state that stores a counter indicating the number of tuples containing a specific word under a time window. Initially the operator v 2 has a state hello_counter=9. Thus, after counting the incoming event the internal state will be hello_counter=10. This state can be stored in memory or in an external database. The management of its state increases complexity, due to data consistency, fault tolerance, and scalability of the operator, because, as the number of replicas of this operator increases, it is necessary to ensure the consistency of the replica states.

SPS Frameworks

In this section, we will present two popular SPS frameworks: Storm and Flink. The logical architecture will be presented, with the components that make up the DAG, as well as the physical architecture, explaining the components necessary for its deployment and communication.

Storm

Storm [START_REF] Toshniwal | Storm@ Twitter[END_REF] is a distributed SPS framework implemented in Java and Clojure that enables the processing of unbounded streams of data. It was initially developed by the BackType team, which was later acquired by Twitter.

It is an open-source Apache project, so there is a strong community of developers involved.

Logical architecture

A Storm application is represented by a DAG, also referred as Topology. There are three types of components in a Topology: Streams, Spouts, and Bolts.

Streams are a sequence of tuples created and processed in parallel following the DAG model. These are distributed around the application. Streams are composed of a structure of key-value tuples. Tuples can be defined by bytes, numbers, booleans or strings. Each Stream must be declared by a unique identifier.

Spouts are responsible for capturing the input data from external sources. They consume events and generate tuples sharing them to other Bolts downstream in the Topology. Spouts can be implemented in either reliable or unreliable fashion. A reliable Spout is capable of resending the same tuple in case it fails. Otherwise, it forgets to resend the tuple once it is emitted.

In Storm, operators are called Bolts. In general, these are lightweight tasks, so complex operations are designed as a pipeline of operators. Bolts can receive the tuples emitted from one or more Streams. Similarly to Spouts, Bolts can send the processed tuples through one or more Streams. To guarantee processing, Bolts can send an ACK (acknoledgement) message to indicate a tuple was processed. Each Bolt is parallelizable, so it has an associated parallelism degree, which corresponds to the number of replicas of an operator. In the presence of Bolt's replicas in Figure 2.10, Stream is partitioned and shared among them as shown in Figure 2.11, which is defined by Stream grouping. Every Bolt's replica processes the received tuple and sends the processed tuple to the next based on some stream grouping. Therefore, for each defined Stream it is necessary to define Stream grouping for sending a tuple between two components.

Storm has defined eight stream groupings, can be used. If necessary, it is possible to define new stream groupings.

1. Shuffle grouping: Tuples are sent randomly to each replica.

Fields grouping:

The stream is partitioned by a field given in the tuple that determines which replica will receive the tuple.

Partial Key grouping:

This stream grouping is proposed by [START_REF] Anis | The power of both choices: Practical load balancing for distributed stream processing engines[END_REF]. Similarly to Fields grouping, it used a specific field for the choise of the replica, but also considers the load balancing between the available replicas.

All grouping:

Tuples are sent to all replicas.

Global grouping:

Tuples are sent to a single replica. 6. None grouping: Tuples are sent randomly to each replica but in this case the same thread is always used by the assigned Bolts.

SPS Frameworks

Spout

7. Direct grouping: Tuples are sent directly to the desired replica according to some attribute declared in Stream.

Local grouping:

Tuples are sent to a group of replicas sharing the same resources.

Except Partial Key grouping, the drawback of these approaches is the potential lack of load balance. To cope with such a problem, other existing SPS propose hash-based data partition [START_REF] Mehul | Flux: An Adaptive Partitioning Operator for Continuous Query Systems[END_REF], partial-key based [START_REF] Anis | The power of both choices: Practical load balancing for distributed stream processing engines[END_REF] or executor-centric solutions [START_REF] Wang | Elasticutor: Rapid Elasticity for Realtime Stateful Stream Processing[END_REF].

Storm parallelism implementation is based on three concepts: Tasks, Executors and Worker processes.

Tasks are defined as data processing units, being replicas of a Bolt or Spout. While by default a task is assigned to one thread, it is possible to assign several tasks to the same thread.

An Executor is a thread. One or more Tasks of the same type (Bolt or Spout) can be assigned within the Executor.

Chapter 2 Background

Worker processes are a subset of a Topology, so its in charge to run a set of Executors.

A Worker process is a physical JVM. Each Worker process is assigned to a machine and allocated resources can be configured according to the latter.

Physical architecture

Storm Topology is deployed in a Storm Cluster, which is a physical environment for its execution. Its components, hosted on machines, cores or VMs, are distributed on a platform such as a Grid, Cluster or Cloud. There are two kinds of components: Worker node and Master node.

The Worker node is responsible for hosting Storm Topology tasks. Each Worker node runs a daemon called Supervisor. The role of the Supervisors is to host one or more Worker processes, so they deploy a subset of Storm Topology. Thus, when deploying a Storm Topology on a set of Worker processes, each associated task will be distributed around Storm Cluster.

The master node runs a daemon called Nimbus, which is responsible for distributing the associated Storm Topology code around the Storm Cluster. At the same time, Nimbus detects distributed component failures (i.e. node crashes or loss of messages) and the state of nodes. Finally, Nimbus runs the scheduling algorithm.

The scheduling algorithm is in charge of performing the mapping. This case, load balance problems may arise. For instance, when a random policy is applied (as Storm can do it), there is no guarantee that the workload will be homogeneously distributed among the operators since an operator may be more complex than another one [START_REF] Xing | Dynamic Load Distribution in the Borealis Stream Processor[END_REF] and machine resources could be heterogeneous. Even with homogeneous computation nodes, load balance issues can also arise [START_REF] Xu | T-Storm: Traffic-Aware Online Scheduling in Storm[END_REF]. Figure 2.13 shows the scheduling of Storm Topology (shown in Figure 2.10) in a physical platform. According to some scheduling algorithms, each operator replica is placed in an available worker node.

Architecture

For coordination between Nimbus and the Supervisors, Storm Cluster uses Zookeeper. Additionally, the states of the latter are stored in Zookeeper or on a local disk, to provide fault tolerance in case of node failure.

Flink

Flink [KS16] is a distributed SPS framework implemented in Java and Scala that enables the processing of unbounded data streams. It was initially developed by TU Berlin, and is currently part of the Stratosphere project.

Logical architecture

Like Storm, a Flink Application is a DAG. There are two types of components in a

Flink Application: Sources and Operators.

Sources are the components necessary for sending external data to Flink Application.

Each Source can be associated with logs, written data to external sinks, message queues, and interface with other systems. In this way, a Flink Application can have one or more Sources in the data processing.

Operators are the components that process the data in the Flink application pipeline. Unlike Storm, Flink provides a high-level model, providing several default functions, which can be used. In this way, the system provides a high level of abstraction, simplifying various aspects of application construction. Like Storm, each Operator can be parallelised according to the application's requirements. Operator, the last one, sends the processed data to either a database or an external system. This type of Operator is called Sink Operator.

Physical architecture

A Flink application is deployed on Flink Cluster, which has two components: Job Manager and Task Manager.

The Job Manager is responsible for distributing and coordinating the execution of the Flink Application. It decides on which machine each Task will be deployed. At the same time, it monitors the status of the tasks, in case of a possible failure or progress in the stream, as well as the coordination of checkpoints and their recovery.

Task Manager is the component that hosts one or more Tasks of Flink Application. Each Task is associated to an Operator, so this component must also allocate resources for its parallelisation. It also sends the state of the Tasks to the Job Manager.

Outside the Flink Cluster, there is a component named Client, which is responsible for the code of the Flink application and for sending it to the Job Manager for deployment. This component also receives statistics, results and the state of the execution of Flink Application.

Discussion

Although there are differences at the programming level, because Storm does not provide a high level of abstraction as Flink, both the logical and physical architecture are quite similar. On the one hand, the logical architecture is based on a DAG model, where each vertex is a component that performs a function and the edges are the data stream. As for the physical architecture, both are based on a model of main and secondary nodes, whereby the former is in charge of distributing the DAG as well as monitoring its state, and the later are responsible for processing the data.

It is worth noting that while we have given two examples, there have been other SPS frameworks, such as Apache S4 [START_REF] Neumeyer | S4: Distributed Stream Computing Platform[END_REF], Apache Heron [START_REF] Kulkarni | Twitter Heron: Stream Processing at Scale[END_REF] or Apache Apex [START_REF] Gundabattula | Apache Apex[END_REF], but they have been given up over time. One of the main reasons is the flexibility provided by cloud services such as Amazon Web Services, Google Cloud Platform or Microsoft Azure, where they provide their own SPS (Amazon Kinesis Data Streams, Google Cloud Dataflow and Microsoft Azure Stream Analytics), with a high level of abstraction, simplifying both deployment and programming.

Conclusion

In this chapter, we present the concepts used in an SPS, as well as its requirements. We have explained each component, i.e. Stream, Operator, and Tuple, and the concept of DAG, which is the theoretical basis of the paradigm. In addition, we explained the parallelism of the operators, detailing their implementation and replication, as

Conclusion

well as the types of Operator, stateful and stateless, which are differentiated by the state handling in them. Finally, we present two SPS frameworks, Storm and Flink, explaining both their logical and physical architecture, as well as a discussion of the differences and similarities between them.

Related work on adaptive| SPS 3

One of the challenges of SPS is the dynamic availability resources. For example, if the external data source has an exponential increase of data, its transmission rate will increase considerably, so that the resources (either logical or physical) might be insufficient for processing all the data. One solution is to modify the amount of logical resources (operators) to maintain stable SPS performance. The works presented below focuses on the operator scaling. This chapter presents two types of SPS adaptation: manual and automatic. The first one, as the name suggests, must be done manually by a user. The second one provides automatic scalability, being, thus, capable of modifying resource allocation according to the needs of the application. There are two approaches to automatic adaptation: reactive and predictive.

Manual adaptation

One of the requirements of an SPS is its scalability when the load increases. The distribution of the application across nodes and the parallelisation of its tasks need then to be adapted. The SPS frameworks mentioned above (see Section 2.3) provide both features, as well as the modification of their parallelisation through a client as shown in Figure 3.1. Figure 3.1a shows the request to modify the parallelisation of a SPS, and then in Figure 3.1b, once the modification has been performed, the response that the modification has been successful is sent to the client.

SPS

Basically, there are two ways to manually modify resources: GUI application or terminal command. In both cases, the parallelisation degree of the components must be indicated. For Storm, the number of Executors per Bolt and the number of Workers must be informed. While for Flink it is necessary to inform, the degree of parallelisation of each Operators and the number of Tasks. In both SPS framworks, reconfiguration requires stopping the application and restarting it. A comparative summary of the two is presented in Table 3 To cope with such a manual intervention drawback, ELK Stack [START_REF] Hasani | Improvement of Implemented Infrastructure for Streaming Outlier Detection in Big Data with ELK Stack[END_REF] proposes to read the SPS statistics and notify the user if the system is overloaded, aiming at helping him/her to decide on the system reconfiguration. Although this alert system indicates an overload in the system, the logic of the application or the system parameters remain dependent on the user's expert knowledge, which is a limitation of this solution.

Automatic adaptation

As a consequence of the limitations manual adaptation, some works have proposed self-adaptive SPSs [KG20; Ark+21] by modifying the amount of resources aiming at increasing the performance of the system and/or decreasing the costs associated with excessive idle resources. The works that follows are able to automatically modify the allocated resources in order to satisfy the requirement of instantaneous processing and response.

In general, this type of adaptation is achieved by using SPS statistics, through a monitoring system. Unlike manual adaptation, which is passive since it depends on a user to perform the adaptation, automatic adaptation actively and constantly analyses the SPS. Figure 3.2 presents the gathering of statistics by the monitor, which are performed according to some set of parameter (number of messages processed, time windows, etc.). Subsequently, the model in charge of the automatic adaptation determines that a new configuration is needed, so it actively signals this change to the SPS. Existing works apply either a reactive or predictive approach to implement an adaptive SPS. The reactive approach corresponds to adaptations that are performed in situ, i.e. it considers the states or/and variables of the system to determine whether it is necessary to adapt the system. This approach considers algorithms focused on modifying the system as a response to its current state, based on metrics and thresholds. On the other hand, the predictive approach analyses the history of the states and/or variables system to give a proposal of a new configuration based on prediction models. Therefore, more complex systems such as Machine Learning, time series, or mathematical models are used to determine the proposed configuration.

Model SPS

Reactive approach

The reactive approach is based on state analysis via a monitor, which periodically receives system statistics (utilization, queue size, throughput, CPU utilisation, etc.). These statistics determine an objective function, so that if their value exceeds a threshold, the system will modify the number of replicas of the operators.

Automatic adaptation

Figure 3.3 presents a variation of the objective function, which over time changes the state of the system. Consider that the used metric is the workload. If the metric value is lower (resp., higher) than the lower threshold (resp., upper threshold), the system is considered underloaded (resp., overloaded). And in the case where the value of the metric is between the lower and upper threshold, the system is considered stable. Once the state of the system is determined, if an instability happens (e.g. overloaded or underloaded), the reactive model proposes a new configuration. Figure 3.4 shows an example of a SPS, which uses a reactive approach. The metric determines the state of the system according the thresholds. Therefore, if the system is unstable, the model proposes a new configuration, which modifies the number of replication of the operators according to the needs. Otherwise, no configuration is proposed.

Lower threshold

StreamCloud [START_REF] Gulisano | StreamCloud: An Elastic and Scalable Data Streaming System[END_REF] is an adaptive SPS, based on Borealis SPS [START_REF] Daniel | The Design of the Borealis Stream Processing Engine[END_REF], that modifies the number of replicas in the system according to CPU utilisation. Depending on the number of queries issue to the system, the number of operators processing the requested tasks increases or decreases. For this, a specific operator, called slit, distributes the data, and an other one, called merge, gathers the information delivered by the operator's replicas. So this system only supports certain operations, so that split and merge operators are automatically created. In this way, there are no problems with stateful operators, such as counters and sorting algorithms, since it automatically performs the procedure of separation and union of data.

The MEAD SPS [START_REF] Russo | MEAD: Model-Based Vertical Auto-Scaling for Data Stream Processing[END_REF] was implemented in Flink [START_REF] Carbone | Apache Flink™: Stream and Batch Processing in a Single Engine[END_REF]. Operator autoscaling takes place based Markovian Arrival Processes approach, where the system load is analysed according to a queuing model. real environments. However, if the authors state that MEAD supports operators scaling-out, such a feature has not been implemented. On the other hand, similar to other works that use Flink, such as [START_REF] Hamidreza Arkian | Modelbased Stream Processing Auto-scaling in Geo-Distributed Environments[END_REF], reconfiguration induce performance degradation.

Enorm [START_REF] Grud | Enorm: efficient windowbased computation in large-scale distributed stream processing systems[END_REF], a work based on Storm, proposes elasticity that integrates with fault tolerance. For this, it uses a check-pointing mechanism to enable fast and low-cost state migration by increasing or decreasing parallelism. This work focuses on how to adapt the system by increasing or decreasing parallelism, but not on the amount of parallelism required by the system. In addition, the work uses a threshold-based approach to determine when to scale. We believe the author's proposal can be integrated into our solution to provide integrity when replicating stateful operators.

Some works also focus on the dynamic adaptation of SPS under a reactive approach, with the difference that they use their own SPS. For instance, the SPS Joker, which claims to provide elasticity with "organic" adaptation, is presented in [START_REF] Kahveci | Joker: Elastic stream processing with organic adaptation[END_REF]. The authors denote "organic adaptation", when the execution of streaming application scales safely, transparent, dynamic, and automatic. Joker continuously monitors the runtime performance of the SPS and runs optimization algorithms to resolve bottlenecks, using the throughput metric as an objective function. It then scales

Automatic adaptation

the application by adjusting the degree of pipeline and data parallelism. While parallelising SPS tasks, distribution between machines has not been implemented.

Other works such as [Ged+14; Sch+09] also use parallel elastic tasks and a metric (the throughput in both works) for determined the state of each task, and, the SPSs are deployed in the Cloud. Therefore, parallelization of tasks uses different VMs. Other similar work, [START_REF] Fernandez | Integrating scale out and fault tolerance in stream processing using operator state management[END_REF] propose to increase the number of replicas from the operators to reduce bottlenecks, where each replica is hosted in a VM. For the detecting bottlenecks, a monitor inquires the state of each operator from time to time. If an operator exceeds the established load threshold on the CPU usage, it is replicated.

In Esc [START_REF] Satzger | Esc: Towards an Elastic Stream Computing Platform for the Cloud[END_REF] the authors proposed to dynamically modify the replicas of the operators, as well as to couple and release machines to adjust the computational capabilities to the current nodes. To determine the number of replicas of the operators, the latency of the system is analyzed and the modification of replicas is determined according to thresholds. In the case of machines, their respective workload is used. Thus, the Esc implementation provides elasticity of both physical and logical resources automatically.

Another work that uses threshold-based decisions for operator replica adaptation is [START_REF] Heinze | Latency-aware elastic scaling for distributed data stream processing systems[END_REF]. The authors compared decisions based on local thresholds on the processing operators, global thresholds of the system, and a reinforcement learning approach. Besides lower and upper bounds a target utilization value is taken into account. A grace period after a scaling decision is considered in order to maintain stability. The reinforcement learning approach uses the actions: scaling up, scaling down, and no action. The system uses as reward a weighted average of the difference between the current value and respective target system utilization.

Flood [START_REF] Alves | Flood: elastic streaming MapReduce[END_REF] proposes a DPS (Distributed data stream processing) which, while modifying the replicas of the operators, is determined according to the physical resources. To this end, a manager module gathers runtime statistics such as the amount of used CPU, latency or available memory of the VMs. These statistics make it possible to see in which load range the machine can be placed, according to the established thresholds, in order to subsequently modify the VMs. Thus, by increasing (resp. decreasing) the VMs, the replicas associated with the VM are also added (resp. removed).

To this end, above works presented use only one metric, which does not take into account the behaviour of other variables, either local (operators) or global (SPS).

Predictive approach

The predictive approach is based on prediction of the future behaviour of an SPS to determine its new configuration. For this, its behaviour is estimated according to some predictive model based on its history. Examples of such models are regressions, time series, neural networks, etc. Data used for prediction can be system statistics, such as CPU utilisation, latency, throughput, etc. In [START_REF] Cardellini | Decentralized self-adaptation for elastic Data Stream Processing[END_REF], the authors propose a hierarchical decentralized adaptive SPS in Storm, using the MAPE model to design the solution. Regarding the scaling policy, the used metric is CPU utilization of the operator replicas, which defines whether a system adaptation is necessary or not. The proposed solution also analyzes the costs associated for each reconfiguration and on a latency-based reinforcement learning model to predict system behavior. One of the parameters is the downtime, i.e., the time necessary to restart the system which can induce much overhead.

There also exist some works that use SPS frameworks, such as Gesscale which is implemented in Flink [START_REF] Hamidreza Arkian | Modelbased Stream Processing Auto-scaling in Geo-Distributed Environments[END_REF]. In this work, a model is proposed in order to compute the maximum processing capacity of a physical node. For this purpose, like our approach, the SPS defines multiple different metrics which are the maximum sustainable throughput capacity of a single node, maximum network delay, and parallelization inefficiency. By applying these metrics, the model analyzes the behavior of the system in every time window and, if necessary, modifies the replicas elastically. One of the disadvantages of Gesscale solution is that, for reconfiguring the system, it is necessary to restart the application, which takes a considerable time.

The authors in [START_REF] Kotto Kombi | A Preventive Auto-Parallelization Approach for Elastic Stream Processing[END_REF] present a predictive SPS called AUTOSCALE which analyzes the data stream to predict traffic congestion on tasks. Queue theory principle is applied for gathering information about utilization, arrival rate, and departure rate of the tasks. A centralized system then analyzes the statistics, predicting data congestion in tasks according to a sliding window. Whenever the system detects a possible congested operator, the number of replicas is increased. However, the article does not present evaluation results in scenarios with high variations in the data flow rate.

DABS-Storm, a congestion prevention SPS, is presented in [START_REF] Kotto Kombi | DABS-Storm: A Data-Aware Approach for Elastic Stream Processing[END_REF]. Its aim is to reduce the degradation of the quality of the results. To this end, a metric is used to estimate the level of activity of the operators. A monitor gathers statistics about the operators activity and then, based on a metric, decides if the amount of resource allocated to each operator should be modified or not. Such a metric is defined by predicting the system input by using a regression function as well as taking into account pending events. The capacity of the operators is also estimated, considering both the physical capacity of the machine where the operator is located and the latency of the system. As DABS-Storm has been implemented in Storm, its operators reconfiguration approach carries the drawback of Storm reconfiguration downtime cost.

ELYSIUM [START_REF] Lombardi | Elastic Symbiotic Scaling of Operators and Resources in Stream Processing Systems[END_REF] is a Storm-based SPS that scales in and out the number of replicas of the operators and, if necessary, modifies the number of workers associated with the application (horizontal and vertical scalability). It provides both a reactive and predictive approach based on time window and an ANN model. ELYSIUM was not been evaluated with a real prototype integrated in Storm.

The authors in [START_REF] Balkesen | Adaptive input admission and management for parallel stream processing[END_REF] propose a predictive model implemented in Borealis SPS [START_REF] Daniel | The Design of the Borealis Stream Processing Engine[END_REF], taking into account not only the input rate as a metric, but also the capacity of the nodes as well as data processing complexity. Then, the model provides an equation that characterizes the workload of the system and determines the amount of required parallelism for processing events. Therefore, its objective is both the balance of the workload between the nodes and the reduction of latency.

Although the system is capable of scaling-out, it does not perform scale-in, so it does not consider the reduction of allocated resources.

Based on look-ahead approach, PLAStiCC is a predictive scheduling proposed by [START_REF] Gautam Kumbhare | PLAStiCC: Predictive Look-Ahead Scheduling for Continuous Dataflows on Clouds[END_REF]. Its model analyzes the system performance through the balance of resource overload. Furthermore, as it is conceived to run on clouds, allocated resources can have different costs. Therefore, the model considers not only the workload of the system, but also the costs associated with the increase in resources. The work is not evaluated in a real platform, nor does it use a real application, because it uses for evaluation the cloud simulator CloudSim [START_REF] Calheiros | CloudSim: A Novel Framework for Modeling and Simulation of Cloud Computing Infrastructures and Services[END_REF], as well as synthetic dataflows.

Automatic adaptation

The Elastic-PPQ SPS [START_REF] Mencagli | Elastic-PPQ: A two-level autonomic system for spatial preference query processing over dynamic data streams[END_REF] proposes to analyze the system at short-term and medium/long-term levels. The first one performs an analysis on the events that arrive in a time interval while the second one takes into account longer periods to perform a more complex analysis, using Fuzzy Logic Controller. To this end, an autonomous system, based on QoS, manages the system resources according to a runtime strategy, which considers the complexity of the system components.

In this way, the parallelism of the tasks, associated with a set of threads, can increase or decrease. For the evaluation and validation of system load analysis, both synthetic and real data were used. Although the solution is quite robust, since it is implemented in FastFlow [START_REF] Aldinucci | Fastflow: high-level and efficient streaming on multi-core[END_REF] framework, its focus is more on high performance processing than on distributed data processing.

Finally, [START_REF] Hidalgo | Self-adaptive processing graph with operator fission for elastic stream processing[END_REF] performs an analysis of operator replicas based on queue theory.

Operator utilization due to data arrival and departure rates is used as a metric, which determines the state of the operator, thus modifying the number of its replicas, if necessary. The solution uses a hybrid approach, composed of a predictive approach, based on Markov chains, and a reactive one, based on thresholds. The aim of the former is to analyse possible system behaviours, and the latter is to adjust the system configuration in short periods of time.

Conclusion

This chapter has presented several works on the adaptation of the number of operator replicas of an SPS, specifically the number of replicas of each operator in a DAG SPS.

First, we present the manual adaptation of two popular SPS frameworks, Storm and Flink, that allows the modification of their respective resources. Although they provide a way for reconfiguring a SPS, the solution depends on a user for adaptation. In addition, both frameworks have a downtime when modifying their configurations, which is an important limitation to consider.

36

Chapter Subsequently, we analyse automatic adaptation solutions, which is can be divided into two approaches: reactive and predictive. The ones based on reactive approach generally focus on metrics thath either considerer one variable on multiple variables whose relevance for decision can not be configured. These based on the predictive approach mostly use a SPS framework, which present performance degradation when modifying resources, given that there is a downtime of the system. Furthermore, they usually use only one predictive model, and the result may vary depending on the workload scenario.

Conclusion

Proposed adaptive SPS 4

In this chapter, we present our self-adaptive SPS capable of providing high throughput, scalability, and low latency, while guaranteeing the integrity of the results obtained from the analysed data. Our proposal, which is an extension of Storm, dynamically adapts the number of operator replicas to cope with environments that present highly variable input rate environments.

This chapter is structured in three sections. Section 4.1 presents the extensions made in Storm to circumvent its limitations. Then, Section 4.2 presents RA-SPS, our proposed reactive approach, published in [START_REF] Wladdimiro | A Multi-Metric Adaptive Stream Processing System[END_REF], which is based on multimetrics. Finally, Section 4.3 presents PA-SPS, our predictive approach, published in [START_REF] Wladdimiro | A predictive approach for dynamic replication of operators in distributed stream processing systems[END_REF], which focuses on predicting the number of replicas required by for the next time interval.

New features of the adaptive SPS

Most SPSs require expert knowledge for configuring the system's processing resources according to the environment requirements. Furthermore, they can not reconfigure themselves at runtime. Highly variable input rate environments may induce resource over-provisioning, wasting processing resources, while underprovisioning, may induce the loss of data processing.

In order to handle such scenarios, RA-SPS and PA-SPS can dynamically allocate/deallocate operators' replicas at runtime. Such an elasticity enables the two adaptive SPSs to adapt their processing logic, optimizing resource usage while minimizing data loss.

Both SPSs exploit two mechanisms that do not exist in the original Storm: a pool of replicas and a load-aware grouping strategy. The first one attempts to reduce adaptation reaction delay reducing, therefore, event processing latency. Storm's original version provides a rebalancing feature to reallocate the number of executors for a bolt. However, this action involves system downtime, loss of messages, and increasing of end-to-end latency which degrades performance. The second mechanism handles operators' replicas event distribution. Grouping strategies in Storm define how events are distributed among operators. Due to the heterogeneous nature of the processing resources and operators' tasks complexity, balance issues may occur, creating bottlenecks and increasing latency.

Pool of replicas

At initialization, RA-SPS and PA-SPS assign, for each operator, a pool of replicas deployed by the scheduler. Replicas can be either in active or inactive state. The state of a replica can be modified at runtime. An inactive (resp., active) replica consumes negligible (resp., non negligible) CPU power and can be dynamically activated (resp., deactivated) whenever the prediction model of the system detects the need for increasing (resp., decreasing) the replicas for the operator in question. Thus, based on the number of available cores by VMs and the fact that, in general, each replica is associated with a thread, it is possible to set the size of a pool (p), which is the same for each operator pool. Despite its simplicity, performance results showed that the pool of replica is very effective, since RA-SPS and PA-SPS are self-adaptive at runtime at a negligible cost. Moreover, as it is implemented in Storm, there is no need to restart the application, so downtime is avoided.

Load-Aware Grouping

A grouping technique specifies how a stream (tuples) should be partitioned among operators. Using traditional methods like Shuffle grouping (see Section 4.1.2), tuples are randomly distributed across operators, ensuring that each operator receives an equal number of tuples. Due to the complexity of the tasks or the heterogeneous nature of the processing resources, load balance issues may occur. In this case, while pending events are still in the processing queue, new events can go on arriving.

To overcome this problem, we propose a load-aware grouping strategy, which considers the load state of active replicas in terms of µ i.j (t), i.e., the number of events processed by replica j of an operator i during a time interval t, et i , the average execution time of one event at operator i, and td, the time interval duration. Note that we define a replica j of an operator i as O i.j .

Therefore, the proportional distribution of events considers the current utilization of active replicas. The utilization is computed following Equation 4.1, where U ranges between 0 and 1 for a replica j of an operator i. A 0 and 1 values represent 0% and 100% utilization of the replica respectively. If all replicas present the same utilization value, events are sent in a round-robin fashion. Otherwise, events will be assigned to the replica with the lowest load.

U i.j (t) = µ i.j (t) × et i td (4.1)
Algorithm 1 shows the pseudo-code of the load-aware grouping strategy. The loop of lines 3-7 is responsible for selecting the replica j of operator O i with the lowest utilisation. Then, if the utilisation is 100% (U = 1), which means that all the replicas are overloaded, a replica candidate is randomly chosen (lines 8-9). Otherwise, the overhead of processing the event is added to the replica candidate utilisation (line 11). Finally in line 13, the event is sent to the replica candidate of the O i operator.

U i.m ← U i.m + et i td 12: end if 13: sendEvent(O i.m )

Reactive approach

Due to the high variability of the input rate, the adaptation of SPS resources is necessary for performance sake. In this first proposal, denoted RA-SPS, we use a reactive model to analyse the state of the operators to modify the number of active replicas of the operators in the DAG. Table 4 Queue metric is defined by Equation 4.4. The aim of the metric is to analyze the impact of the input queue on the operator with respect to its current processing capacity. The Q i (t) tackles the input traffic behavior (traffic shape). Sudden peaks will increase the number of the events queued, q i (t) value, generating higher values of Q i (t). If q i (t) value is 0 or Q i (t) value is negative, its value is set to 0.

Q i (t) = 1 - µ i (t) q i (t). (4.4)
In Figure 4.4, the three operators have 10 processed events (µ i (t) = 10 events). In the case of O 1 , it has a small tail, so its impact is minimal. Compared to O 1 , O 2 has a larger number of queued events, which is reflected by the increase of metric value. Finally, O 3 has a considerable high Q 3 (t) value, due to the large number of queued events. To determine the overall state of an operator, the State metric, denoted δ i (t), is defined by Equation 4.5. It uses the previously defined metrics U i (t), Q i (t), and E i (t). By introducing weights (ω U , ω Q , and ω E ), the impact of the 3 metrics can be balanced, allowing, therefore, the study of the relevance of each metric in different load scenarios.

O 1 O 2 O 3 O 1 O 2 O 3

44

Chapter 4 Proposed adaptive SPS

δ i (t) = U i (t) × ω U + Q i (t) × ω Q + E i (t) × ω E (4.5)
The δ i (t) value characterizes the overall state of an operator as shown in Figure 4.5. Following a threshold-based approach, two bounds are defined: the upper bound δ u and the lower bound δ l . Considering these bounds, an operator can be in one of the three following states: overloaded, stable, or underloaded. These states give information about an operator's effectiveness and efficiency. Effectiveness is the capacity to fully process the input data, while efficiency is the capacity to process data by taking advantage of the available resources. If an operator is overloaded, it is not capable of processing all the input data, losing efficiency. On the other hand, if the operator is underloaded, it processes the input data effectively but not efficiently.

Finally, an operator whose state is stable processes input data both efficiently and effectively.

MAPE implementation

Monitoring, Analysis, Planning, and Execution compose the MAPE control loop. This model is exploited in most autonomic systems. By repeating these four steps, the system can detect issues by analysing data. If a problem is found, a strategy is developed and the executed to solve the issue. The MAPE control loop brings the system autonomic features such as self-configuration and self-optimization. Figure 4.6 presents the architecture of RA-SPS:

In our system, the MAPE model integrates the before-mentioned components where each of the four MAPE modules performs the following tasks:

1. Monitor: module in charge of collecting statistics from the DAG. At a predefined time interval t, the monitor requests the values of µ i (t), et i (t), and the number of queued events q i .

Analysis:

The module analyses the metrics of each operator and determines its status according to Equation 4.5.

3. Plan: Based on the previous analysis, the Plan module defines whether it is necessary or not to modify the system resource capacity. See Algorithm 2 presented in Section 4.2.3.

4.

Execute: module in charge of carrying out the change in an operator's current number of replicas, if required by the Plan module.

Planning

Algorithm 2 that presents the algorithm executed by the Plan module for the O i operator, deciding if the number of replicas of O i should be increased (or decreased) by k or remain the same. It uses a fixed value for k for the RA-SPS evaluation (see Section 5.1.4). However, it is possible to dynamically modify this variable for the adaptation of the SPS.

Therefore, the aim of the algorithm is to modify the number of replicas of an operator O i . If the operator is overloaded in line 3 (δ i (t) > δ u ) or underloaded in line 7 (δ i (t) < δ l ), the number of replicas should be increased or decreased respectively by k replicas. Otherwise, if the operator state is stable, the number of replicas will not change. In order to ensure system stability, we define that an operator must remain in the same state by at least two consecutive time windows in line 2 (δ i (t) = δ i (t-1)) before carrying out any change in the number of replicas.

While the size of the pool of replicas is defined by the amount of cores, in case there is an overload of physical resources (VMs or nodes), it is necessary to limit the number of active replicas. For this, we determine a threshold δ E for the metric E i (t), because the value of E i (t) indicates the degradation of the processing of the operator O i . If E i (t) is greater than the threshold δ E (line 4), the planning cannot increase the number of active replicas. In this way, we avoid an overload on the physical resources of the system and performance degradation of the SPS.

Algorithm 2 Adaptive planning algorithm according to the reactive approach for the operator O i .

Require

δ i (t) ← calculateMultiMetric(U i (t) , Q i (t) , E i (t)) 2: if δ i (t) = δ i (t -1) i then 3: if δ i (t) > δ u then 4: if E i (t) > δ E then 5:
Add k active replicas for O i 6:

end if 7:
else if δ i < δ l then 8:

Remove k active replicas for O i 9:

end if 10: end if

Predictive approach

Under highly variable input rate environments, input prediction is crucial in order to adapt the system processing logic over the time, which will keep events flowing, ensuring accurate results.

In our second proposal, denoted PA-SPS, we use a prediction model for the estimation of the optimal number of replicas for a given operator in the DAG. Therefore, unlike RA-SPS, a reactive approach that focuses on detecting traffic peaks, PA-SPS, a predictive approach, focuses on finding patterns in the traffic, predicting possible overloads or underloads. Table 4.2 summarizes all the parameter's notations used in PA-SPS.

Predictive model

The aim of our predictive model is to estimate how many active replicas would be necessary for operator O i to process the number of input events for the next time interval t + 1 ( λ i (t + 1)). It considers the processing capacity of O i as the average We point out that since prediction of a number of active replicas depends on multiple factors, dynamically determining it is not trivial. Thus, we model this complex problem so that the value of λ i (t + 1) is determined by Equation 4.7 which, in turn, is determined by the prediction of received events ( λ r i (t + 1)) and the prediction of queued events ( λ q i (t + 1)) during the next time interval.

λ i (t + 1) = λ r i (t + 1) + λ q i (t + 1) (4.7)
In most SPS, the processing logic is represented by a DAG. The latter establishes a dependency condition where operators share a stream of events according to their location in the DAG. For example, let's consider a linear DAG with two operators (see Figure 4.8), O 1 and O 2 , and their respective values of λ r i (t) and µ i (t) (see Table 4.2). µ 1 (t) and λ r 2 (t) are equal since operator O 1 has sent all the events it has processed to its single successor O 2 . If i is the initial single DAG operator, then

λ r i (t) equals λ G (t) (λ r 1 (t) = λ G (t)).
Note that the increase of O p 's number of active replicas at the end of the interval t has a direct impact in O p 's successors, since, in this case, µ p (t + 1) increases and thus, λ r i (t + 1) too, inducing a domino effect that the prediction formulations should avoid. For example, in Figure 4.8, if µ 1 (t + 1) increased from 5 to 10 events, due to the replication of O 1 at the end of t, λ r 2 (t + 1) would increase as well. Hence, if the operators process all received events during t + 1, we have that λ G (t + 1) = λ r 1 (t + 1) = µ 1 (t + 1) = λ r 2 (t + 1) and, consequently, all operators O i are dependent on λ G (t).

The above example is a borderline case. In a SPS DAG, not always, all the output processed events of O p i , the predecessor operator of O i , will be sent to O i . It might happen that O p i splits, filters, or replicates the events into several streams, sending each of them to one of its different successor operators in the DAG. Thus, we define

θ p i (t) = λ p i (t) µ p (t) (4.8)
Considering the condition of dependence among the operators, θ i (t) is defined as the sum of the events processed by the predecessors of O i , where it is determined by all the predecessor operators O p i as presented in Equation 4.9. Note that θ i (t) must be calculated from the initial operators.

θ i (t) = p∈pred(O i ) θ p i (t) × θ p (t) (4.9)
Thus, we define Equation 4.10, which predicts the number of events received by the operator O i during t + 1. For its calculation, the percentage of events processed by operator O i is used (θ i (t)), according to the prediction of the number of events sent by the input data ( λ G (t + 1)). Note that a predictive model is used to calculate λ G (t + 1) (see Section 4.3.2).

λ r i (t + 1) = λ G (t + 1) × θ i (t) (4.10)
Figure 4.9 shows a DAG example and θ p i value which was estimated applying Equation 4.8. It is worth pointing out that, given the dependence among the operators, it is necessary to start from the initial operator downstream to the last one. Since θ 1 is equal to 1, O 1 receives all the events sent from the input and then splits them among O 2 and O 3 . In this case, the two operators do not receive all the events from their respective predecessor, O 1 ; therefore θ 2 and θ 3 have values 0.7 and 0.3 respectively. Finally, operator O 4 receives events from its predecessor operators O 2 and O 3 . However, O 2 does not send all its processed events to O 4 , but 50 Chapter 4 Proposed adaptive SPS only θ 2 4 = 0.4, unlike O 3 which sends all processed events to O 4 (θ 2 4 = 1). The value of θ 4 is, therefore, 0.58, according to Equation 4.9. Finally, we should consider the events received and not processed during t by O i which are kept in q i (t). Hence, the number of input events λ i (t + 1) that O i should actually process in t is composed not only of received events λ r i (t + 1) but also of the events queued in O i and its predecessor operators O i p because of the domino effect on the DAG. For this reason, Equation 4.11 is defined, where λ q i (t + 1) consists of the number of events queued in O i and the percentage of queued events that will be sent by its predecessor operators O p i during t + 1. Note that there is one queue per operator.

λ q i (t + 1) = |q i (t)| + p∈pred(O i ) λ q p (t + 1) × θ p (t) (4.11)

Input prediction

As mentioned above, Equation 4.10 used λ G (t + 1), which indicates the predicted number of events sent by the input data during t + 1. Its prediction is based on the previous observation windows according to a number of samples s. The observation time interval is determined by to. The prediction is applied by one of the following models:

Predictive approach

• Basic: considers that the input data values during t + 1 will behave the same way as they did during t.

• LR: uses a simple linear regression similar to the one presented in [START_REF] Douglas C Montgomery | Introduction to linear regression analysis[END_REF].

• FFT: Fast Fourier Transform decomposes functions depending on space or time into functions depending on frequency. It allows to predict the input data by modeling its behavior as a time series [START_REF] Henri | The fast Fourier transform[END_REF].

• ANN: uses a neural network regression model, specifically a Multi-Layer Perceptron (MLP). It implements an MLP algorithm for training and testing data sets using backpropagation and stochastic gradient descent methods [START_REF] Riedmiller | Multi layer perceptron[END_REF].

The parameters values used in this model where extracted from [START_REF] Pedregosa | Scikit-learn: Machine Learning in Python[END_REF].

• RF: Random Forest combines learning methods with the decision tree framework to create multiple randomly drawn decision trees from the data. [START_REF] Steven | Random forest[END_REF].

The parameters used are the values for defects in [START_REF] Pedregosa | Scikit-learn: Machine Learning in Python[END_REF].

MAPE implementation

As with the reactive approach, in PA-SPS we will use the MAPE model for the automation of system adaptation. The big difference is in the Analysis module, because it performs a request to the predictor model for input rate prediction. Figure 4.10 presents the architecture of PA-SPS:

Adaptive SPS O 1.1 O 2.1 O 2.2 O 3.1 O 2.3
Input Data

Monitor Analysis Plan Execute

Control loop Predictor Fig. 4.10: The architecture of PA-SPS.

In our system, the MAPE model integrates the before-mentioned components where each of the four MAPE modules performs the following tasks:

1. Monitor: Module in charge of collecting statistics from the DAG. At a predefined time interval t, the monitor requests the values of λ i (t), et i , and the number of queued events q i (t).

Analysis:

The module analyses input data and predicts its behaviour following Equation 4.7. Note that the analysis is performed for each operator in the DAG.

3. Plan: Based on the previous analysis, the Plan module defines whether it is necessary or not to modify the system resource capacity. For this, we define Algorithm 3 which is explained in Section 4.3.4.

4.

Execute: Module in charge of carrying out the change in an operator's current number of replicas, if required by the Plan module.

Planning

Algorithm 3 presents the algorithm executed by the Plan module for the O i operator, deciding if the number of replicas of O i should be increased (or decreased) by k or remain the same.

First, the algorithm determines the number of replicas for the next time interval is computed (line 1). To this end, the number of necessary replicas is calculated according to Equation 4.6, using an input data predictor to determine the future behaviour of the input stream. Then, the getReplicas(O i ) function returns the current active replicas of O i . Hence, the difference between the prediction and the current number of replicas will be the number of replicas (k i ) to be modified (line 2). If the value of k is positive, then k i replicas are activated (line 4). If the value of k is negative, then k i replicas are deactivated (line 6). If the value of k is 0, then the same number of active replicas is kept.

Conclusion

In this section we propose two adaptive SPSs, RA-SPS and PA-SPS, under a reactive and predictive approach respectively. These adaptive SPSs dynamically modify the number of operator replicas based traffic fluctuations. Both proposals are implemented on top of Storm.

Moreover, two other features were included in the two adaptive SPSs. The first consists of a pool of replicas aiming at avoiding the restart the SPS when the number of active replicas is modified. The second consists of a grouping strategy, which denoted Load-Aware grouping. The latter determines which replica will process the event according to the load of the operator's active replicas, the latter the load balance among the replicas.

The first adaptive presented SPS is RA-SPS, which is based on a reactive approach. Its aim is to adapt resources according to the analysis of traffic peaks in short periods of time. For this propose, the multi-metric δ i (t), called State metric is used to define the state of a given operator: overloaded, stable or underloaded. Each state indicates the behaviour of each operator, so that the Planning Algorithm can determine if any modification is necessary. The State metric is defined by three weighted metrics: U i (t) that determines the percentage utilisation of O i during interval t; E i (t) that determines the processing degradation of the O i operator during interval t; Q i (t) that determines the impact of queue size on the O i operator during interval t.

The second adaptive presented SPS is PA-SPS, which is based on a predictive approach. It aims to find patterns in the traffic to predict possible overloads or underloads in the SPS. To this end, the number of replications needed by operator O i during the next time interval t + 1, denoted r i (t + 1), is predicted. This value is defined by the average execution time of an event by operator O i and the predicted number of events received by operator O i during the next time interval t + 1. To predict the number of events received by operator O i , the predicted number of events sent by the input data and the number of queued events are considered. Furthermore, the prediction also considers the dependency among operators, in order to mitigate a possible cascade effect due to the change of the number of replicas. For the prediction of events sent by the input data, the use of a predictive model based on mathematical model or artificial intelligence model is proposed. and one for running both the Nimbus and the adaptive SPS. Our test environment is presented in Figure 5.1.

Three types of machines were used: a n1-standard-1 (1 CPU, 2.2 GHz, 3.75 GB of RAM) machine for hosting Zookeeper VMs, a n1-standard-4 (4 CPU, 2.2 GHz, 15 GB of RAM) for Nimbus and the adaptive SPS, and a n1-highcpu-8 (8 CPU, 2.2GHz, 7.2GB of RAM) machine for the Supervisors VMs.

Dataset

We have considered four dataset for the experiments: Twitter Gaussian, Twitter Smoothed, Twitter Raw, Logs and DNS.

• Twitter Gaussian: This dataset is traffic that follows a Gaussian distribution, whose data are tweets related to COVID-19 [START_REF] Gruzd | COVID-19 Twitter Dataset[END_REF]. Figure 5.2 shows the traffic provided by this distribution.

• Twitter Raw: This dataset is based on data from Twitter related to COVID-19, with 237 million tweets [START_REF] Gruzd | COVID-19 Twitter Dataset[END_REF]. Figure 5.3 shows the traffic provided by this dataset.

• Twitter Smoothed: This dataset is based on Twitter Raw. However, we have considered only a sample of these tweets in the experiments, i.e., those in periods of the datasets that present high rate variation. In other words,

56

Chapter 5 Experimentation we select a combination of traffic spikes and under spikes to compose the input traffic for the experiments. The methodology adopted to build the testing dataset was introduced in [Bod+10]. • DNS: This dataset is based on network traffic, which was created to test DNS over HTTPS, a more secure version of the DNS protocol [START_REF] Montazerishatoori | Detection of DoH Tunnels using Time-series Classification of Encrypted Traffic[END_REF]. We used the same methodology mentioned above was used to build the dataset. Figure 5.5 shows the log traffic provided by this dataset. • Log: This dataset is based on system logs from a distributed system [START_REF] He | Loghub: A Large Collection of System Log Datasets towards Automated Log Analytics[END_REF]. We used the same methodology mentioned above was used to build the dataset. Figure 5.6 shows the log traffic provided by this dataset.

Application

For each traffic, we have at least one application to be used. The applications designed for the experimental phase are listed below:

• Twitter linear: It is a linear DAG for Twitter dataset, which is composed of four operators.

-Detection: Its goal is to detect tweet events of a Twitter data streaming according to a topic, subtopic, category, and subcategory list of keywords. Each operator has one of these lists and verifies if any of the words of a tweet is included in the list in question. Therefore, in the sequence, the first, second, third, and fourth operators respectively determine the topic, subtopic, category, and subcategory of each tweet as shown in -Classification: Its goal is to classify tweet events of a Twitter data streaming. The first classifier determines whether the text is positive, negative or neutral, and the second identifies the person who has published them. Classified tweets are stored in a database. Figure 5.8 shows this application.

Stopword

Sentiment classifier

User classifier

Database store Twitter streaming Fig. 5.8: Classification Twitter application in SPS.

• Twitter complex: It is a complex DAG for Twitter dataset, which is composed of eight operators. Figure 5.9 shows the DAG of this application. It analyses Twitter streaming containing information such as news or opinions. Depending on the type of information, the stream can be split. Finally, results are stored in a database. • Log: It is a linear DAG for Log dataset, which is composed of four operators. Figure 5.10 shows its representation of DAG. Its objective is to determine the importance of the log trace.

• DNS: It is a linear DAG for DNS dataset, which is composed of four operators. Figure 5.11 shows its representation of DAG. Its goal is analyses and classifies events based on DNS traffic. For Storm parameters, we used default values1 , except timeout to detect the failure of an event, t out = 30s, and queue size, q size = 100000.

Parameters

Metrics

For the evaluation, we have defined six evaluation metrics.

• Saved resources: proposed in [START_REF] Lombardi | Elastic Symbiotic Scaling of Operators and Resources in Stream Processing Systems[END_REF], this metric expresses the proportion of resources (active replicas) saved with respect to a statically over-provisioned configuration. It is defined by 1 -r rover , with r the number of active replicas, and r over the overestimated number of replicas. r over is the number of replicas needed to process all the events during the highest input rate peak of the benchmark. Note that if the value of the metric is close to 1, a high number of resources has been saved.

• Throughput degradation: this metric, also described in [START_REF] Lombardi | Elastic Symbiotic Scaling of Operators and Resources in Stream Processing Systems[END_REF], aims at analyzing the behavior of the system in terms of throughput stability. It is defined by |inputrate-outputrate| inputrate . If the metric value is close to 0, the system has a good stability. On the other hand, if it is close to 1, the system is not capable to process the input rate and the system is unstable.

• Latency: is the average time taken by an event between the moment it enters and leaves the SPS (end-to-end latency). This metric is relevant since SPSs are supposed to deliver real-time processed events.

• Difference in the number of processed events: is the difference between the total number of processed events and the total number of received events. It is an important metric since SPSs are used to process high volumes of data.

• Error estimation input: is the mean absolute percentage error of the difference between the input rate and the predicted input rate during each interval.

• Error estimation replica: is the mean absolute percentage error of the difference between the number of replicas needed to process all events and the number of predicted replicas during each interval.

Impact of the new features

The aim of this section is to evaluate the modifications made to our SPS, being an extension of Storm, with respect to the standard version of Storm. These modifications are described in Section 4.1, which correspond to Pool of replica (Section 5.2.1) and Load-Aware grouping (Section 5.2.2).

Pool of replica

In this experiment, we compared Storm, denoted Storm-Default, with our modified version of Storm that uses the pool of replicas, denoted Storm-Pool, described in Section 4.1.1. We used the Twitter Gaussian dataset, the Twitter linear -Detection application, Shuffle grouping for stream grouping strategy. Both versions of Storm use the reactive approach presented in Section 4.2. Note that Gaussian traffic shape allows to evaluate the capacity of the system to scale-out and scale-in. For calculating the Saved resources metric, we have fixed r over = 60 (i.e., r i = 15). The size of each pool of replicas (p) was set to 12. Figure 5.12 presents the number of replicas required by each of the two systems. We observe that the difference between both curves is not very significant. Furthermore, the difference in Saved nodes values of both systems shown in Table 5.3 is of 0.03% and, in terms of memory usage, Storm-Pool requires only 2.6% of extra memory when compared to Storm-Default, which is due to the pre-allocation of the pools of replicas when deploying the application.

Figure 5.13 shows the output data rate for each system. The main drawback of Storm-Default is the need to restart the system at each reconfiguration. We can observe that output rates drop at each reconfiguration. On the other hand, Storm-Pool exploits the preloaded replicas that enable the system to process events continuously 62 Chapter 5 Experimentation while adapting itself. Table 5.3 shows a difference of almost 20% on throughput degradation between both systems.

The number of cumulative processed events is shown in Figure 5.14. Once again, we can observe the impact of reconfiguration downtime over the performance of the Storm-Default: there is a 13.6% difference between both systems in terms of the total number of processed events. We highlight that message loss in real stream processing applications can be critical (e.g., fraud detection systems). Therefore, based on the above results and discussions, we can conclude that Storm reconfiguration approach is not suitable for real-time applications that require timely responses. difference. On the one hand, Storm-Pool is always available to process incoming data. On the other hand, Storm-Default is down during the reconfiguration phase. In this way, there is a greater number of queued events part of the Storm-Pool, resulting in an increase in latency. We should also point out that the MAPE integrated in our SPS does not use much more extra resources since the monitored information exploited by it is the same one collected by Nimbus. Furthermore, MAPE algorithms do not require much computation time because they just consist in simple mathematical calculations according to the formulas presented.

Grouping

This experiment compares our stream grouping strategy, denoted Load-Aware Grouping, described in Section 4.1.2 with the shuffle grouping where events are randomly distributed among the replicas. We used the Twitter Smoothed dataset, the Twitter linear -Classification application and the predictive approach (PA-SPS) 2 , which used Basic model for input prediction. For calculating the Saved resources metric, we have fixed r over = 32 (i.e., r i = 8). The size of each pool of replicas (p) was set to 12. Table 5.4 shows the metrics for both grouping strategies. There is no significant difference in terms of processed events and only an increase of 4.04% in saved resources using our grouping strategy. Regarding the use of VMs resources, CPU utilization increases by 0.9% with respect to a random distribution and the difference in memory usage is negligible. On the other hand, there is a great difference in latency and throughput metrics of the two strategy since shuffle grouping does not take into account replicas' load.

Grouping

When a new replica is activated, the old loaded replicas must process both the new events that arrive and the ones it previously queued, which explains a decrease of 60.18% in latency and 57.73% in throughput degradation. This is also reflected in Figure 5.16, because in situations where there is a higher load on the operators (high input rate peaks), Shuffle grouping has a considerably higher latency than Load-Aware grouping.

Conclusion

In this section, we have been able to evaluate the performance of our SPS, being an extension of Storm, which has a considerable improvement to the standard version of Storm. This is due to the limitations of the standard version. First, we analysed the impact of the use of the replica pool, which ends up being a benefit for the SPS. This is because there is no downtime when modifying the logical resources (replicas of the operators) of the application. Secondly, we analyse the impact of a load distribution in the dispatch of events, whereby a more sophisticated strategy increases the performance of the SPS.

Impact of Storm parameters

Aiming at tuning their value, we propose in this section to discuss the impact of the two Storm parameters. The parameters are: timeout to detect the failure of an event (t out ), and queue size (q size ). We consider that an event has failed when its processing time on all operators exceeds the end-to-end timeout, t out . This parameter removes events that have been queued for a long time, reducing the load on an operator's replicas. We used the Twitter Smoothed dataset, the Twitter linear -Classification application, Load-Aware grouping for stream grouping strategy, and PA-SPS, which used Basic model for input prediction. For calculating the Saved resources metric, we have fixed r over = 32 (i.e., r i = 8). The size of each pool of replicas (p) was set to 12. 66 Chapter 5 Experimentation

Timeout

Table 5.5 shows the four scenarios, when the value of t out varies. We observe that t out has an impact in both latency and loss of processed events. For example, a t out = 1s improves the latency in 82.40% when compared to t out = 30s but, at the same time, there is a decrease of 9.5% in the difference of processed events. Therefore, on the one hand, if the proposed application does not require full data processing but low latency, one solution is to set the timeout to a low value for system deployment. On the other hand, if the application requires full event processing, it is recommended to use a high timeout value.

Queue size

Table 5.6 summarizes the four scenarios values for different sizes of the pending message queue, q size . The greater the queue size, the higher the number of queue events, thus reducing the loss rate and increasing the number of processing events (Diff. Proc. Events), as we can observe in the table. On the other hand, since many incoming events are dropped when using small queues, operators are less loaded and we observe an increase of the saved resources. For example, with q size = 100, we observe a 19.61% improvement in saved resources and a 98.57% decrease of the latency compared to q size = 100000. However, the number of dropped events highly increases, inducing a decrease of 46.12% in processed events.

Discussion

In this section, we have observed the impact on the processing of the number of events, which can decrease depending on the chosen configuration. The timeout parameter (t out ) is inversely proportional to the latency and inversely proportional to the percentage of processed events, because as its timeout decreases, events that take a long time to process are discarded. The same happens with the queue size (q size ), so it is only recommended to use low values for both parameters if there is a very limited amount of resources.

Reactive approach

This section presents performance results related to the evaluation of RA-SPS and its ability to adapt to the dynamics of the data stream, without reducing the rate of processed events.

The evaluation is composed of two parts: (1) a comparison of RA-SPS with different configurations (Section 5.4.1); and (2) evaluation results of a tweet-based more complex application (Section 5.4.2).

Weight evaluation

The current experiment aims at evaluating the performance of RA-SPS, which are based on the δ value (see Equation 4.5) and the weights (see Table 5.1). We used the Twitter Smoothed dataset, Twitter linear -Detection application, and Shuffle grouping for stream grouping strategy. For calculating the Saved resources metric, we have fixed r over = 40 (i.e., r i = 10). The size of each pool of replicas (p) was set to 50.

In order to quantify the impact of U and Q in the adaptation process, we evaluate them independently. The results of the 3 metrics are presented in Table 5.7.

Figure 5.17 shows the number of active replicas for each metric. A considerable increase in active replicas is observed in the first third of the three experiments. which generates a decrease in the performance of the system, due to the overhead of managing a high number of replicas. Likewise, in Table 5.7, the Saved nodes metric shows that the experiment with metric U requires 129.3% more active replicas than the one with δ metric. Note that for the U metric, we observe a negative value of saved nodes since sometimes the number of replicas becomes greater than the overestimated value. On the other hand, the Q metric has a 24.5% Saved nodes improvement over the δ metric but it succeeds to process only less than 32% of events whereas δ metric can process more than 89% of events. The behaviour of δ, Q, U output rate as well as input rate are shown in Figure 5.18. Regarding the three metrics, they are similar in the first and second peaks but not in the third one, since the Q experiment was not able to process events similarly to the other two: the queue increased, generating an overload in the system, which led the application to crash after the fourth peak. On the other hand, both the δ and U experiments continue to process events, until the eighth peak, which generates a saturation in the system of the U experiment, making the application to crash after the ninth peak. As mentioned above, a large number of active replicas induces an overhead for handling them, decreasing the performance of the system. We also observe that there is a 15.05% difference between the δ and U regarding the Throughput degradation metric (Table 5.7), which indicates a higher stability of δ 5. 4 Reactive approach experiment then U one. Also, due to the early instability of the Q experiment, there is a difference of 57.38% with respect to the δ one. The total number of processed events is shown in Figure 5.19. Because of application crash, after t = 1600s (resp., t = 3300s) the curve is a constant for the Q (resp., U ) experiment. The difference using δ instance with respect to U and Q is 15.05% and 57.3% respectively (Table 5.7). Figure 5.20 presents the latency of the three metrics. At t = 1600s, there is a strong rise in the latency for Q until the application crashes. The same happens at t = 3300s for the U experiment. Due to the number of events and the system overload, the latest queued events can not been processed. The system becomes then saturated and is not able to continue processing. On the other hand, the latency with δ is on average higher, but the system is capable of processing a greater number of events. Therefore, although the δ instance does not have better performance in terms of latency, it is able of processing a greater amount of data without having to cope with the problem of over or under estimated number of per operator replicas, as in the case of U and Q experiments respectively. The δ latency increase relative to U and Q is 69.3% and 37.8% respectively.

Complex application

We have also evaluated RA-SPS with a complex application. For this experiment, we have compared RA-SPS with an overprovisioning Storm which always uses a fixed number of replicas per operator, denoted S over . Such numbers are fixed (r i = 5) at the beginning of the data processing and do not vary during the experiment. We used the Twitter Smoothed dataset, Twitter complex application, and Shuffle grouping for stream grouping strategy. For calculating the Saved resources metric, we have fixed r over = 40 (i.e., r i = 5). The size of each pool of replicas (p) was set to 6.

Table 5.8 presents the results of both systems, RA-SPS and S over . Due to the time gap to process events, we observe a 42.52% difference between Throughput degradation values of the two systems. However, such a difference is not a real problem since the numbers of processed events of the two systems are quite close, as shown in the same table. On the other hand, in RA-SPS, we observe a high reduction of used resources with 50.23% fewer active replicas, when compared to S over . Also, in this application, RA-SPS succeeded to process almost 98% of the total events while consuming 2.7 times less CPU than the static configuration which overestimates the number of replicas.

Discussion

We have evaluated RA-SPS based on δ metric which aggregate three metrics, the average load of the operator (U ), the average execution time of an event (E), and the operator input queue (Q) for characterizing the state of an operator at runtime. By assigning a weight to each of these metrics, RA-SPS can decide whether the operator is overloaded, underloaded or stable, respectively increasing, reducing, or keeping the same number of active replicas. Performance results with Twitter input data and different evaluation metrics confirm the advantages of using the three metrics compared to a single one.

Predictive approach

This section presents performance results related to the evaluation of PA-SPS and its ability to adapt to the dynamics of the data stream, without reducing the rate of processed events.

The evaluation is composed of six parts: (1) the impact of interval time (Section 5.5.1); (2) an analysis of predictive models (Section 5.5.2); (3) a comparison of PA-SPS with the original Storm, using a fixed number of replicas (Section 5.5.3) as well as with (4) the other predictive adaptive SPS, DABS-Storm, proposed in [Kom+19] (Section 5.5.4); (5) an experiment with a complex application (Section 5.5.5) and finally (6) with other datasets (Section 5.5.6).

Impact of the time interval

Aiming at tuning their value, we propose in this section to discuss the impact of the time interval (td) used in Equation 4.6 3 . We used the Twitter Smoothed dataset, the Twitter linear -Classification application, the Basic model for input prediction, and Load-Aware grouping for stream grouping strategy. For calculating the Saved resources metric, we have fixed r over = 32 (i.e., r i = 8).

Table 5.9 shows the values of the four metrics when the value of td varies. Note that the greater the time interval, the greater the number of samples used for calculating Equation 4.6. We observe an improvement in the results when the time interval is small, which is also in accordance with the dynamic behavior of the input rate. Latency and throughput degradation confirm the latter, given that by increasing td the system needs to wait longer to adjust the number of replicas and stabilize.

It is important to highlight that, unlike Storm's traditional solution, which must restart the application to reconfigure the number of resources of each operator, PA-SPS should only activate or deactivate replicas in the pool. Therefore, the reconfiguration downtime does not exist. Furthermore, the computational cost of calculating the equations is minimal since they are basic operations carried out by the system. Consequently, even if reconfiguration occurs quite often, we do not observe a decrease in the number of processed events. Tab. 5.9: System metric values with different time intervals using Twitter Smoothed.

Comparison of predictive models

We propose in this section to discuss the use of predictive models for the calculation of the number of events sent by input data ( λ G (t + 1)) and their impact on system performance. We used the Twitter Smoothed dataset, the Twitter linear -Classification application, and Load-Aware grouping for stream grouping strategy. For calculating the Saved resources metric, we have fixed r over = 32 (i.e., r i = 8).

For each predictive model, Table 5.10 shows the respective values for the above discussed metrics. There is no difference in processed events, except for the RF model that presents a slight decrease in the number of processed events, representing 5.5 Predictive approach amount of resources. Consequently, more replicas are available for event processing as shown in Figure 5.21.

Conversely, if the number of replicas is underestimated, the margin for processing events is smaller, making it more likely that events will be stuck and the system will be more unstable. In this case, events will probably get stuck in queues, making the SPS more unstable. Regarding accuracy, LR has the best one with an improvement of 57.54% over ANN, which does not mean that it present better performance, because there are moments when underestimating the number of replicas decreases the processing of events in the execution of the system. Unlike ANN that presents an overprovisioning of resources whenever the curve rises. On the other hand, in FFT, its estimation error has a strong impact in PA-SPS performance since it does not accurately predict the behaviour of the input rate, as shown in Figure 5.22.

Comparison of PA-SPS with Storm

This section compares PA-SPS with the original Storm where the number r of replicas per operator is fixed. We have considered three configurations for Storm: no replication (r = 1); four replicas (r = 4); eight replicas (r = 8). The latter corresponds to the overprovisioning configuration where the total number of replicas, r over = 32 (i.e., r i = 8), same value used to calculate Saved resources. The total number of replicas of each configuration is shown in Figure 5.23. We used the Twitter Smoothed dataset, the Twitter linear -Classification application, the ANN model for input prediction, and Load-Aware grouping for stream grouping strategy.

Predictive approach

means that it greatly adapts itself in order to process most of incoming events (see Figure 5.24).

Comparison between PA-SPS and DABS-Storm

The aim of this experiment is to compare PA-SPS with DABS-Storm (denoted DABS), proposed in [START_REF] Kotto Kombi | DABS-Storm: A Data-Aware Approach for Elastic Stream Processing[END_REF] (see Section 3.2.2). We used the Twitter Smoothed dataset, the Twitter linear -Classification application, the ANN model for input prediction, and Load-Aware grouping for stream grouping strategy. For calculating the Saved resources metric, we have fixed r over = 32 (i.e., r i = 8). Table 5.12 gathers the metric values related to PA-SPS and DABS. PA-SPS is able to process all events, but not DABS as shown in Figure 5.25. The latter decreases by 17.2% the number of events processed. Similarly to Storm (see Section 4.1.1), DABS needs to restart the application for reconfiguring the number of replicas, which is not the case of PA-SPS due to their pool of replicas. Therefore, downtime has an impact in the number of events that are processed.

System

As we have already discussed, the increase of resources has a correlation with the decrease of latency, but there are scenarios where it does not apply. For example, in 5.5 Predictive approach DABS, saved resources have been decreased by 16.63% when compared to PA-SPS, but its latency is 291.36% higher. Such a behavior can be explained since DABS overestimates resources in non-critical intervals, as observed between t = 800s and t = 900s in Figure 5.26, which is useless in the case of curve peaks, where more resources are needed.

Complex application

We have also evaluated PA-SPS with a complex application. For this experiment, we have compared PA-SPS with an overprovisioning Storm which always uses a fixed number of replicas per operator, denoted S over . Such numbers are fixed (r i = 5) at the beginning of the data processing and do not vary during the experiment. We used the Twitter Smoothed dataset, Twitter complex application, and Load-Aware grouping for stream grouping. For calculating the Saved resources metric, we have fixed r over = 40 (i.e., r i = 5).

Other datasets

The aim of this section is to evaluate our system with other datasets, to analyse its adaptability and the behaviour of the results according to the predictive model used.

Twitter Raw

We used the Twitter Raw dataset, the Twitter linear -Classification application, and Load-Aware grouping for stream grouping strategy. For calculating the Saved resources metric, we have fixed r over = 32 (i.e., r i = 8).

Table 5.14 shows the results obtained, which have similar values with Table 5.10, related to smoothed data. PA-SPS, regardless the model, has processed most of the received events, with only 1.2% (resp., 1.3%) of events not processed by LR (resp., RF). Also, the lowest latency corresponds to ANN, although the difference in latency with respect to the second best model (Basic) is 0.47%.

By using a more unstable input rate (see Figure 5.28), the estimation error of the models increases as the input behaviour is more complex to predict. Since PA-SPS also becomes more unstable, the throughput degradation increases. FFT presents the highest difference because the input rate does not have a stationary behaviour. Consequently, there is a large percentage of error in the prediction of the input and the number of replicas which degrades performance (see Figure 5.27).

DNS

The aim of this experiment is to verify the adaptation ability of PA-SPS with an input with a different fluctuation than the previous inputs and then analyse the behaviour of it with each predictive model. We used the DNS dataset, the DNS application, and Load-Aware grouping for stream grouping strategy. For calculating the Saved resources metric, we have fixed r over = 12 (i.e., r i = 3). number of processed events. The highest difference percentage is around of 2.5% (RF) when compared to LR.

Figure 5.30 shows both the input rate and the throughput. Despite the high dynamics of the input rate, PA-SPS is able to adapt its number of resources in order to process the largest number of events in each time interval. In this experiment, the model with the best performance is FFT, having the lowest value of latency and throughput degradation. On the contrary, the input prediction error is the highest. Considering the values of saved resources, we can conclude that there was an overestimation of the input which led to an overestimation of resources as shown in Figure 5.29. If a model with lower resource utilisation is required, RF is a good choice, given that it has a difference of 22.08% of the saved resources value with respect to FFT. It is worth remarking that due to the above difference, FFT throughput degradation and latency decrease by 29.37% and 27.25% respectively.

Log

We used the Log dataset, the Log application, and Load-Aware grouping for stream grouping strategy. For calculating the Saved resources metric, we have fixed r over = 32 (i.e., r i = 8). Table 5.16 summarizes the obtained results. The processing capacity of PA-SPS is once again confirmed, where each proposed model has a none or a negligible difference of processed events and the highest difference percentage of events processed is around of 1.52% (ANN). Basic presents the best performance, both in terms of resource usage and latency. Figure 5.31 shows the amount of used replicas, where we can observe that the amount used by Basic does not vary much. Although there are high peaks of the input rate (see Figure 5 short for periods. Thus, it is more appropriate to use a constant amount of replicas rather than to adapt the SPS many times according to the input rate behaviour.

Discussion

In this section, we have seen the importance of the parameters, because depending on the size of the time interval, it increases or decreases the latency of the SPS.

Compared to DABS, evaluation results confirm the effectiveness of the dynamic replica adaptation of PA-SPS. In the experiments, latency decreases by 74.44% and saved resources increase 19.94%, when compared to DABS.

On the other hand, we observed that the most appropriate predictor model depends on the type of input rate behavior. In this way, we also corroborated that our solution is able to adapt to different types of applications and datasets.

Conclusion

This section evaluated two adaptive SPSs, RA-SPS and PA-SPS, which to dynamically adjust the number of operator replicas based on traffic fluctuations. Implemented on top of Storm, these adaptive SPS incorporate new features: a replica pool to prevent SPS restarts during changes in active replicas and Load-Aware grouping for balanced among replicas, which were evaluated improving in the performance of the adaptive SPS. RA-SPS, which is based on a reactive approach, adapts resources by analyzing shortterm traffic peaks. The State metric, composed of weighted metrics, categorizes operators as overloaded, stable, or underloaded to adapt the number of active replicas of the SPS. The evaluations showed an improvement in performance when using multic-metric compared to single-metric. PA-SPS, which is based on a predictive approach, adapts resources by analyzing historical input data and finding patterns in their behaviour. The evaluations corroborated the adaptability according to the predictive model, as well as the improvement in performance with respect to other proposals. It was also evaluated with different datasets and applications.

The core of this research work was the adaptation of a SPS by dynamically modifying the number of operator replicas without reconfiguring the SPS. Our models adapt in an online fashion reducing event loss and end-to-end latency in the presence of data stream fluctuations.

Contributions

This thesis proposes two adaptive SPSs, RA-SPS and PA-SPS. By using a reactive and predictive approach respectively, they automatically adapt their processing logic cope with traffic dynamics. Both SPSs are an extension of Storm and include the pool of replicas and the load-aware grouping. Results showed that the use of a pool of replicas increases by 13.6% of the total number of processed events, due to the non-existence of downtime in the reconfiguration of the application. It also shows a negligible increase in CPU usage, demonstrating that the use of inactive replicas is a viable proposal in terms of cost. Regarding the use of Load-Aware grouping, the experiments show a 60.18% decrease in latency compared to Shuffle Grouping, and a negligible computational cost with only a 0.9% increase in CPU usage.

The work proposes two adaptive SPS: RA-SPS (a reactive mechanism) and PA-SPS (a predictive mechanism). Both SPSs are based on a planning algorithm that modifies the number of active replicas of an operator. Moreover, for the automation of the adaptive feature, a control-loop-based on a MAPE model was proposed. It is composed by four modules: Monitor, Analysis, Plan, and Execution. The Monitor module is responsible for the recollection of statistics necessary for the analysis of the system, either to determine its state or to predict its behaviour. The Analysis module is in charge of performing the calculations necessary for system planning, thus, this module varies according to the approach used. The reactive approach determines the state of the operator, in order to activate o deactivate replicas, and the predictive approach, determines the number of replicas needed according to the prediction models. The Plan module determines the number of activated replicas to be modified according to the defined planning algorithm. Finally, the Execution module performs the modification to the SPS.

The reactive RA-SPS (Section 4.2) bases its decision on the state of an operator at runtime. To this end, three metrics were proposal: the load of the operator (U ), the average execution time of an event (E), and the operator input queue (Q). The aim is to adapt the number active operator of replicas according to the analysis of traffic peaks in short periods of time. These metrics are integrated in a function, assigning weights to each of them. Based on the function value, the SPS can decide whether an operator is overloaded, underloaded, or stable, and therefore increases, reduces, or maintains the same number of active replicas. Performance results based on Twitter data confirm the advantages of using the three metrics compared to a single one. The use of a multi-metric increases total event processing by up to 181.06% compared to a single-metric.

The PA-SPS (Section 4.3) dynamically adapts the active number of operator replicas according to the behaviour of the input data. It aims to find patterns in the traffic to anticipate possible overloads or underloads in the SPS, and then determine the number of replicas needed by operator O i during the next time interval. This value is defined by the predicted number of events received by the operator O i , which is defined by the predicted number of events sent by the input data and the number of queued events. In addition, prediction also takes into account the cascade effect in the DAG, due to the dependence among operators. For the prediction of events sent by the input data, the use of a predictive model based on mathematical model or artificial intelligence models was proposed, whose performance will depend on the behaviour of the data flow. Evaluation results confirm the effectiveness of the dynamic replica adaptation of PA-SPS. In the experiments, latency decreased by 74.44% and saved resources increases by 19.94%, when compared to DABS. On the other hand, we observed that the most appropriate predictor model depends on the type of input rate behavior. For instance, in the Twitter application scenario, the best performing model was ANN. In contrast to the DNS scenario, where FFT performed better in latency and throughput degradation. And finally, in the Logs application scenario, the best performing model was Basic, both in resource usage and latency.

Future work

This section discusses some future directions of the current work.

Short term

• Extend both RA-SPS and PA-SPS in order to manage stateful operators and the perform new experiments to evaluate the cost of managing operator states.

• Considerer the parameter values of Algorithm 2 (Table 4.1) as adaptive, i.e., they would vary according to the application execution state (e.g., operator load, input rate fluctuation, etc.). To this end, we could use an artificial intelligence model to determine the parameterization according to the history, as well as a training database.

• Add conditions to Algorithm 3 in order to not to overload the physical resources (VMs or nodes) of the SPS. For this purpose, a global analysis of the system could be performed using the metrics presented in RA-SPS.

• Evaluation of our solution against an existing benchmark, such as [START_REF] Shukla | RIoTBench: An IoT benchmark for distributed stream processing systems[END_REF] or [START_REF] Arasu | Linear Road: A Stream Data Management Benchmark[END_REF].

Mid term

• Design and implementation of an adaptive system that considers logical and physical resources. Therefore, in addition to the planning algorithm already proposed, we would have another one that determines whether it is necessary to modify the physical resources, either to reduce costs or to increase performance. For its design, a hierarchical system that considers both global and local components (operators) would modify the necessary resources according to different states or predictions.

• Deployment of applications using specific VMs in the Cloud, which have lower cost. For instance, in the case of GCP, the so-called E2 shared-core machines1 make it possible to increase the number of cores for a short period of time without the need to modify or restart the VM. So it would be possible to vertically scale the physical resources, while scaling the logical resources as well.

• Implement a hybrid adaptive SPS, which considers the reactive and predictive approaches proposed in RA-SPS and PA-SPS. The planning algorithm would be determined by both current and future time window contents. To this end, a possible solution would be to have hierarchy of resource adaptations, according to the analyses performed by the two approaches.

Figure 1 .Fig. 1 . 1 :

 111 Figure 1.1 presents an example of interactions between users and applications, such as uploading or downloading music, commenting on a post, and receiving notifications, among many other possibilities.

Fig. 1 . 2 :

 12 Fig. 1.2: Batch processing concept.

Fig. 1 . 3 :

 13 Fig. 1.3: Stream processing concept.

Fig. 1 . 4 :

 14 Fig. 1.4: Example of a SPS application.

Fig. 2 . 1 :

 21 Fig. 2.1: Streaming example among server and clients.

Fig. 2 . 2 :

 22 Fig. 2.2: Stream processing paradigm.

Fig. 2 . 3 :

 23 Fig. 2.3: Example of the representation of the DAG in an SPS application.

Fig. 2 . 4 :

 24 Fig. 2.4: DAG in an SPS.

Figure 2 . 2 . 2

 222 5 presents a complex DAG, where Stream Processing Systems

Fig. 2 . 5 :

 25 Fig. 2.5: DAG complex in an SPS.

Fig. 2 . 6 :

 26 Fig. 2.6: The tuple exchange in an SPS.

Fig. 2 . 7 :

 27 Fig. 2.7: Example of parallelism of an SPS operator.

Fig. 2 . 8 :

 28 Fig. 2.8: An example of stateless operator.

Fig. 2 . 9 :

 29 Fig. 2.9: An example of stateful operator.

4 Fig. 2 .

 42 Fig. 2.11: Stream grouping in a Storm Topology.

Figure 2 . 2 , 2 = 1 ,ExecutorFig. 2 .

 22212 Figure 2.12 shows the parallelism of the Storm Topology of Figure 2.10. The Topology has three subsets represented by Workers processes, which are distributed on physical machines. Each Bolt and Spout represent a Task, and each Task is associated with a single Executor, except in the case of Bolt 2 , where all Tasks are associated with a Executor. In summary, the number of associated threads for each Bolt is Bolt 1 = 2, Bolt 2 = 1, Bolt 3 = 3 and Bolt 4 = 1.

Fig. 2 .

 2 Fig. 2.13: Storm physical architecture.

Finally, 2 Fig. 2 . 14 :

 2214 Fig. 2.14: Storm architecture.

Figure 2 .SourceFig. 2 . 15 :

 2215 Figure 2.15 shows a Flink Application (DAG) example, composed of a Source, and three Operators. In this example, can observe that the first Operator performs the mapping task, while the second Operator groups by a key and operates in a time window. These Operators are named Transformation Operators. After, the third

Figure 2 .Fig. 2 .

 22 Figure 2.16 shows the scheduling of the Flink Application (shown in Figure 2.15) in a physical platform. According to some scheduling algorithms, each Operators is placed in an available Task.

  Response of SPS reconfiguration.

Fig. 3 . 1 :

 31 Fig. 3.1: Manual adaptation of a SPS.

Fig. 3 . 2 :

 32 Fig. 3.2: Automatic adaptation in a SPS.

Fig. 3 . 3 :

 33 Fig. 3.3: Thresholds in a reactive model.

Chapter 3 Fig. 3 . 4 :

 334 Fig. 3.4: Adaptive SPS using a reactive approach.

Figure 3 .Fig. 3 . 5 :

 335 Figure 3.5 shows an example of a predictive approach, where the system analyses CPU utilisation. With the sample history of this statistic, a prediction is made about the future behaviour of the system based on some predictor. Finally, based on the prediction, the model determines a possible state of the SPS, and if necessary, mitigate the possible CPU overloads.

Figure 4 .

 4 Figure 4.1 considers a DAG with operators O A and O B and their respective pool of replicas of size p. When initialising the application, there is only one active replica per operator as shown in Figure 4.1a there is only one active replica per operator (O A.1 and O B.1). Therefore, the other replicas are inactive and do not receive a data stream. It is important to note that it is necessary to keep one replica active, since if all replicas are inactive, the DAG components related to the operator would be lost. The state of an operator can be modified, as is the case of O A.2 , which changes from inactive to active as shown in Figure4.1b. When this is done, the stream data sharing is determined by the grouping strategy of the operator (see Section 4.1.2).

Fig. 4 . 1 : 40 Chapter 4

 41404 Fig. 4.1: Example of pool replicas for two operators.

Fig. 4 . 3 :

 43 Fig. 4.3: Example of the Execution Time metric.

Fig. 4 . 4 :

 44 Fig. 4.4: Example of the Queue metric.

Fig. 4 . 5 :

 45 Fig. 4.5: Example of the evolution of the State metric in the operator O i .

O 1 Fig. 4 . 9 :

 149 Fig. 4.9: DAG example of predicted received number of events according to Equation 4.10.

Fig. 5 . 1 :

 51 Fig. 5.1: Development environment on GCP.

Fig. 5 . 2 :Fig. 5 . 3 :

 5253 Fig. 5.2: Traffic shape of Twitter Gaussian dataset.

Figure 5 Fig. 5 . 4 :

 554 Fig. 5.4: Traffic shape of Twitter Smoothed dataset.

5. 1 Fig. 5 . 5 :

 155 Fig. 5.5: Traffic shape of DNS dataset.

Fig. 5 . 6 :

 56 Fig. 5.6: Traffic shape of Log system dataset.

Figure 5 Fig. 5 . 7 :

 557 Fig. 5.7: Detection Twitter application in SPS.

Fig. 5 . 9 :

 59 Fig. 5.9: Classification Twitter application in SPS.

5. 1 Fig. 5 . 10 :Fig. 5 .

 15105 Fig. 5.10: Log application in SPS.

Fig. 5 .Fig. 5 .

 55 Fig. 5.12: Total number of replicas of Storm-Pool and Storm-Default.

Figure 5 . 5 . 2 Fig. 5 .

 5525 Figure 5.15 evaluates the latency for both implementations. The difference in latency is 92.46% between the two systems. The availability of the system explains this huge

Fig. 5 . 15 :

 515 Fig. 5.15: Comparison of latency between Storm-Default and Storm-Pool.

Fig. 5 .

 5 Fig. 5.16: Comparison of latency between Load-aware grouping and Shuffle grouping.

Fig. 5 .

 5 Fig.5.17: Total number of active replicas of δ, U, and Q.

Fig. 5 .

 5 Fig. 5.18: Throughput of δ, U, and Q.

Fig. 5 .

 5 Fig. 5.19: Total number of processed events of δ, U, and Q.

Fig. 5 . 20 :

 520 Fig. 5.20: Latency of δ, U, and Q.

Fig. 5 .

 5 Fig. 5.22: Total number of replicas of PA-SPS using different predictor models.

Fig. 5 .

 5 Fig. 5.24: Throughput using PA-SPS and different configurations in Storm.

Fig. 5 .Fig. 5 . 26 :

 5526 Fig. 5.25: Throughput using PA-SPS and DABS.

Fig. 5 .

 5 Fig. 5.27: Total number of replicas of PA-SPS using Twitter raw dataset.

5. 5 Fig. 5 .Fig. 5 .

 555 Fig. 5.29: Total number of replicas of PA-SPS using DNS dataset.

Fig. 5 .Fig. 5 . 5 . 5

 5555 Fig. 5.31: Total number of replicas of PA-SPS using Log dataset.

2.2 Stream Processing Systems of

  the operator v 2 , where k replicas of the operator are created to process data in parallel.

	v 1	v 2	v 3
	(a) Representation by DAG.
		2.1
		v 2.1
	1	2.2	3
	v 1	v 2.2	v 3
		2.k
		v 2.k
	(b) Representation of operator parallelisation.

  The number of replicas for Bolt 1 , Bolt 2 , Bolt 3 and Bolt 4 are 2, 3, 4 and 1 respectively. Spout is in charge of sending the raw data to Bolt 1 by distributing the stream data input to each of Bolt 1 's replicas based on some stream grouping. After processing the tuple, Bolt 2 sends the processed tuple to Bolt 2 and Bolt 3 , the following Bolts downstream in the DAG. If there is not Bolt downstream, we are in presence of a sink operator and dataflow is terminated. The last operator that processes the data in Topology is Bolt 4 .

		Topology
		Bolt 2 Bolt 2 Bolt 2
	Spout	Bolt 1 Bolt 1	Bolt 4
		Bolt 3 Bolt 3 Bolt 3 Bolt 3
	Fig. 2.10: A Storm Topology and components.
	Figure 2.10 shows an Storm Topology (DAG) example, composed of Spout, and four
	Bolts.

  .1.

	Framework Client	Mod. of operators	Mod. of nodes used	Reconf. downtime	State restoration
	Storm	GUI / Terminal cmd	Yes	Yes	Yes	Manual
	Flink	GUI / Terminal cmd	Yes	Yes	Yes	Automatic

Tab. 3.1:

Comparison of manual adaptation of SPS.

Table 3 .

 3 2 shows a comparison of the discussed adaptive SPSs that use a reactive approach. Objective and Model are the objective function and model used for adaptation respectively. VM Scaling indicates whether it uses horizontal or vertical scalability of the physical resources (VMs). Stateful operator indicates whether the SPS supports type of operator. Then the Infrastructure used for its deployment is indicated. Finally the used SPS framework, None means that the work does not use any framework, because it has implemented its own SPS.

	Reference	Objective	Model	VM Scaling	Stateful Operator	Infrastructure	SPS Framework
	[Gul+12]	CPU	Threshold Non Yes	Cloud	Borealis
	[Rus+21]	Throughput	Regression Yes Non	Cluster	Storm
	[MZS16]	Latency	Threshold Non Yes	Cloud	Storm
	[KG20]	Throughput	Heuristic	Non Yes Single Machine	None
	[Ged+14]	Throughput	Threshold Non Yes	Cluster	None
	[Sch+09]	Throughput	Heuristic	Non Non Single Machine	None
	[Fer+13]	Throughput	Threshold	Yes	Yes	Cloud	None
	[Sat+11]	Latency	Threshold Non Non	Cluster	None
	[Hei+14]	Latency	Heuristic	Yes	Yes	Cluster	None
	[ABM10] CPU, Network Threshold	Yes Non	Cloud	None

Tab. 3.2:

Comparative table of adaptative SPS that use reactive approach.

Table 3 .

 3

3

shows a comparison of adaptive SPSs using predictive approach. Objective and Model are the objective function and predictive model used for adaptation. VM Scaling indicates whether it uses horizontal or vertical scalability of the physical resources (VMs). Stateful operator indicates whether it supports this type of operator. Then the Infrastructure used for its deployment is indicated. Finally, the SPS framework used. None means that a frameworks is not used, because it implemented its own SPS.

3

  Related work on adaptive| SPS

	Reference	Objective	Model	VM Scaling	Stateful Operator	Infrastructure	SPS Framework
	[Car+18]	Latency	Reinforcement learning	Non Yes	Cluster	Storm
	[Ark+21] Throughput	Model-based	Yes	Yes	Cluster	Flink
	[Kom+19] Throughput	Time series	Non Yes	Cluster	Storm
	[Lom+18] Utilization	ANN	Yes	Yes	Cluster	Storm
	[BTÖ13]	Utilization	Heuristic	Non Non	Cluster	Borealis
	[KSP14]	Throughput	Heuristic	Non Non	Simulation	None
	[MTD18] Throughput	Fuzzy logic	Non Non Single Machine FastFlow
	[HWR17] Throughput Markov Chain Non Non	Cluster	S4

Tab. 3.3:

Comparative table fo adaptative SPS that use predictive approach.

  .1 summarises all parameters used in RA-SPS.An adaptive SPS should define metrics to characterize the state of the operators at runtime on a given scenario. Traditional metrics such as throughput, latency, and CPU are the most used in literature (see Section 3.2.1).Utilization metric is defined by Equation4.2. The aim of the metric is to characterize the operator load: if its value is close to 1 (resp., 0), the operator is overloaded (resp., underloaded). Figure 4.7 shows an example composed of three independent operators (O 1 , O 2 , and O 3 ) that have different U i (t).Execution Time metric is defined by Equation4.3. The aim of the metric is to characterize execution degradation of operators: if the value of et i (t) is greater than et i , the physical machines are overloaded. It is a QoS metric which allows to detect struggle operators. Therefore, the difference between et i (t) and et i is that the latter is the average execution time of one event processed by an operator O i without any extra load. The variable et i is considered as a baseline, and it is estimated by previously benchmark execution. Reactive approach In Figure4.3, the three operators have an average execution time of 200 milliseconds (et i = 100). O 1 has a large extra load, possibly due to overutilisation of resources, so the execution time is much longer. Unlike O 3 , which has no extra load, or O 2 which has less extra load.

	4.2.1 Metrics
				In RA-SPS, we propose to
	integrate the metrics denoted Utilization (U ), Execution Time (E), Queue (Q) and
	State (δ).
		O 1 O 1	O 2 O 2	O 3 O 3
		U i (t) =	µ i (t) × et i (t) r × td	(4.2)
	Parameter Description O i operator i In Figure 4.2, the three operators have 2 replicas (r O 1 O 2 O 1 O 2	O 3 O 3
	t	time interval number
	td	time interval duration
	et i (t)	average execution time of one event by O i during t
	et i	average execution time of one event by O i according to the benchmark
	q i (t)	queue of events received and not processed by O i at the end of t
	µ i (t)	number of events processed by O i during t
	r i (t)	number of active replicas of O i during t
	U i (t)	Utilization metric by O i during t
	E i (t)	Execution Time metric by O i during t
	Q i (t)	Queue metric by O i during t
	δ i (t)	State metric by O i during t
	Tab. 4.1: Parameters notation and their description in RA-SPS. E i (t) = 1 -et i et i (t)	(4.3)

i = 2) and an average execution time of 200 milliseconds (e i = 200 ms). As the number of events processed is different during the time interval t for each operator, their loads are different. O 1 has no idle time, so it is at maximum utilization, so the operator is overloaded. O 2 has a load level which is 50% of the time idle in time period t. O 3 is almost underloaded. Fig. 4.2: Example of the Utilisation metric. 4.2

:

  Statistics Operator O i in time interval t. Ensure: Modifying the current number of active replicas of operator O i .

1:

  Figure 4.7 shows an example composed by three independent operators (O 1 , O 2 , and O 3 ) which have different average execution time of one event (et i ). At the beginning of t, all the three operators have two replicas (r i = 2). And by prediction, they will receive the same number of events λ i (t + 1). Example of the number of replicas calculation, according Equation 4.6.

		Parameter Description
		O i	operator i
		t	time interval number
		td et i	time interval duration average execution time of one event by O i O 1 O 2 O 1 O 2	O 3 O 3
		q i (t)	queue of events received and not processed by O i at the end of t
		λ G (t)	number of events sent by input data during t
		λ r i (t)	number of events received by O i during t
		λ p i (t) Fig. 4.7:	number of events received by O i sent from O p during t
		µ i (t)	number of events processed by O i during t
		θ x (t)	percentage of events processed of λ G (t) by O x during t
		O p i	predecessor operator of O i in the SPS DAG
		θ p i (t)	percentage of events produced by O p i sent to O i during t
		λ G (t + 1) predicted number of events sent by input data during t + 1
		λ i (t + 1)	predicted number of events to process by O i during t + 1
		λ r i (t + 1)	predicted number of events received by O i during t + 1
		λ q i (t + 1) predicted number of queued events to be processed by O In this example, due to et i 's differences, Equation 4.6 will render r 1 (t + 1) = 2,
		r 2 (t + 1) = 1, and r 3 (t + 1) = 4 at the end of t. Such results inform that the number
		of O 1 's active replicas should not change but that of O 2 (resp., O 3 ) is overestimated
		(resp., underestimated) and should be reduced (resp., increased) to one (resp.,
		four).
	48	Chapter 4 Proposed adaptive SPS	4.3 Predictive approach

i during t + 1 r i (t + 1) number of replicas of O i computed at the end of t Tab. 4.2: Parameters notation and their description in PA-SPS.

execution time of one event at the operator (et i ). At the end of each interval, the number of replicas is calculated following Equation 4.6.

r i (t + 1) = λ i (t + 1) × et i td (4.6)

Let us consider that the time interval duration (td) equals 1000 ms.

Algorithm 3

 3 Adaptive planning algorithm according to the predictive approach for the operator O i .

	4:	Add k i active replicas to O i
	5: else if k i < 0 then
	6:	Remove k i active replicas from O i
	7: end if

Require: Statistics Operator O i in time interval t. Ensure: Modifying the current number of active replicas of operator O i .

1:

r i (t + 1) ← computeReplicas( λ i (t + 1) , et i , td) 2: k i ← r i (t + 1) -getReplicas(O i ) 3: if k i > 0 then 4.3 Predictive approach

Table 5 .

 5 1 summarizes RA-SPS parameters and their respective values. These values were determined according to expertise and learning outcomes obtained from the results of several previous experiments using different parametrizations.

	Parameter Description	Value
	td	Time interval duration	25 sec
	δ u	Operator state upper limit	0.7
	δ l	Operator state lower limit	0.3
	δ E	Limit for adding replicas	0.7
	ω U	U metric weight	0.45
	ω Q	Q metric weight	0.45
	ω E	E metric weight	0.1
	k	Number of active replicas to add/remove	1

Tab. 5.1: RA-SPS parameters and their values.

Table 5 .

 5 2 summarizes PA-SPS parameters and their respective values. To determine the values of predictive model parameters, the model has considered obtaining a representative sampling based on the methodology presented in[START_REF] Winters | Practical predictive analytics[END_REF].

	Parameter Description

Table 5

 5

	.3 summarizes the results obtained for

Tab. 5.3: System metric values of Storm-Pool and Storm-Default using Twitter Gaussain

dataset.

  System metric values of Load-Aware and Shuffle grouping using Twitter Gaussain dataset.

		Saved	Throughput	Diff. Proc.	Latency
		Resources	Degradation	Events	(ms)
	Shuffle	0.5390	0.4332	0.9979	5271.01
	Load-Aware 0.5617	0.1831	0.9987	2098.91
	Tab. 5.4:

  System metric values with different timeout using Twitter Smoothed dataset.

	Timeout	Saved Resources	Throughput Degradation	Diff. Proc. Events	Latency (ms)
	t out = 1s	0.6164	0.1000	0.9030	369.35
	t out = 5s	0.5875	0.1038	0.9303	946.69
	t out = 10s 0.5633	0.1056	0.9529	1298.45
	t out = 30s 0.5617	0.1831	0.9987	2098.91
	Tab. 5.5:

  System metric values with different queue size using Twitter Smoothed dataset.

	Queue size	Saved Resources	Throughput Degradation	Diff. Proc. Events	Latency (ms)
	q size = 100	0.6719	0.1919	0.6835	29.91
	q size = 1000	0.6656	0.1375	0.6922	311.93
	q size = 10000	0.6602	0.1687	0.7249	1394.56
	q size = 100000 0.5617	0.1831	0.9987	2098.91
	Tab. 5.6:
				5.3 Impact of Storm parameters

  Then, in the second period, since the U experiment only analyzes if another replica is necessary to improve operator utilization, it continues to increase the active replicas, System metric values of δ, U, and Q using Twitter Smoothed dataset.

		Saved	Throughput	Diff. Processed	Latency
		Nodes	Degradation	Events	(ms)
		δ 0.3996 0.1092	0.8907	39687.51
		U -0.8934 0.2597	0.7402	23441.39
		Q 0.4975 0.6830	0.3169	28799.60
		Tab. 5.7:
	68	Chapter 5 Experimentation

  System metric values of RA-SPS and S over using Twitter Smoothed dataset and Twitter complex application.

	System	Saved Nodes	Throughput Degradation	Diff. Processed Events	Latency (ms)
	δ	0.5023 0.4252	0.98	179.5401
	S over	0	0	1.00	31.990
	Tab. 5.8:
					5.4 Reactive approach

Table 5 .

 5 13 shows evaluation results obtained with PA-SPS and Storm. In PA-SPS, we observe a high reduction of used resources with 68.8% fewer active replicas, when compared to Storm. Such a decrease has an impact on the physical used resources: CPU consumption of PA-SPS (resp. Storm) is in average, 9.57% (resp. 14.66%). This difference happens because each replica is associated with a thread. Therefore, with a fixed number of 5 replicas, Storm requires more CPU than PA-SPS where the number of replicas dynamically varies.

	System	Saved Resources	Throughput Degradation	Diff. Proc. Events	Latency (ms)
	PA-SPS 0.688	0.031	1.000	209.270
	S over	0.000	0.000	1.000	31.990

Tab. 5.13: System metric values of PA-SPS and S over using Twitter Smoothed and Twitter complex application.

Table 5 .

 5 15 summarizes the obtained results. The processing capability of PA-SPS is confirmed since each proposed model has none or a negligible difference in the Throughput of PA-SPS using Twitter raw dataset.

	80	Chapter 5 Experimentation

Tab. 5.15: System metric values of PA-SPS using different predictive models using DNS dataset.

  .32), as in t = 700s, they are System metric values of PA-SPS using different predictive models using Log dataset.

	Pred.	Saved	Throughput	Diff. Proc.	Latency	Error Est.	Error Est.
	Model	Resources	Degradation	Events	(ms)	Input	Replica
	ANN	0.603	0.236	0.987	905.290	0.355	0.537
	FFT	0.503	0.572	1.000	9184.060 0.746	1.191
	LR	0.565	0.252	0.994	1021.800 0.412	0.413
	RF	0.661	0.335	0.998	1673.560 0.243	0.394
	Basic	0.655	0.306	0.989	855.970	0.250	0.449
	Tab. 5.16:

1.2 Publications

2.2 Stream Processing Systems

Chapter 2 Background

2.3 SPS Frameworks

Chapter 3 Related work on adaptive| SPS

Tab. 5.2: PA-SPS parameters and their values.

http://github.com/apache/storm/blob/master/conf/defaults.yaml5.1 Experiment scenario

Load-Aware grouping has only been implemented in PA-SPS, so RA-SPS cannot use it.5.2 Impact of the new features

In this experiment to calculate Equation

4.11, the dependence among operators was not considered, because it was tested with a previous version of the prediction model.

Chapter 5 Experimentation

Chapter 6 Conclusion

https://cloud.google.com/compute/docs/general-purpose-machines#e2-shared-core 6.2 Future work

Bibliography

Remerciements

Experimentation 5

The goal of this chapter is to evaluate our adaptive SPS, both the reactive and predictive approaches. In this way, we will analyze the impact of adaptation on the performance of the SPS. Section 5.1 presents the experimental scenario, specifying the environment, the dataset, the parameters and the evaluation metrics. Section 5.2 presents the evaluation of our extended version our extended version of Storm, which uses a replica pool and Load-aware grouping. Section 5.4 presents the evaluation and analysis of RA-SPS, using different configurations in their weights and applications. Finally, Section 5.5 presents the evaluation and analysis of PA-SPS, both of the parametrization used as well as a comparison with other existing adaptive SPS.

Experiment scenario

In this section, we describe the components necessary for the creation of an experiment scenario, which has three components: environment, dataset and application. For each experiment, we considered a set of measures for which the difference in behaviour was negligible. Section 5.1.1 presents the environment that is the architecture used for the deployment of the adaptive SPS. Section 5.1.2 presents the dataset refers to the traffic used to provide an input data stream. Section 5.1.3 presents the applications designed with its respective processing according to the DAG and the type of data used. Section 5.1.4 presents the parameters used by the adaptive SPSs. Finally, Section 5.1.5 presents the evaluation metrics.

Environment

All the experiments were conducted on the Google Cloud Platform (GCP) using seven Virtual Machines (VMs): three in charge of Zookeeper, seven as Supervisor nodes, a loss of only 0.4% of the incoming events. Therefore, we all models are reliable to be used in whole event processing experiments.

Pred. Model

We can also observe that ANN has the lowest latency, with a difference of 39.12% compared to the second lowest latency model (RF). Such values mean that PA-SPS using ANN processes the received events faster than the others. However, in this case, it needs a larger amount of resources, as shown by the saved resources metric, where ANN has the lowest value, i.e., 15.17% worse than the Basic model metric. Thus, there is a trade-off between latency and the amount of resources: on the one hand, if the requirement is a SPS that processes all incoming events with low latency, ANN is the most suitable model; on the other hand, if the aim is the reduction of costs and the loss of events is not an issue, having an acceptable latency, RF is more suitable. Regarding the estimation error of the input rate and number of replicas, a lower estimation error can not be interpreted as better performance. Overestimating the input rate implies an overestimation of the number of replicas, using a larger Table 5.11 summarizes the results of the different configurations. We observe that the system without replication (r = 1) has a very low performance, because it only processes 33.1% of the incoming events (see Figure 5.24). Such a result is due to the lack of adaptation when incoming events increase. Consequently, there exists a bound for the number of events to process while the others are queued. Therefore, although such a configuration presents low resource usage, it is not recommended for performance sake. On the other hand, the r = 4 configuration has only a 1.3% difference between incoming and processed events. The decrease in the amount of resources by 50% increases latency by 43.03%. Thus, once again, there is a tradeoff between performance and used resources, corroborating to our previous discussion.

System

Finally, PA-SPS decreases by 5% Saved Resources with respect to r = 4, but it is able to process all incoming events. Compared to the r = 8 configuration PA-SPS presents:

(1) a 131.57% higher latency, whose impact should be balanced with its ability to dynamically adapt itself; (2) a difference of 7.01% in throughput degradation, which