Ecole Sigle

Doctorale

Renée El

Melhem Bât

Pascal Blaise

R Gourdon

M Stéphane

ÉLECTROTECHNIQUE E E A Électronique

M Gérard Scorletti
email: gerard.scorletti@ec-lyon.fr

M Philippe
email: philippe.normand@univ-lyon1.fr

Mme Emmanuelle
email: emmanuelle.canet@univ-lyon1.fr

Étage

M Luca

Jean-Yves Buffière
email: jean-yves.buffiere@insa-lyon.fr

M Jocelyn Bonjour

Viviane Polsinelli
email: viviane.polsinelli@univ-lyon2.fr

Brigitte Dubois

J Y Toussaint

M Christian Montes

Roland Kotto Kombi

Nicolas Lumineau

Philippe Lamarre

" Approche

Cécile Favre

Chloé Artaud

Clément Duau

Ophélie Fraisier

Roland Kotto

" Stream

Posters

1.1 Traitement distribué et élastique de ux de données Ces travaux ont été réalisés dans le cadre du projet ANR Socioplug 1 (ANR-13-INFR-0003), nous nous intéressons à une plateforme de traitement de ux distribuée où chaque unité de traitement est un nano-ordinateur connecté à Internet ayant des capacités de traitement et de stockage limitées. Chaque utilisateur dispose d'une unité de traitement et peut interroger un ensemble de ux via des requêtes continues. An de garantir le traitement de ces requêtes indépendamment du débit des ux d'entrée, la plateforme regroupe tous les utilisateurs intéressés par les mêmes résultats en une communauté. La requête continue caractérisant une communauté est alors distribuée et exécutée une seule fois sur l'ensemble des unités de traitement associées à la communauté. Cela permet aux utilisateurs d'interroger des ux potentiellement massifs tout en leur donnant le contrôle sur le traitement de leurs données et leurs requêtes. 1.2 Problématique En considérant des ux variants, deux phénomènes peuvent être observés ; phénomènes pouvant être problématique du point de vue de l'utilisateur, si cela implique une dégradation de la qualité des résultats et du point de vue du fournisseur de ressources si des ressources allouées ne sont pas exploitées. Le premier phénomène dit de sur-allocation apparait lorsque le débit des ux d'entrée est largement inférieur au débit de traitement des tâches d'un opérateur. Cela implique que le degré de parallélisme est trop important et donc que des ressources superues sont sollicités par le SGFD. Ces ressources peuvent être libérés an de bénécier à d'autres tâches. Le second phénomène dit de risque de congestion apparait lorsque des tâches ne peuvent plus traiter les n-uplets aussi vite qu'ils arrivent. En eet, le débit de traitement d'un opérateur est borné par la complexité de la fonction qu'il implémente. Dans ce cas, les nuplets en entrée de la tâche s'accumulent en attendant d'être traité ce qui rallonge la latence de traitement globale (i.e., le temps mis par un n-uplet pour traverser le graphe d'opérateurs). Lorsque le volume de n-uplets accumulés en entrée d'une tâche dépasse la quantité de mémoire prévue au stockage temporaire des données, cela entraine la congestion de l'opérateur. Dès lors, l'opérateur congestionné entraine la perte irréversible de n-uplets et requiert davantage de ressources disponibles pour traiter l'intégralité de ses entrées. Or, le SGFD n'a pas de contrôle sur l'évolution des ux d'entrée du point de vue du volume et de la distribution des valeurs. Il est donc nécessaire d'adapter l'exécution d'une requête continue sur l'infrastructure distribué 1. http://socioplug.univ-nantes.fr/index.php/SocioPlug_Project 2 Cette thèse est accessible à l'adresse : http://theses.insa-lyon.fr/

Résumé étendu en français de la thèse intitulée Distributed query processing over uctuating data streams

Roland Kotto Kombi 1 Introduction et motivation

Avec la démocratisation d'applications consommant des données en temps réel (géolocalisation, recommandations ciblées, appareils intelligents, etc.), de larges volumes de données, dits ux de données, transitent entre les sources ayant produites ces données et des services de traitement. Un ux peut être représenté comme une séquence potentiellement innie de données estampillées et émises en temps réel. Les utilisateurs peuvent interrogés un ensemble de ux via un langage de requête dédié [START_REF] Arasu | The cql continuous query language : Semantic foundations and query execution[END_REF]2,3]. Cependant, à l'inverse des requêtes sur des données statiques, les requêtes sur des ux de données génèrent de nouveaux résultats dès que de nouveaux n-uplets sont reçus en entrée et sont alors dites continues. Ces requêtes continues sont traitées par des systèmes de gestion de ux de données (SGFD) académiques [4,5,6],

industriels [2,[START_REF] Biem | Ibm infosphere streams for scalable, real-time, intelligent transportation services[END_REF][START_REF] Akidau | Millwheel : faulttolerant stream processing at internet scale[END_REF] et libres [START_REF] Peng | Rstorm : Resource-aware scheduling in storm[END_REF][START_REF] Carbone | Apache ink : Stream and batch processing in a single engine[END_REF][START_REF] Noghabi | Samza : stateful scalable stream processing at linkedin[END_REF] reposant sur une infrastructure distribuée. Une infrastructure distribuée est composée d'un ensemble d'unités de traitement dénies par des ressources initiales (processeur et mémoire) potentiellement partagées.

An de répartir l'utilisation des ressources nécessaires à l'exécution d'une requête continue, un SGFD distribué transforme chaque requête soumise par un utilisateur en un graphe d'opérateurs [3,[START_REF] Peng | Rstorm : Resource-aware scheduling in storm[END_REF][START_REF] Carbone | Apache ink : Stream and batch processing in a single engine[END_REF]. En fonction de son type, un opérateur peut être répliqué en plusieurs tâches. Ce nombre de tâches est appelé le degré de parallélisme de l'opérateur. Les tâches sont ensuite aectées sur des unités de traitement. Il est important de noter que le ux en entrée de chaque tâche a potentiellement un débit variant en terme de volume et de distribution des valeurs au cours du temps. La qualité des résultats générés par les requêtes continues émises par l'utilisateur dépend de la capacité de l'infrastructure à absorber la charge induite par les ux 3.1 Optimisation algébrique À l'instar des SGBD, certains SGFD [START_REF] Arasu | Stream : The stanford data stream management system[END_REF]3,5] permettent aux utilisateurs de dénir des requêtes à l'aide d'un langage déclaratif. La requête correspond alors à une expression traduite par le SGFD en une topologie équivalente. En fonction des propriétés algébriques des opérateurs, le SGFD peut chercher l'ordonnancement optimal des opérateurs an de minimiser le volume de données transitant entre les opérateurs. Cette phase d'optimisation se base sur les solutions éprouvées depuis de nombreuses années dans les systèmes de gestion de base de données (SGBD).

Parallélisation des opérateurs

Une fois un graphe d'opérateurs choisi, un SGFD distribué peut xer le degré de parallélisme de chaque opérateur. Ce choix a un impact direct sur le débit de traitement théorique de l'opérateur et donc le débit maximal que l'opérateur peut recevoir en entrée sans accumuler de n-uplets sur sa le d'attente. Il apparait donc évident qu'un degré de parallélisme sousévalué par rapport au débit réel du ux entrainera à terme une congestion.

À l'inverse, un degré de parallélisme signicativement surévalué implique une réservation de ressources superues pour les besoins de traitement.

Allocation des tâches

Lorsque le degré de parallélisme de tous les opérateurs est xé, le SGFD doit décider du plan d'aectation des tâches sur les unités de traitement. Ce plan d'allocation est déni selon une stratégie d'allocation. Certaines stratégies [START_REF] Xu | T-storm : Trac-aware online scheduling in storm[END_REF][START_REF] Aniello | Adaptive online scheduling in storm[END_REF] ont pour objectif de minimiser les échanges réseaux entre les unités de traitement tandis que d'autres stratégies [START_REF] Peng | Rstorm : Resource-aware scheduling in storm[END_REF] ont pour objectif de concentrer toutes les tâches sur un sous-ensemble minimal d'unités de traitement. Pour chacune de ces stratégies, des contraintes sur les ressources processeur et mémoire nécessaires sont considérées. En fonction des opérateurs et de la structure du graphe d'opérateurs, l'adoption d'une stratégie peut amener à la formation de goulots d'étranglement réseaux ou à des surcharges fréquentes des unités de traitement actives. Dans les deux cas, cela entrainera des déplacements de tâches, réduisant la stabilité du système et augmentant la latence de traitement globale. Grâce à l'étude approfondie de diérents SGFD, nous avons identié que les choix de stratégies pour la parallélisation des opérateurs, l'équilibrage de charge intra-opérateur et l'allocation des tâches sont primordiaux pour maintenir un équilibre entre besoins des traitement et ressources allouées.

Bien que des solutions automatiques et quasi-optimales existent pour l'équilibrage de charge intra-opérateur et l'allocation des tâches, les stratégies de parallélisation des opérateurs présentent des limites importantes en terme de réactivité et d'automaticité.

4 autoscale : Une approche préventive pour le traitement élastique de ux de données

Dans le cadre de cette thèse, nous proposons une approche préventive d'auto-parallélisation des opérateurs, nommée autoscale, identiant les opérateurs présentant un risque de congestion dans un futur proche et analysant le contexte d'exécution pour estimer le degré de parallélisme adapté.

Pour ce faire, autoscale repose sur les deux grandes étapes suivantes : i) l'estimation de l'activité de chaque opérateur dans un futur proche grâce à une métrique d'activité ii) une analyse à l'échelle du graphe d'opérateurs permettant d'identier le sous-ensemble cohérent d'opérateurs à recongurer pour éviter l'apparition d'opérateurs congestionnées.

Estimation de l'activité des opérateurs

An d'estimer les risques de congestion dans un futur proche, nous observons chaque opérateur individuellement sur une fenêtre d'observation. À intervalle de temps régulier, le nombre de n-uplets en entrée est mesuré. À partir de toutes les mesures eectuées durant la fenêtre d'observation, nous estimons, par régression linéaire, le volume global de n-uplets que l'opérateur aura à traiter durant la prochaine itération de la fenêtre d'observation. Dans le même temps, nous mesurons la latence moyenne par n-uplet de l'opérateur an de dériver le nombre théorique de n-uplets qu'il pourra traiter durant la prochaine itération de la fenêtre d'observation en tenant compte également du nombre de tâches traitant les données en parallèle. En comparant ces deux volumes, nous pouvons alors estimer si l'opérateur est capable, avec son degré de parallélisme courant, de traiter le volume estimé de n-uplets dans un futur proche. Pour chacune de ces topologies, nous pouvons appliquer 2 types de ux :

un ux ayant une évolution progressive du débit en entrée et un ux ayant des augmentations et des diminutions soudaines du débit en entrée.

Nous dénissons également deux congurations pour chaque topologie :

une conguration minimale dénissant un degré de parallélisme pour chaque opérateur adapté à des débits faibles en entrée et une conguration experte dénissant des degrés de parallélisme adaptés au débits maximaux des ux.

Sur l'ensemble des évaluations, nous mettons en évidence l'intérêt d'autoscale pour éviter la congestion d'opérateurs en conguration minimale au démarrage. En eet, lors de l'augmentation du débit en entrée, autoscale augmente le degré de parallélisme des opérateurs à forte latence de traitement an de maintenir un débit de traitement supérieur au débit d'entrée. Dans le cas de Storm, la congestion entraîne un arrêt complet des traitement et des pertes irréversibles de données. Lorsque la conguration experte est choisie, autoscale parvient à maintenir une latence de traitement globale susamment faible pour éviter l'apparition de congestions tout en réduisant lorsque cela est possible les degrés de parallélisme des opérateurs. Au plus, autoscale parvient à réduire la consommation de ressources de 37% pour des performances équivalentes.

5 Vers un modèle d'auto-parallélisation prenant en compte le contexte d'exécution autoscale propose une approche d'auto-parallélisation entièrement guidée par les données. En eet, l'adéquation entre ressources nécessaires pour les traitements et ressources disponibles avec le degré de parallélisme courant est estimée sous réserve que les deux conditions suivantes soient remplies : Chaque tâche dispose d'autant de ressources que nécessaire sur l'unité de traitement sur laquelle elle est assignée.

La charge de traitement est équitablement répartie entre les tâches associée à un même opérateur indépendamment des propriétés du ux.

La première condition peut ne pas être vériée lorsque, sur une unité de traitement donnée, la somme des ressources requises par les tâches dépasse les ressources disponibles. Dans ce cas, une concurrence pour l'utilisation du processeur et de la mémoire apparait, causant potentiellement un décalage entre débit de traitement estimé et débit de traitement réel.

Concernant la seconde condition, il est important de noter que certaines applications sont composées d'opérateurs naturellement sensibles aux valeurs passées en entrée. Par exemple, une jointure génère un volume de données en sortie fonction du volume de données en entrée et de la répartition des valeurs dans l'ensemble des données en entrée. Si le résultat de cette jointure est calculé en parallèle grâce à un partitionnement des entrées, il n'est pas garantie que chaque partition nécessite un temps de traitement équivalent aux autres du fait de distributions de valeurs potentiellement diérentes.

Il apparait alors nécessaire de prendre en considération les ressources réellement utilisables pour chaque tâche an d'estimer un débit de traitement plus proche de la valeur réelle. De plus, la combinaison d'autoscale avec une stratégie d'équilibrage de charge adaptée est nécessaire an de maintenir une charge équivalente entre les tâches associées à même un opérateur.

Prise en compte des ressources pour l'évaluation de l'activité

An d'estimer la capacité de chaque opérateur tout en prenant en compte les ressources réellement utilisables, nous nous focalisons sur le temps processeur utilisable pour chaque tâche. À la diérence de certains systèmes [START_REF] Peng | Rstorm : Resource-aware scheduling in storm[END_REF] considérant uniquement des réservations de temps processeur statiques, nous suggérons de mesurer régulièrement, l'usage réel du processeur par chaque tâche. Ainsi, pour un opérateur donné, nous pouvons estimer le temps processeur moyen réellement utilisable par chacune de ses tâches et donc, déduire la capacité globale de l'opérateur avec une précision accrue par rapport à l'approche autoscale.

Évaluation expérimentale d'autoscale+

Lors de l'évaluation d'autoscale+, nous avons choisi de nous concentrer sur l'apport d'une approche préventive par rapport à des approches courantes de la littérature. Nous avons ainsi implémenté deux stratégies d'autoparallélisation : une approche par exploration incrémentale des degrés de parallélisme et une approche par apprentissage avec renforcement.

Nous avons modié le micro-benchmark en considérant une topologie linéaire possédant un opérateur à forte latence et insensible aux valeurs en entrée, une topologie linéaire équivalente mais avec un opérateur à latence sensible aux valeurs en entrée et enn une topologie complexe possédant un opérateur sensible aux valeurs en entrées.

Nous avons considéré cette fois-ci 4 ux :

2 ux ayant respectivement des variations progressives et soudaines du débit et une distribution uniforme des valeurs.

2 ux ayant les mêmes variations en terme de débit mais ayant une distribution des valeurs biaisée selon une loi de distribution prédénie.

Étant donné que nous comparons uniquement des stratégies de parallélisation automatique des opérateurs, toutes les topologies sont initialisées avec une conguration minimale. Nous observons qu'autoscale+ adapte au moins aussi bien le degré de parallélisme d'une topologie qu'une straté-gie basée sur de l'apprentissage par renforcement déjà entraînée. De plus, autoscale+ parvient à réduire la latence de traitement en ne réduisant le degré de parallélisme que lorsque cela présente un intérêt signicatif en terme d'économie de ressources.

6 DABS : Association d'autoscale+ avec une stratégie d'équilibrage de charge

Principe

Pour pallier le déséquilibre de charge entre les tâches associées à un même opérateur, plusieurs stratégies existent. Certaines stratégies établissent des partitions en fonction de la valeur de chaque n-uplet présent dans le ux.

D'autres stratégies tendent à répartir équitablement le nombre de n-uplets dans chaque partition indépendamment des valeurs des n-uplets. Dans les deux cas, la charge entre les tâches n'est équilibrée que sous l'hypothèse que le ux d'entrée a une distribution des valeurs ne variant pas signicativement tout au long des traitements. Nous suggérons alors de combiner autoscale+ avec une stratégie d'équilibrage de charge prenant en compte l'évolution de la distribution des valeurs dans le ux. Cette stratégie, baptisée osg, est basée sur une évaluation de temps de traitement de chaque valeur présente dans le ux d'entrée. Grâce à une structure de données compressée, osg peut évaluer la charge de chaque tâche en terme de temps de traitement, an d'équilibrer les les d'attentes de chaque tâche.

Évaluation expérimentale de dabs

An d'évaluer la performance de l'approche dabs, nous avons étudié la performance combinée de stratégies d'auto-parallélisation avec diérentes solutions pour l'équilibrage de charge entre tâches d'un même opérateur. En eet, certains opérateurs étant naturellement sensible aux valeurs en entrée, en terme de latence de traitement, le choix de la méthode d'équilibrage de charge peut avoir un impact majeur sur les performances.

Nous avons testé dabs face à autoscale+ sur une topologie complexe face à des ux ayant un biais important dans la distribution des valeurs.

Il apparait que dabs permet de réduire considérablement le déséquilibre de charge entre les tâches d'un même opérateur. Cela a pour conséquence de réduire le taux de remplissage des les d'attente jusqu'à 11% sur un même ux et donc de réduire la latence de traitement globale. De plus, dabs peut délivrer de meilleures performances qu'autoscale+ avec un degré de parallélisme moins élevé. En eet, grâce à une meilleure répartition de la charge, le débit de traitement global d'un opérateur est plus important pour un degré de parallélisme donné. During the last decade, stream processing has become a very active research domain as presented in [START_REF] Babcock | Models and issues in data stream systems[END_REF], Cherniack et al., 2003a, Ishii and Suzumura, 2011, Hummer et al., 2013, Lohrmann et al., 2015, Hochreiner et al., 2015]. These researches are motivated by the growing number of domains using stream-based applications. For instance, in the e-marketing domain, advertising applications analyze in real time logs of users browsing online in order to suggest relevant advertisements. Suggesting promotional oers in real time is more ecient than delayed oers because it relies on current interests and activity of users. In digital entertainment, online servers collects massive amount of data describing actions performed by players through sensors like gamepads or keyboards. It allows game providers to synchronize multiplayer sessions with low latency, which is necessary to oer a satisfying gaming experience. Another use case is geolocalization. Stream-based applications take in input the localization of users at high frequency and return their positions on a synthetic map. In addition, some sponsorized places (restaurants, shops, etc.) are also suggested to users. To guide users and promote some places, it is crucial that applications approximate positions of users and cross them with geographic data as soon as new GPS signals are received. This proliferation and diversication of sources emitting data in real-time has led to the development of specic techniques for data stream management [START_REF] Heinze | Autoscaling techniques for elastic data stream processing[END_REF]. In opposition to persistent data stored on disk, data streams are potentially innite sequences of transient data emitted by sources as soon as they are produced.

Users can process these data streams by submitting specic queries, denoted continuous queries, as introduced in [START_REF] Babcock | Models and issues in data stream systems[END_REF], Arasu et al., 2006, Arasu et al., 2004]. Contrary to oneshot queries on persistent data which are submitted each time a new result is expected, continuous queries are submitted once and potentially never terminate. A continuous query generates updates of the result as soon as new data arrive through input streams.

So processing data streams continuously requires to deal with Big Data issues linked to velocity and volume [START_REF] Babcock | Models and issues in data stream systems[END_REF], Cherniack et al., 2003a, Abadi et al., 2005]. Indeed, stream-based applications require that new data are processed on the y to return results before they become obsolete for end users. Moreover, most stream-based applications requires to keep the recent history of data streams in memory to perform aggregate or join operations. Nevertheless, the volume of data sharing a common timestamp may signicantly vary and critically increase the resources needed to store and process the recent history.

Initially, some centralized solutions [START_REF] Schreier | Alert: An architecture for transforming a passive dbms into an active dbms[END_REF] have been suggested to compute continuous queries over database management systems (DBMS). These solutions store both data and queries to compute updates of results at regular interval. Continuous queries are turned into equivalent direct acyclic graphs (DAGs) of operators, denoted workows. An operator is an atomic sequence of instructions applying a predened, eventually user-dened, function on inputs. So, operators are applied sequentially on new data to generate result updates at regular interval. Nevertheless, it presents several limitations in terms of velocity and volume. First, as data arrive continuously, it requires frequent disk accesses which involve important latency to store and retrieve data. While the frequency of data arrival exceeds the frequency of disk access, it causes an accumulation of data in main memory until overow. Likewise, when a continuous query contains one or many operators with high processing latency (e.g., cartesian product) applied sequentially, the data input rate may exceed the throughput of the continuous query. Then, as data streams are potentially innite they cannot be stored fully in memory or disk, so relying on nite memory and disk spaces require to manage data lifecycle to discard obsolete data. It involves additional management overheads having an impact on the performance of the system.

Then, some centralized solutions [Arasu et al., 2004, Abadi et al., 2003, Chandrasekaran et al., 2003] exploiting the parallelism of a multicore architecture have been suggested. They are intrinsically dierent from DBMS derived solutions because they keep data on main memory for faster accesses and they push new data in a pipeline of operators running in parallel. Moreover, an operator with high latency may be split in equivalent tasks running in parallel. It assumes that it is feasible according to operator properties (e.g., algebraic properties). These architectural modications signicantly improve the scalability of systems and dene a new class of systems called data streams management systems (DSMS). Nevertheless, at the era of Big Data, volumes of data transferred in real time exceed signicantly the resources (CPU, memory and bandwidth) of a single multicore machine. Even if operators can be replicated to scale treatments, the number of threads a machine can handle is limited and it prevents computing complex queries over multiple input streams.

To tackle these issues, some industrial [Peng et al., 2015, Akidau et al., 2013, Gedik et al., 2008, Biem et al., 2010], academic [Abadi et al., 2005, Chandrasekaran et al., 2003, Gulisano et al., 2012, Balazinska et al., 2004], and open-source [Peng et al., 2015, Noghabi et al., 2017, Carbone et al., 2015] distributed DSMS have been developed. They aim at exploiting the inherent of a distributed infrastructure without requiring specic knowledge to end users. They allow users to dene continuous queries over sets of data streams through stream-oriented languages [Arasu et al., 2006, Jain et al., 2008] or API [Peng et al., 2015, Akidau et al., 2013, Yang et al., 2012[START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF] in a programmation language like Java, C or Python. Distributed DSMS use parallelization frameworks as MapReduce [START_REF] Dean | Mapreduce: Simplied data processing on large clusters[END_REF] to improve the scalability of streambased applications. So, contrary to centralized DSMS [Arasu et al., 2004, Abadi et al., 2003, Backman et al., 2012], distributed DSMS are not bounded in terms of resource by the features 2. Problem Statement of a single machine but can exploit an extensible cluster. It allows users to process larger volumes of data with low latency. To distribute continuous queries over a cluster of machines, operators are potentially replicated into many tasks. Then, all tasks are assigned for a parallel execution.

In the context of the ANR project Socioplug1 (ANR-13-INFR-0003), we consider a sets of users, denoted communities, interested by results of some continuous queries. Each user has at its disposal a processing unit with limited resources in terms of CPU and memory. The objective of the project Socioplug is to provide a platform allowing each user to compute results over large volumes of data arriving over time. Users should be able to submit multiple continuous queries simultaneously and so, belong to many communities. The problem is that users have their disposal limited resources but also want to get results with low latency independently of stream rates and the number of continuous queries running in parallel. So, an architectural adaptation of data management systems is required to satisfy users requirements. It appears that the exploitation of a distributed infrastructure is the key feature to suggest a reliable stream processing platform as expected in the project Socioplug. However, one objective of this project is to rely on resources owned by users instead of processing continuous queries on a third cluster. To answer this problem, processing units of users belonging to a same community are interconnnected to build a cluster. This way, each community has at its disposal a cluster to compute once results of the continuous query.

As mentioned above, each user may belong to many communities. Thus, for the provider managing the cluster, the general problem is to balance resource usage of each processing unit between all continuous queries it should process. In consequence, it becomes crucial for the provider to t resource usage of continuous queries to their respective processing requirements. In a stream processing context, it is a major challenge as processing requirements vary as stream rates uctuate over time. The dynamic adaptation of workows representing continuous queries, called elastic stream processing [START_REF] Ishii | Elastic stream computing with clouds[END_REF], Hochreiner et al., 2015[START_REF] Heinze | Autoscaling techniques for elastic data stream processing[END_REF],

is then necessary to maintain a balance between processing requirements and resource usage at runtime.

Problem Statement

On one side, the elastic treatment of continuous queries satisfy user requirements independently of variations happening in the execution environment. On the other side, it allows DSMS to take advantage dynamically of available resources without needing an oversized cluster. Nevertheless, reconguring workows representing queries implies important reconguration overheads. Indeed, as operators process continuously transient data stored in main memory, modifying the conguration of some operators requires to pause treatments, apply some transformations on a workow and spread them over the cluster. Moreover, it may involve code and data transfers through network. All those operations are time and resource consuming so they degrade momentarily the performance of the DSMS.

To take fully advantage of elastic stream processing, a DSMS must identify when a reconguration is needed or recommended to improve processing latency and result quality of continuous queries. To ensure that a reconguration ts eectively processing requirements to resource usage, it is crucial that the type of reconguration corresponds to the problem caused by a variation in the execution environment. Three types of reconguration are usually considered to adapt the conguration of a DSMS:

• Operator scheduling aims at assigning operators such as the global scheduling plan satisfy some conditions. It can be achieved by spreading evenly the processing load over all available machines [START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF] or minimize network trac to avoid network bottlenecks [Xu et al., 2014]. So, operator scheduling can prevent congestion due to massive network transmissions but it cannot increase the global processing rate of an operator.

If computations are bounded by the time complexity of the operator, reassigning it on another machine is inecient.

• Load balancing should serve as guarantee that the load is evenly distributed over tasks associated to a given operator [START_REF] Rivetti | Ecient key grouping for near-optimal load balancing in stream processing systems[END_REF]. If it is not the case, it signicantly reduces benets brought by parallel execution. To improve the global processing rate, a load balancing strategy assumes there is a natural imbalance in data that a specic routing plan may compensate.

• Parallelization of operators [START_REF] Lohrmann | Elastic stream processing with latency guarantees[END_REF], Schneider et al., 2009, Gedik et al., 2014, Shukla and Simmhan, 2017] consists in increasing (scale-out) or decreasing (scale-in) the number of tasks, also denoted parallelism degree, of an operator. It has a direct impact on the processing rate of an operator and bounded theoretically the input rate an operator is able to handle without going to congestion. So, adapting the number of tasks associated to an operator denes mainly the input rate an operator can process.

Most distributed DSMS integrate parallelization mechanism, scheduling and load balancing strategies. All reconguration types can be triggered oine [Aniello et al., 2013[START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF], Neumeyer et al., 2010] to dene a static conguration which is not modied unless they are triggered manually. Some DSMS [Xu et al., 2014, Gedik et al., 2014, Lei and Rundensteiner, 2014] can also adapt execution of workows at runtime according to a continuous monitoring of the execution environment. Nevertheless, to our knowledge, most elastic solutions focus on one reconguration type and neglect the synergy between the dierent mechanisms adapting the execution of workows at runtime.

From these observations, we look after a solution which integrates both dynamic mechanisms for parallelization, scheduling and load balancing of operators in a distributed stream processing context. Many DSMS integrate near optimal scheduling [Peng et al., 2015, Aniello et al., 2013] and load balancing strategies [Rivetti et al., 2015, Lei andRundensteiner, 2014] but the automatic parallelization, or auto-parallelization, of operators raise several challenges: (i) when a modication of parallelism degree should be triggered? (ii) which metrics should be considered to adjust the parallelism degree? (iii) how to avoid massive reconguration overheads system but prevent congestion of operators? (iv) how to avoid recongurations which do not adapt eectively resource usage? Thus, we focus on auto-parallelization of operators to perform elastic stream processing. The rst challenge (i) is linked to the reactivity of the system. Actually, as presented above, the global rate in input of an operator may vary at any time. So, if the system reacts with an important delay to a signicant increase in input rate, it may only correct an eective congestion and let result quality degrades for a certain period of time. Nevertheless, if the system performs a scaleout or a scale-in each time the input varies signicantly, it may recongure itself continuously, bringing important overheads and causing a major degradation of performance and result quality.

The second aspect (ii) deals with the accuracy of each reconguration. While considering recent uctuations in input rate, which value or aggregated value should be considered as the expected input rate? Several approaches have been proposed to forecast values in near future under some assumptions on the evolution of stream rate.

Then, the stability of the system (iii) can be evaluated at the scope of each task (i.e., controlling the frequency of reconguration) but it can also be evaluated at the scope of the entire continuous query. Indeed, limiting the occurrence of scale-in/out for each operator does not guarantee that the system does not recongure operators belonging to the same continuous query with a domino eect. For instance, if a workow performing a join between two input streams receives larger volumes in input, it is benecial for the system to recongure simultaneously all downstream operators as the system can anticipate that the join operator will generate large volumes of outputs.

Finally, in a distributed multicore context, some aspects like concurrency for resource usage may create a gap between expected and eective resource usage(iv). Catching such restrictions imposed by the execution support is necessary to adapt accurately resource usage for each continuous query.

Contributions and Organization

In a rst time, we analyze which aspects have an impact on distributed stream processing. We suggest an abstract architecture for elastic stream processing. This architecture highlight the dierent levels of query execution from logical layer to physical layer. For each layer, we detail its role, how a reconguration needed is detected and which impact it has on global execution of continuous queries. We also give an overview of common adaptation strategies used at each level of query execution. In addition, we expose dependencies between adaptations and discuss about the compatibility of dierent adaptation strategies.

From this generic architecture, we highlight lacks in elasticity while facing critical uctuations in input rate and distribution in terms of reactivity and automaticity. To tackle this issue, we suggest an original auto-parallelization strategy which allow to perform preventive elastic stream processing. Indeed, our auto-parallelization analyzes both stream and operator properties like the selectivity factor to estimate accurately processing requirements at runtime. Compared to other solutions suggested in the literature, our solution requires neither a learning phase nor user expertise or intervention to size available resources to treatments.

The contributions presented in this manuscript are the following:

• A survey on congestion management presenting common techniques used to enable elastic stream processing over a distributed infrastructure. In addition, we suggest a classication of detection methods for congestion management extending the classication of auto-scaling methods presented in [START_REF] Lorido-Botran | A review of auto-scaling techniques for elastic applications in cloud environments[END_REF].

• The review of a selection of DSMS covering the variety of solutions existing in the literature.

For each DSMS, we present its specic features in terms of query denition and congestion management.

• The abstract architecture for elastic stream processing, called ORACL loop, which identify the dierent steps of dynamic reconguration for congestion management.

• The original auto-parallelization autoscale which relies on a monitoring module observes the recent history of operator activity at regular interval of time. Through this module, autoscale can analyze the behavior of the operator at local scope and compute an activity metric. This metric estimates the gap between input and processing rates at operator scope and in near future. It allows to adjust the parallelism degree of operators in a proactive way and prevent congestion.

• An algorithm checking the consistency at workow scope of recongurations detected at operator scope and integrated to autoscale. This algorithm takes as input local reconguration requirements and the structure of the workow and computes the set of recongurations to trigger simultaneously in order to improve the performance and the stability of system. It aims at avoiding recongurations with domino eect and antagonist modications of parallelism degrees. We highlight the eectiveness of this algorithm through an experimental study against dierent strategies.

Then, we extend this auto-parallelization strategy to the approach autoscale+ with the support of concurrency and load imbalance between tasks. To do so, the following improvements have been integrated:

• The consideration of eective resource usage for each task of operator. We focus on CPU usage as it is the most crucial resource for computation in a stream processing context.

• A resource-aware variation of the activity metric which takes eective resource usage into account to dene the gap between input and processing rate.

We integrate autoscale+ within a solution, named dabs, which adapts resource usage to processing needs. The solution has the following properties:

• The auto-parallelization strategy autoscale+ associated to a complementary load balancing strategy. The association of compatible approaches for auto-parallelization of operators and load balancing ensures that each operator has the capacity to process input streams independently of data volume and distribution.

• Congurable parameters allowing users to adapt treatments to their needs.

We implemented these auto-parallelization strategies over the DSMS Apache Storm2 and suggest a comparative evaluation with the native behavior of this solution.

In this research, we focus our eorts on preventive auto-parallelization of operators with consideration for the trigger of scale-in/out, the consistency of reconguration and the stability of the system. All aspects linked to data partitioning and state management [START_REF] Shukla | Toward reliable and rapid elasticity for streaming dataows on clouds[END_REF], Castro Fernandez et al., 2013, Gedik, 2014, Wu and Tan, 2015, Ding et al., 2015, Nasir, 2016, Cardellini et al., 2016] are out of the scope of this research.

Organization of the manuscript

In chapter 2, we introduce the background about data stream representation and stream processing. In chapter 3, we suggest a survey of elastic stream processing oriented on congestion management. It is composed of a catalog of techniques for congestion detection and management and also a classication of DSMS illustrated by representative academic, industrial and open-source solutions. The generic architecture ORACL for elastic stream processing is detailed in chapter 4. Then, in chapter 5, we present the auto-parallelization strategy autoscale and explain how it adapts dynamically and automatically parallelism degrees of operators. We expose an experimental study of optimizations integrated in the algorithm. The extension, called autoscale+, is detailed in chapter 6 and we test its compatibility with a load balancing strategy of the literature. Experiments on microbenchmark and complex topologies are commented in chapter 7. Finally, we summarize conclusions of our works and present future works in chapter 8. In this chapter, we briey remind basic concepts about data streams and data stream processing. We present the execution context that we consider in the remainder of this work. In addition, we give an overview of main methods used to detect congestion in a stream processing context. Finally, we expose issues linked to operator congestion and concepts related to system elasticity.

1 Data Streams

Modeling and features

Intuitively, data streams are sequences of stream element sets [Arasu et al., 2006, Tucker et al., 2003] arriving continuously over time. More formally, a stream is dened as follow: Denition 1. (Stream) [Arasu et al., 2006] Let consider a stream S described by a schema and an ordered timestamp set τ . A stream is a potentially innite multiset of elements < s i , τ i > where s i is a tuple of the stream respecting the schema of S and τ i ∈ τ the associated timestamp.

A timestamp τ i belongs to a timespace τ which denes the chronological order over stream elements. Denition 2. (Timespace) [START_REF] Petit | An algebric window model for data stream management[END_REF] A timespace τ is an isomorphism of R naturally and totally ordered. Any timestamp τ i belongs to τ .

It is worth noting that a timestamp τ i may be assigned explicitly by the source which emit the stream element or implicitly by the processing system upon arrival according to wall-clock time. For example, let consider a stream consumed by a heatwave monitoring application. Stream elements belonging to this stream are described by attributes corresponding to climatic properties like temperature, humidity and CO 2 concentration. According to such schema, a stream element can be < [23, 60.0, 320], 1508154964> where the rst value is the temperature in celsius degrees, the second the humidity in percentage and the third the CO 2 concentration in parts-per million. The timestamp is the time elapsed since an initial timestamp (e.g., January, 1st 1970) in seconds.

Data streams are not only collections of timestamped data, they are emitted as soon as they are produced and this ephemeral nature raises several issues while processing streams:

• Most data stream sources cannot delay emission of stream elements or store the recent history of emitted elements so any loss of stream elements due to computation or network failure is denitive. So, processing systems must remain available to receive new data at anytime.

• The number of stream elements arriving at each timestamp is nite but potentially huge implying that processing systems must be able to receive large stream element sets.

• Data streams may be described by simple attributes (e.g., numerical attributes for meteorological sensors) but also complex attributes (e.g., large matrices for high denition video streams). The variety of data a system is able to handle denes the perimeter of applications it is able to support.

Classication of data streams

After this formalization of data streams and the presentation of their inherent properties, we suggest a classication of data streams and illustrate each category with some typical examples. It highlights specic features of streams to take into account while processing each category of streams.

Criteria of classication

As illustrated on Figure 1, streams can be described according to the evolution of two properties over time: rate and distribution of values.

Stream rate

Stream rate refers to the quantitative aspect of streams where we only consider the number of stream elements arriving at each timestamp. Considering an innite stream, the distribution of values denes the relative occurrence of each possible value.

Data Streams

We distinguish four types of streams: steady, bounded, per-pattern and erratic streams. Steady streams are emitted by sources sending stream elements of predened size and at regular time interval. Typically, it corresponds to measurements generated by sensors or monitoring systems. There are several application domains manipulating such streams: in a context of home automation, sensors send at congurable time intervals measurements about the temperature, the humidity and the luminosity. In military logistic context3 , sensors send latitude, longitude and speed of vehicles for GPS tracking and coordination. We also nd steady streams in stock analysis context like Nasdaq4 updating values of stock prices with a frequency up to the minute.

Bounded streams are variable but have absolute bounds in input rate as illustrated on Figure 2. For example, let consider a xed number of motion sensors connected to a single wireless receptor and generating a stream element each time they detect a movement. This receptor sends the union of all sensor streams to an application analyzing this merged stream. Strictly in term of input rate, the maximal workload the application has to process corresponds to all sensors sending a stream element at the same timestamp. As the number of sensors is xed, the input rate cannot be greater than this case. At the opposite, if no sensor detects a movement the stream is empty.

Per-pattern streams uctuate according to patterns more or less long over time. According to the complete history of stream variations on a signicant time period, uctuations in near future are predictable with high probability. In a real-world context, many domains use per-pattern streams. For example, streams generated by road trac monitoring uctuate periodically in terms of volume at dierent granularities. At day granularity, the volume is important around working hours and low during them. At year granularity, the volume increases signicantly around summer holidays. Another example of per-pattern streams are click logs generated by websites especially online stores. Actually, at year granularity, users visit massively such websites around holiday season compared to the rest of year.

As shown on Figure 2, rate of erratic streams at a given timestamp are independent of previous variations making those streams unpredictable. An example of erratic stream is online auctions supported by platform such as eBay where new items can be added for auction at anytime and bids arrive depending on the popularity of items, their prices and several other parameters.

As mentioned above, for all these types of stream, the distribution of values may change over time too but contrary to stream rate, the impact of this property on computations depends completely of the consuming application. Streams can be grouped in two categories independently of the four types presented above: even and uneven streams.

If an application has a xed overall latency no matter which value is passed in input, the stream is said even. According to a given application, the distribution of values of an even stream could be ignored as it has no impact on computation. A stream may be even for several reasons:

• Some values with similar impacts on processing latency are extremely frequent (i.e., the sum of their relative occurrence is close to 1).

• All computations performed within a given application are independent of values in input (e.g, projection of attributes).

• Computations which have a time complexity depending on the value in input have a marginal impact on the overall latency of the considered application.

Contrary to even streams, variations in data distribution of uneven streams may have a signicant impact on processing latency. It is due to an application containing treatments with time complexity depending of input values and consuming streams with varying distribution of values over time. It may cause important variations of processing latency for a given application and with xed resources.

To sum up, streams are intrinsically dierent from static data. They are potentially innite and arrive continuously over time. Moreover, they may or may not signicantly uctuate in terms of volume and value distribution over time as shown on Figure 2. Nevertheless, if uctuations in stream rate have a direct impact on computations, uctuations in distribution of values have only an impact under assumptions mentioned above. In all cases, processing streams raises several issues due to their ephemeral nature and their variety in a Big data context.

Data Stream Processing

To tackle issues linked to velocity and volume of data streams, some solutions based on traditional Database Management Systems (DBMS) [START_REF] Schreier | Alert: An architecture for transforming a passive dbms into an active dbms[END_REF], Rosenthal et al., 1989] have been suggested but they show some serious limitations [Abadi et al., 2003, Stonebraker et al., 2005] essentially due to data availability. Indeed, processing data streams involves two main features: on-the-y computation of ephemeral data and an active processing model to submit query.

So, it appears that DBMS cannot satisfy high velocity requirement [START_REF] Stonebraker | The 8 requirements of real-time stream processing[END_REF] because of the important latency due to I/O accesses on disk. Thus, reversing active and passive protagonists is necessary to compute data on-the-y. This processing paradigm allow to keep ephemeral data in movement without requiring a costly storage. Moreover, it favors availability and velocity of systems [START_REF] Stonebraker | The 8 requirements of real-time stream processing[END_REF] as data remain in memory for all computations.

DBMS suggest a human active database passive (HADP) paradigm [START_REF] Schreier | Alert: An architecture for transforming a passive dbms into an active dbms[END_REF], Abadi et al., 2003] for data processing. Each time a user wants an updated result, he has to submit a query over static data stored on disk. It means that every time there is an update on data, the user must submit again his query to update the result. While processing streams, a user wants to submit a query once and to receive result updates as soon as the system receives new stream elements. So, it corresponds to a paradigm database active human passive (DAHP).

To turn from HADP to DAHP paradigms, queries must support operators turning substreams into nite relations. That means dividing the potentially innite sequence of stream elements into nite subsequences which can be processed within a nite time.

Continuous queries

To enable DAHP paradigm, a new class of queries, called continuous queries [START_REF] Babu | Continuous queries over data streams[END_REF], appeared. Denition 3. (Continuous query) A continuous query is an endless and deterministic application taking as input a set of streams and generating one or many output streams. A result may be generated as soon as a new stream element arrives or at regular interval.

Contrary to queries on static data, continuous queries may never terminate as input streams are potentially innite. They can be applied on a single stream element (e.g., ltering or transformation) or on sets of stream elements (e.g., aggregation). Nevertheless to perform an aggregation, it requires theoretically to store the entire stream on memory but it is impossible as streams may be innite. So a common trade-o between completeness and storage space consists in considering only the recent history of data streams to perform aggregation.

Computation windows

As data are produced and arrive in real time, recent data have a greater impact in most application domains. For instance, geolocalization applications rely on GPS signals received within the last seconds to approximate the position and the direction of the end user. Likewise, most stream-based applications generate results which loose quickly interest for end users over time. Thus, using recent history is a common approximation while processing data streams. It ensures that results are relevant for users and reduce the volume of data to consider at each computation.

Window semantic

Recent history, or window [START_REF] Babcock | Models and issues in data stream systems[END_REF], denes the nite portion of the stream to consider. Denition 4. (Computation Window) A computation window is a logic stream discretization [START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF] which is dened by a size and a slide (see Figure 3). Considering the front of a window and the chronological order, the size denes the subset of elements to consider. The slide denes the step between two consecutive window fronts.

When the slide is smaller than the size, the window is called a sliding window [Arasu et al., 2006, Arasu et al., 2004, Golab et al., 2004]. Sliding windows are characterized by an overlap between one or many consecutive iterations. Several works [START_REF] Golab | On Indexing Sliding Windows over Online Data Streams[END_REF], Kang et al., 2003, Qiao et al., 2003] exploit this property to optimize query execution over data streams. When the slide equals the size, the window is called tumbling. In this case, the intersection between two consecutive iterations is empty.

A window is an operator turning a nite subset of stream elements into an instantaneous, or temporal [START_REF] Petit | An algebric window model for data stream management[END_REF], relation R(τ i ) [Arasu et al., 2006]. As discussed in section 1.1 and considering stream elements timestamped explicitly or implictly, the problem is to dene which stream elements are relevant for computation. Indeed, there are two ways to dene window size and step: time-based and count-based windows [START_REF] Golab | On Indexing Sliding Windows over Online Data Streams[END_REF], Kang et al., 2003, Qiao et al., 2003]. Considering the current timestamp τ now , a time-based window [Arasu et al., 2006, Golab et al., 2004] denes as relevant all stream elements associated to a timestamp τ i with τ i ∈ ]τ now -∆; τ now ].

The duration ∆ is the size of the window. Formally, it produces an instantaneous relation R(τ now ) from a stream S dened in formula [START_REF] Arasu | The cql continuous query language : Semantic foundations and query execution[END_REF] (see [Arasu et al., 2006]):

R(τ now ) = {s| s, τ i ∈ S ∧ (τ i ≤ τ now ) ∧ (τ i ≥ max{τ now -∆, 0}} (1)
From this denition, it appears that a time-based window contains a nite but potentially variable number of stream elements.

Count-based windows

A count-based window, or tuple-based window [Arasu et al., 2006, Arasu et al., 2004], is dened by a size N dening how many stream elements are relevant for computation. The slide denes the number of new elements to receive before computing a new result on the updated window content. As zero, one or many stream elements may be associated to a timestamp, there is no guarantee about acquisition time. Contrary to time-based, count-based windows are not deterministic. Indeed, stream elements associated to a same timestamp cannot be ordered explicitly so considering last N stream elements may not produce a similar result depending on stream sources.

A variation of count-based windows are partition-based windows [Arasu et al., 2006]. A partition-based window is dened by a size N and a nite set of attribute/value pairs dening each partition. A partition-based window is complete when there are N stream elements satisfying each partition.

Window implementation

There are two main mechanisms to implement windows:

• Punctuations [START_REF] Tucker | Exploiting punctuation semantics in continuous data streams[END_REF] to consider stream as the union of nite substreams.

They are specic elements in a stream which dene the end of a substream. This mechanism is particularly useful because it allows elements to be processed on the y [START_REF] Backman | C-mr: Continuously executing mapreduce workows on multi-core processors[END_REF] 2. Data Stream Processing without meeting issues with blocking operators. Nevertheless, punctuations rely on a a priori knowledge on data. To insert punctuation correctly in a stream, it is required to know when the last element of the last timestamp belonging to a given window is arrived in the system.

• Buer-based stream discretization [START_REF] Gedik | Spade: The system s declarative stream processing engine[END_REF] turns a stream fragment into an inmemory relation. Buers gather stream elements according to window denition. For time-based windows, timestamps are used to determine when a buer is full and ready to be processed.

To sum up, the use of computation windows allows users to specify which data elements to consider. It enables in-memory processing through the reduction of a stream to an instantaneous relation. Depending on operator semantic, an instantaneous relation may be processed as a batch of static data or on-the-y as new stream elements arrive to the system. Indeed a distinction between stateless and stateful operators [START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF] is made.

Operator types

Stateless operators

Stateless operators, for example lters based on an attribute value, process data streams element by element. They return a new result with an unpredictable frequency an input may not generate an output depending on its value (e.g., if a ltering predicate is not statised). Moreover, these operators do not have information about previous and current computation windows when a data stream element is processed. Nevertheless, a stateless operator may used historic data stored on local memory or disk. It allows to compute joins with a static dataset within a stateless operator [Abadi et al., 2005, Abadi et al., 2003[START_REF] Cherniack | Scalable Distributed Stream Processing[END_REF], Balazinska et al., 2004].

Stateful operators

In opposition, stateful operators take as input a set of elements grouped on a window to compute a single result. When stateless operators can be applied on-the-y on each new stream element, stateful operators can only return a result after the completion of a window iteration. In addition, these operators keep information like the identier of the current window or any intermediate result. Information generated during a stateful operator runtime is denoted its state [START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF]. For example, a window-based stateful operator computes the sum of values associated to an attribute. Its state contains the identier of the current window, the attribute to group and the current sum value.

Thus, stateless and stateful operators are dened with dierent parameters. When stateless operators require only a function to apply on each input, a stateful operator relies on a computation window to complete its denition. This semantic gap has to be taken into account by the query language.

Query languages

To dene continuous queries, we distinguish three main categories of query languages: declarative, imperative and graphical languages.

Declarative denition

Declarative approaches [Arasu et al., 2006, Abadi et al., 2005, Chandrasekaran et al., 2003] suggest algebraic extensions of traditional declarative query languages like SQL to support stream management. As any declarative language, they allow users to use a predened set of operators to describe expected results. The system takes charge of turning the declarative expression into an ordered set of operators.

CQL [Arasu et al., 2006, Arasu et al., 2004] is declarative language for continuous query definition derived from SQL-99 and supporting both traditional relation and data streams. There are three classes of operators in CQL:

• Stream-To-Relation operators turn a data stream into an instantaneous relation. It corresponds to window declaration through a clause range taking as parameter window size.

It is worth noting that window slide is not explicitly dene in CQL queries and is considered as a global parameter of the system. CQL supports timed-based, count-based and partition-based windows.

• Relation-To-Relation operators take as input a set of instantaneous relation and produce an instantaneous relation. It correspond to standard SQL operators (Filter, Projection, Join...) and can be applied on instantaneous relations like SQL operators on static data.

• Relation-To-Stream operators produce a stream from a set of instantaneous relations. There are three Relation-To-Stream operators: IStream returns only updates between the current and the previous iteration of the window. DStream returns results produced during the previous iteration of the window and which not belong to results of the current iteration.

Finally, RStream returns all results.

There are several declarative languages to process data streams. In most cases, they include a subset of SQL relation-to-relation operators and provide specic stream-to-relation and relationto-stream operators.

SQLStream5 is an extension of SQL which support complex window denition through a clause window. Compared to CQL, SQLStream has a greater expressiveness as time-based and count-based windows are dened with explicit window size and slide. It allow to x the oset between the reception of the last stream element included in the iteration of the window and the arrival of updated results to end users.

SparkSQL 6 oers a support of common SQL operators over Apache Spark7 . This language restricts SQL expressiveness and is not especially dedicated to stream processing even if it supports the window clause for aggregation.

There are several declarative languages for stream processing built around SQL. If they induce low eorts for maintenance and reusability due to short denition of continuous queries, some applications are intrinsically dicult to turn into a SQL expression (e.g., matrix multiplication). To increase expressiveness of the denition language, some solutions [Peng et al., 2015[START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF], Neumeyer et al., 2010] support high-level imperative languages.

Imperative denition

Some solutions [Peng et al., 2015[START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF], Neumeyer et al., 2010] support continuous queries dened through an imperative high-level language like Java, Python or C++. From user side, query denition is signicantly dierent. Instead of dening the entire query through a single expression, users have to dene the query operator by operator through some operator patterns and the implementation of business logic. These patterns specify the form of input and, optionally, output data and manage data lifecycle automatically [Peng et al., 2015, Neumeyer et al., 2010]. Some patterns replace explicit declaration of stream-to-relation operators. Relation-to-relation operators are entirely part of business logic implemented by users and relation-to-stream operators are managed implicitly by patterns.

If imperative denition requires important development and maintenance eorts, it oers more expressiveness for applications managing complex data (e.g., video stream or motion sensors analysis). At the edge of these two categories of query languages, some approaches [Abadi et al., 2005, Abadi et al., 2003] suggest languages more user-friendly which require less development eort than imperative denition but more exibility than declarative languages.

Graphical denition

Box and arrow paradigm [Abadi et al., 2003[START_REF] Cherniack | Scalable Distributed Stream Processing[END_REF], Balazinska et al., 2004] represents an application as a direct acyclic graph (DAG) of boxes connected by arrows. A box corresponds to an operator, stateless or stateful, taking stream elements as input. Arrows indicate how stream elements are routed between boxes. Authors in [Abadi et al., 2003] suggest a query language named SQuAl composed of some standard SQL operators like Join and Filter enriched with stream-oriented operators like Tumble and WSort [Abadi et al., 2003]. Operators can be connected to dene the DAG through a graphical interface. SQuAl has a limited expressiveness compared to CQL because it oers only a subset of standard SQL operators. Moreover, streambased operators support only treatments over tumbling windows restricting possible applications. The main dierence between SQuAl and CQL is that continuous queries cannot benet from automatic query optimization based on algebraic properties. Performances of box and arrow DAGs depend more on user implementation than an equivalent CQL query. In [Pruett, 2007], authors present the interface Yahoo Pipes allowing users to dene graphically continuous queries over RSS streams. Yahoo Pipes suggests mainly ltering operators.

As traditional DBMS, a DSMS has to dene an execution plan for each submitted continuous query. It means turning a user denition into a set of connected operators. This operator set is the internal representation of any continuous query for DSMSs. Actually, if users dene continuous queries through an imperative language or a graphical interface, the DSMS can use directly queries as dened by users but if queries are dened through a declarative language, the DSMS takes charge of turning the declarative expression into a set of predened operators.

Execution plans and paradigms

As performed in DBMS, query plans are dened at two levels: logical and physical. Query plans at logical level corresponds to a direct acyclic graph (DAG) of logical operators (e.g., Select-Project-Join for SQL-derived languages). According to user denition, it organizes the sequence of operators to apply on operators. If the query is dened through a declarative language the denition of the query plan is based on algebraic properties to nd the sequence of operators having the lowest cost [Garcia-Molina, 2008] (e.g., transferring the lowest number of stream elements between operators). If the query has been dened through an imperative or a graphical language, the execution plan corresponds directly to user denition.

Then, logical operators are mapped to physical operators [Garcia-Molina, 2008, Graefe, 1993] to dene physical query plans. If a continuous query has been dened using a declarative language, predened implementations of operators are used such as users do not have to implement them. Otherwise, users must implement explicitly the logic of operators through a high level programming language.

So, each continuous query is equivalent to an execution plan within a DSMS. However, all query plans do not share a same structure. We distinguish two structures of query plans: workows and MapReduce jobs [START_REF] Dean | Mapreduce: Simplied data processing on large clusters[END_REF]. These structures impact applicable forms of parallelism between operators. Moreover, they aect data representation and lifecycle.

Workow paradigm

Workows are most generic structures that we will use by default to represent query plans. Denition 5. (Workow) A workow is a DAG where vertices are operators and edges dene data transmission between operators.

Let consider a workow W 1 composed of some stateless operators. As operators process each stream element as soon as it arrives in input, the workow exploits naturally stream pipelining [START_REF] Sattler | Towards elastic stream processing: Patterns and infrastructure[END_REF], also denoted task parallelism [START_REF] Hirzel | A catalog of stream processing optimizations[END_REF]. Denition 6. (Stream pipelining) Let W be a workow which can be divided into k consecutive operators. Each O i , i ∈ [1;k], is denoted the i-th stage of W and is executed on an exclusive process.

According to Denition 6, stream elements are routed through all stages sequentially. Of course, stream pipelining is limited if a stateful operator belongs to the workow. Indeed, as stateful operators group stream elements by windows before computing a result, they limit the benet of parallel treatments. However, stateful and stateless operators may benet from data parallelism [START_REF] Hirzel | A catalog of stream processing optimizations[END_REF], or stream partitioning, depending on their respective semantics. Denition 7. (Data parallelism) Let T 1 i , T 2 i ,..., T k i be k equivalent tasks. We suppose that all tasks T j i take as input, outputs produced by an operator O 0 . In order to process T 1 i , T 2 i ,..., T k i in parallel, O 0 can split its outputs in k partitions and distribute a partition to each tasks T j i .

According to Denition 7, data parallelism can be exploited eciently only under the assumption that tasks T 1 i , T 2 i ,..., T k i are semantically equivalent. Moreover, the global result of operators T 1 i , T 2 i ,..., T k i must be independent of the partitioning method.

MapReduce paradigm

MapReduce [START_REF] Dean | Mapreduce: Simplied data processing on large clusters[END_REF]] is a well-known framework developed initially to process huge amount of disk-based data on large clusters. The strength of this framework is to exploit eciently data parallelism (see Denition 7) with a simple programming paradigm. Actually, the core of any MapReduce application relies on two functions: Map and Reduce. These generic functions are dened as follow according to [START_REF] Dean | Mapreduce: Simplied data processing on large clusters[END_REF]:

• Map (k1, v1) -→ list(k2, v2) • Reduce (k2, list(v2)) -→ list(v3)
As mentioned above, MapReduce framework aims disk-based data processing. Contrary to DBMS, MapReduce-based systems do not rely on data model to optimize treatments. In order to distribute great amount of data on a large cluster, data are partitioned with regards to cluster conguration (e.g. number of nodes executing Map and Reduce functions). Each partition is identied with a key used to aect the partition to a Map node. The scheduling between partitions and Map nodes follows distribution strategies like Round-Robin in order to balance computation load.

Each Map node applies the user-dened Map function on one or many partitions. The function produces a list of intermediate key/value list pairs depending on partition contents. Then, outputs of Map nodes are shued and sorted in order to perform Reduce phase more easily. Map node. This phase consists in applying the Reduce function on Map outputs in order to have results for each partition. It may be useful while having potentially several redundant computation like presented in [START_REF] Backman | C-mr: Continuously executing mapreduce workows on multi-core processors[END_REF]. Each Reduce node gathers intermediate key/value list pairs and computes a list of value which are nal results.

To sum up, we have introduced a formal representation of data streams. They dier from static data in the way that they are potentially innite multiset of elements arriving to the system with uctuating rates. From this, traditional DBMSs cannot process these data streams with low latency and a nite amount of memory. To tackle this issue, some DSMSs have been developed and use computation windows to query data streams without restricting the expressiveness of query languages. Even if DSMSs allow processing streams without storing huge amount of data, they need computation resources to process data on the y. As data streams may have high input rates, it is necessary that DSMSs rely on scalable infrastructures. To do so, many DSMSs have been designed such as they can exploit eciently a distributed infrastructure.

Distributed Stream Processing

To meet processing requirements involved by stream processing, DSMSs must take advantage of a distributed infrastructure [START_REF] Stonebraker | The 8 requirements of real-time stream processing[END_REF] able to evolve as streams do. Some processing units can be added to a cluster statically or dynamically to extend the overall processing capacity. In order to facilitate the reading, we refer to execution plans (see section 2.1.4) as workows without distinction between workows or MapReduce jobs in the remainder of this chapter.

Execution context

We present here the execution context of distributed stream processing. We consider users interested by services generating streams. To query streams, users can submit continuous queries on a distributed infrastructure 8 .

Figure 5: Distributed stream processing Let us consider three continuous queries Q1, Q2 and Q3 represented respectively by workows W1, W2 and W3. Q1, Q2 and Q3 are submitted on the distributed infrastructure by some users as illustrated on Figure 5. Each continuous query takes as input one or many streams in the stream set {S 1 , S 2 , S 3 }. An input stream may have uctuations in input rate and value distribution as shown on the left of Figure 5. Workow W1 is linear, W2 is organized in diamond and W3 is a star workow. Most complex workows can be decomposed in such elementary patterns [Peng et al., 2015].

As illustrated on Figure 5, tasks T 2 1 and T 2 2 are devoted to operator O 2 so it means that O 2 has a parallelism degree of two. Tasks are assigned on processing units of machines M 1 to M 8 according to a scheduling plan. It corresponds to a mapping from tasks to processing units setting resources available for each operator composing a workow. On Figure 5, the four tasks of workow W1 are assigned on processing units of machines M 1 to M 4 .

We distinguish three states of machines: machines M 1 to M 4 are actives because some tasks are assigned on their processing units. Machines M 5 and M 6 are congured but inactive because there is no task assignment on their processing units. Finally, machines M 7 and M 8 are available but not congured so no task can be assigned on their processing units.

Result streams S' 1 , S' 2 and S' 3 are respectively associated to queries Q 1 , Q 2 and Q 3 . Each output may result from the treatment of a single stream element or an element set if workows W1, W2 or W3 include some aggregative operators.

Each task T j i applies the operator O i and receives stream elements on an input queue, also called pending queue, which has a xed and nite size.

With each continuous query, users provide a policy, denoted quality-of-service (QoS), specifying expected performance and result quality. Commonly, performance corresponds to a constraint on overall latency. From this specication, latency constraints can be inferred for each operator [Abadi et al., 2005[START_REF] Cherniack | Scalable Distributed Stream Processing[END_REF] or each processing unit [START_REF] Balazinska | Load management and high availability in the medusa distributed stream processing engine[END_REF].

Result quality denes acceptable losses on a computing windows in order to discriminate relevant results from irrelevant ones. The objective is to guarantee results satisfying QoS constraints to users. These QoS constraints are considered as xed for the complete lifetime of each query.

The global processing capacity of a distributed infrastructure is extensible through the add of congured machines. When the all resources of the cluster are exploited, some machines should be added to the congured machine set in order to limit the violation of QoS constraints. Indeed, on a cluster like Amazon EC2 9 , starting a new VM may take several minutes before new resources are available. Thus, it is important that DSMSs integrate some mechanisms to maintain the global capacity of the cluster over processing requirements. Moreover, as streams may decrease signicantly in terms of volume (e.g., a road monitoring stream when the trac is low), some congured machines may be idle for a signicant period. With the generalization of pay-as-you-consume solutions (Amazon EC2, Microsoft Azure, Google Cloud Dataow) and the emergence of GreenIT [START_REF] Murugesan | Harnessing green IT: Principles and practices[END_REF], it represents an important waste of nancial and energetic resources. So, it appears necessary that DSMSs t resource usage to processing requirements.

Congestion issue

According to the execution context presented above, tasks are assigned on processing units. Each task processes a substream which has a varying input rate. In some cases, this input rate may causes the congestion [START_REF] Heinze | Autoscaling techniques for elastic data stream processing[END_REF], Peng et al., 2015, Schneider et al., 2009, Heinze et al., 2014a, Xu and Peng, 2016] of the task and by consequence the congestion of the workow. Denition 8. (Congestion) The congestion is an execution state dened by irreversible losses of stream elements due to an overload in input of a task, an operator or a workow.

According to Denition 8, some query results may be lost denitively when a congestion appears. In a stream processing context, managing congestion of operators consists in solving some Big Data issues. Contrary to query processing over static data, an overload may happen anytime during treatments and implies an irreversible degradation of result quality. In addition, the apparition of a congestion involves a degradation of the processing latency at workow scope.

Detecting a congestion can be performed at dierent scopes (task, operator, workow). However, in order to identify the exact cause of a congestion, it is necessary to observe the co-evolution of some metrics. We identify two characteristic situations, or critical cases, causing the apparition of congestion:

• Critical case 1 (CC1): Considering a task T j i with a xed processing throughput T P j i , if its input rate is greater than its processing throughput, the task starts to accumulate pending elements on its input queue [START_REF] Stonebraker | The 8 requirements of real-time stream processing[END_REF], Babcock et al., 2002]. It delays execution of stream elements and emission of new results on output implying a degradation of the global processing latency.

• Critical case 2 (CC2): Let consider a task T 1 i sending its outputs to another task T 1 j through network interface. If the bandwidth linking those two tasks is smaller than the throughput of T 1 i , the bandwidth limits the input rate of T 1 j and slows down the global processing latency.

Managing congestion is dicult because it depends of several parameters from the complexity of operators to capacities of available processing units. To remove congestion, some adaptations of treatments and resource usage must be performed statically and dynamically to respect QoS constraints. To do so, some techniques adapting treatments and available resources to stream variations have been developed and integrated into DSMSs.

Resource management

Considering a distributed infrastructure managed as a service (IaaS) [START_REF] Nikolov | Cloudfarm: An elastic cloud platform with exible and adaptive resource management[END_REF], there are two methods to adapt congured resources to processing requirements: substituting or adding some processing units. The substitution consists in a replacement of processing units with others having dierent capacities. For example, on a cloud provider like Amazon EC2, it consists in replacing some virtual machines (VM) with a new set of VMs having greater (scale-up) or smaller (scale-down) CPU and RAM resources. Adding processing units consists in increasing (physical scale-out) or decreasing (physical scale-in) the number of processing units. According to the example on Amazon EC2, it consists in adding or deleting VMs from congured machines.

Substitution of processing units

In order to substitute processing units at runtime, some cloud-based solutions have been proposed in [START_REF] Maurer | Enacting SLAs in Clouds Using Rules[END_REF], Sedaghat et al., 2013]. These solutions are based on service level agreement (SLA) of cloud applications to dene optimal VM conguration in order to prevent SLA violations. These approaches are rule-based [START_REF] Lorido-Botran | A review of auto-scaling techniques for elastic applications in cloud environments[END_REF] and when a potential SLA violation is detected according to CPU, memory and bandwith usage, applications are moved to other VMs with more available resources. However, it has been highlighted that many operating systems do not support migration of running applications without rebooting [START_REF] Lorido-Botran | A review of auto-scaling techniques for elastic applications in cloud environments[END_REF], Nikolov et al., 2014] which is time and resource consuming. Another approach integrated in VMWare ESXI10 suggests a scale-up and scale-down approach by shared resources. It consists in overprovisioning VMs and let some CPU, memory and bandwidth resources inactive when they are not necessary to respect SLA. When a potential violation is detected, these resources are used to simulate scale-up without rebooting. The main problem of this approach is that it requires a constant overprovisioning of resources to perform the substitution at runtime. So, most cloud providers (Amazon EC2, Microsoft Azure...) only suggest to add processing units at runtime.

Add of processing units

For distributed and low latency applications, adding processing units appears more natural to adapt dynamically available resources. It is important to notice that approaches supporting the add of processing units [START_REF] Nikolov | Cloudfarm: An elastic cloud platform with exible and adaptive resource management[END_REF], Andrzejak et al., 2010, Zhu and Agrawal, 2012] consider a unique VM conguration in terms of CPU, memory and bandwidth for all processing units. So, scale-up and scale-down cannot be performed at all. In [START_REF] Andrzejak | Decision model for cloud computing under sla constraints[END_REF], authors suggest a probabilistic model taking SLA constraints as input and dening how many VMs are necessary to prevent violation. This approach is limited by reconguration costs. Indeed, while modeling the system to predict how many VMs are necessary, applications with short-term bursts imply heavy reconguration. Moreover, it does not take into account unexpected system failures and may removes useful resources. In [START_REF] Zhu | Resource provisioning with budget constraints for adaptive applications in cloud environments[END_REF], an approach relying on control-based theory is presented. This solution oers controllable parameters dening resource requirements and a feedback loop on application execution after physical scale-out and scale-in to guide manually users to optimal settings. In this chapter, we have presented some formal concepts and denitions for data stream processing. As data streams are ephemeral sequences of data arriving at uctuating rates, it is necessary to process them on the y. To scale available resources to data volumes, it is necessary that DSMSs exploit distributed infrastructures. Nevertheless, it is dicult to maintain consistently enough resources to process all incoming data. To tackle this issue, it is possible to substitute or add processing units at runtime. Nevertheless, these adaptations between processing needs and resource usage are costly in terms of reconguration time and energy. Instead of adding the global capacity of the cluster, it is possible to adjust resources consumed by each continuous query with regards to their respective processing needs. To do so, it is necessary to modify workows according to their denitions, the parallelism degree of their operators, the load balance between their tasks and the assignment of their tasks on processing units. In this chapter, we suggest a survey of congestion management in DSMSs. This survey is composed of three parts. In a rst time, we present a catalog of patterns used in the literature to manage congestion of operators at dierent scopes. Then, we explain how DSMSs trigger these patterns through an analysis of methods used to detect congestion of operators. Moreover, we highlight advantages and limits of each class of methods. Finally, we suggest an classication of distinctive DSMSs selected for their performance and popularity. This classication is based on query representation, congestion management and the expressiveness of the query language.

1 Congestion management

Principle

Congestion management is a major challenge while processing continuous queries over streams with uctuating input rates. As a congestion (see Denition 8 in chapter 2 section 2.2.2) may happen anytime, it is necessary to maintain a balance between processing needs and resource usage for each operator composing a workow. Some elastic mechanisms have been developed in order to tackle congestion issues. We group some elastic mechanisms, or patterns, into elastic mechanisms at logical layer. They have in common to modify workows such as they exploit more or less available resources instead of modifying the execution support. It allows to perform adaptation of resource usage at runtime with short reconguration times compared to substitution and add of processing units at runtime.

We present some common patterns used by DSMSs to tackle congestion issue at logical layer. We distinguish four types of patterns:

• Patterns at query level rely on algebraic properties to organize eciently logical operators according to features like the selectivity factor of operators.

• Patterns at operator level modify the the way a logical operator is executed via one or more physical operator without changing the query plan.

• Patterns at implementation level map operators to an implementation runnable on the execution support.

• Patterns at data level adapt the transmission of streams between operators according to execution parameters.

For each of them, we motivate its interest through an example highlighting a specic case. Then, we explicit assumptions made on operators (algebraic properties, denition, potential parallelism) to specify the applicability of patterns. After this, we expose the principle of each pattern. In addition, we present benets and overheads induced by each pattern. We sum up patterns with main characteristics in Table 1. More patterns and details can be found in [START_REF] Hirzel | A catalog of stream processing optimizations[END_REF] 1, each pattern apply modications which change the entire workow, a single operator, the implementation used for an operator or data lifecycle. It is also worth noting that patterns may require that operators support a type of parallelism (see Denitions 6 and 7 in chapter 2) to be eective.

Adaptation at workow level 1.2.1 Operator reordering

Congestion management Motivation

In many continuous queries, operators have dierent processing cost and produce dierent amount of data. For example, considering an application detecting heatwaves, temperature and air pollution sensors, at distinct locations send their identier with average temperature and variation of CO 2 concentration. A rst operator O 1 enriches this data with information about sensors like the model and the GPS location. A second operator O 2 lters temperature and CO 2 concentration values and forwards critical measures to an alert system. It appears that O 1 enriches stream elements discarded by O 2 downstream. As O 2 does not lter attributes modied by O 1 , some stream elements could be discarded upstream.

Assumptions

Reordering operators can only be performed under some assumptions:

• Before reordering, the upstream operator must be have a xed selectivity factor. Indeed, the probability that the downstream operator lters a stream element must remain independent from the selectivity factor of the upstream operator.

• Operators must be commutative. Moving an operator O j upstream of an operator O i requires that executing O j before O i and executing O i before O j generates same results. Indeed, let consider the attribute set A i composed of attributes added or deleted by an operator O i and A j the set of attributes used by O j . If the intersection between A i and A j is empty, then moving O j upstream of O i is safe. Indeed, attributes used by O j are available before the execution of O i .

Principle

Denition 9. (Selectivity factor) The selectivity factor of an operator is the ratio between the number of stream elements in input and output.

For example, an operator which forwards 25% of received data has a selectivity factor of 0.25. For two consecutive operators, reordering is dened as follow: Denition 10. (Operator reordering) [START_REF] Hirzel | A catalog of stream processing optimizations[END_REF] Let O i and O j be two consecutive operators respectively with a xed selectivity factor Sel i independently of value distribution in input and a uctuating selectivity factor Sel j . If Sel j becomes smaller than Sel i , moving O j upstream of O i eliminates unnecessary data exchanged between O i and O j .

Operator reordering is performed statically before the execution of the continuous query or dynamically at runtime. Thus, static reordering of operators is equivalent to logical query plan selection in traditional DBMS [Chaudhuri, 1998]. In this case, selectivity factors of operators are estimated on a subset of representative data and an optimal query plan minimizing amount of exchanged data is dened accordingly. In the other hand, dynamic reordering of operators requires to monitor selectivity factors at runtime and a support of data re-routing [Arasu et al., 2004, Abadi et al., 2005, Chandrasekaran et al., 2003]. In [START_REF] Chandrasekaran | Telegraphcq: Continuous dataow processing[END_REF], authors suggest a module, named Eddy [START_REF] Madden | Continuously adaptive continuous queries over streams[END_REF], for adaptive routing of stream elements. An Eddy receives some streams and routes each stream element to connected operators. Each output produced by operators goes to the Eddy before being forwarded to next operator. This approach allows to reorder operators through a light modication of the routing policy of the Eddy. Overheads involved by such dynamic approach have been analyzed in [Deshpande, 2004].

Chapter 3. State of the art Benets for congestion management

Operator reordering reduces amount of data exchanged between operators. It avoids risk of network bottlenecks due to massive volumes of data to route from a processing unit to another one. In addition, if a query plan contains operators with high selectivity factors and great processing latency (e.g., joins) and operators with low selectivity factors, reordering ensures that costly operators will be moved downstream and receive as less data as possible.

Motivation

With the proliferation of multicore machines and the development of distributed computing, many data-intensive applications are executed on infrastructures managing several threads in parallel. Stream-based applications should take advantage of this potential to adapt the throughput of a sequential operator to its input rate. For example, let consider a trac jam application consuming data from radars. The input rate depends on the number of vehicles observed at each time unit. An operator taking as input registration plates and extracting information about the driver from a remote database is stateless (see chapter 2 section 2.1.2) but has an important processing latency. If the operator receives important volumes in input, it requires more CPU and memory resources to avoid congestion.

Assumptions

Operator parallelization does not require modication of the continuous query only under the following assumptions:

• Operators should be naturally parallel. An operator is naturally parallel if the union of results computed on disjoint data partitions is equal to the result computed on the union of these data partitions. Operator parallelization can be performed on both stateless and stateful operators if they belong to the class of operators naturally parallel [Karp, 1988].

• For stateful operators, it is worth noting that they should keep their states disjoint or synchronized. Parallelizing a stateful operator raises the problem of state management introduced in [START_REF] Sattler | [END_REF]Beier, 2013, Gedik et al., 2014]. While tasks of stateless operators process stream elements independently from each other, tasks of stateful operators should maintain a global state [Castro Fernandez et al., 2013] or keep independent states on stream partitions. For example, a stock-price analysis application computes average Like reordering, parallelization can be static or dynamic. Static parallelization consists in selecting a number of tasks, or parallelism degree [START_REF] Mehta | Managing intra-operator parallelism in parallel database systems[END_REF], according to an expected workload. The parallelism degree is set to improve throughput with light overheads. In a stream processing context, the workload is not known before execution so dierent approaches have been developed. Some approaches [Xu andPeng, 2016, Heinze et al., 2014a] monitors input rates and processing latencies of operators and adapt parallelism degrees accordingly. It suits to parallel operators under the assumption that a greater parallelism degree improves throughput. To eliminate this assumption, some approaches based on trial-and-error learning algorithms [START_REF] Schneider | Elastic scaling of data parallel operators in stream processing[END_REF], Gedik et al., 2008, Gedik et al., 2014] aim to map throughput to parallelism degree after an exploration phase.

Benets for congestion management

Operator parallelization spreads input data of a single operator among many equivalent tasks in order to exploit data parallelism (see Denition 7). It may reduces the input size of each task depending on the number of tasks and the partitioning method. So, it limits overow in input and improves throughput as long as distribution [Deshpande, 2004] and network [Xu et al., 2014] overheads are compensated.

Task scheduling

Motivation

Processing streams on a distributed infrastructure involves communication costs. Indeed, a continuous query is represented as a DAG of operators (see chapter 2 section 2.1) where edges are inner-streams. These inner-streams are supported by the infrastructure through:

• Main memory if tasks are assigned on the same machine and can share common memory space.

• Network interface if tasks are assigned on dierent machines.

Depending on the volume of inner-streams and available resources (memory and bandwidth), it is better to assign some tasks on the same machine or on distinct machines. For example, let consider an application following the celestial bodies from telescopic images. An operator O 1 turning raw images into matrices sent to an operator O 2 compressing matrices. Then, an operator O 3 receives compressed matrices and analyzes orbits of celestial bodies. As operator O 1 and O 2 manage important volumes of data and reduce signicantly data volume through compression, it is interesting to assign tasks associated to these operators on the same machine to avoid massive network communications.

Assumptions

Task scheduling can be performed on a distributed infrastructure assuming the following conditions:

• Processing cost of operators must not exceed available resource on each processing unit as explained in [Aniello et al., 2013, Peng et al., 2015]. To avoid failure of processing units, a scheduling plan should not assign operators requiring more resources than there are on a processing unit. To avoid this situation, scheduling plans can be dened according to provided resource requirements [Peng et al., 2015] or they can be updated through operation re-assignment, or box sliding [Abadi et al., 2003[START_REF] Cherniack | Scalable Distributed Stream Processing[END_REF], according to metrics monitored continuously.

• Each processing unit must have an access to all hardware and software resources needed for computation. Commonly, stream-based applications communicate with remote and disk-based systems (e.g., databases and HDFS le systems). These communications are established within some operators which can be assigned on any processing unit, so it is crucial that all processing units have an access to remote and disk-based systems.

• States of operators must be transferred without loss. It is necessary that operator states can be moved to perform dynamic operator scheduling. To move states, some coordination systems 11 are used. They rely on checkpointing mechanisms to update operator states. So, when an operator is moved and restarted on another processing unit, its state is initialized according to the last checkpoint.

Principle

Denition 12. (Scheduling plan) Let consider a nite set of tasks

T = {T 1 i , T 2 i , ..., T k i } and a nite set of processing units M = {M 1 , M 2 , ..., M q }. A scheduling plan SP : T → M is an application assigning each task T j i on a processing unit M n with T j i ∈ T , M n ∈ M.
A scheduling plan describes the global layout [START_REF] Hirzel | A catalog of stream processing optimizations[END_REF] of continuous queries on processing units and is dened statically or dynamically. The static denition relies on operator properties in order to reach an objective [Aniello et al., 2013] (e.g., minimizing network 1. Congestion management trac [Xu et al., 2014]). It requires to have algebraic properties on operators and eventually to train the scheduler with representative data to extract selectivity factors. So, the static version is only suitable for solutions based on a query language associated to a predened algebra. Operator scheduling can be performed dynamically if a monitoring mechanism observes operator properties at runtime to compute updated scheduling plans.

Benets for congestion management

The benet of operator scheduling depends on the compromise made between resource usage and communication costs. If two operators assigned on a same processing unit have concurrent usages of some resources (CPU, memory and disk) and they receive critical data volumes in input, it is interesting to move at least one of these operators on another processing unit. The problem is that the performance improvement induced by the re-assignment should compensate communication overheads caused by the migration. On the contrary, if two tasks assigned on dierent processing units exchange important volume of data but there is a processing unit able to provide available resources without starving one of the tasks, it is preferable to collocate these tasks.

Adaptation at implementation level 1.4.1 Algorithm selection

Motivation

Let consider an application dened through a declarative language or a graphical interface (see chapter 2 section 2.1). This application computes a join between two streams S 1 and S 2 over sliding windows (see chapter 2 section 2.1.1). As streams have independent uctuations in input rate, instantaneous relations R 1 (τ i ) and R 2 (τ i ) generated at timestamp τ i respectively from S 1 and S 2 may have dierent cardinalities. Let consider that the system supports two implementations of join operation: nested-loop and sort-merge algorithms. If R 1 (τ i ) has a higher cardinality in comparison to R 2 (τ i ), nested-loop implementation delivers good performance. But if R 1 (τ i ) and R 2 (τ i ) have approximately same cardinalities, the sort-merge implementation is more ecient as it scans both relations only once [START_REF] Mishra | Join processing in relational databases[END_REF]. So, the choice of implementation may have an important impact on performance.

Assumptions

Algorithm selection assumes that continuous queries are dened through a declarative lan-guage. To ensure that algorithm selection does not degrade results or performance of the system, the following assumptions must be valid:

• All implementations of a same operator must be strictly equivalent. Indeed, if there are some specic cases where an implementation I 1 delivers dierent results than an implementation I 2 supposed equivalent, algorithm selection may aect the semantic of the operator and the continuous query may return wrong results. It is specically important after operator parallelization because some operators cannot keep their original semantic (e.g., standard deviation).

• Implementations of operators dened through a declarative expression must be available on all processing units. If operator scheduling is performed only statically [Aniello et al., 2013], a site-aware scheduling of operators may ensure that all operators have available implementations where they are assigned. But if operator scheduling is performed dynamically to avoid congestion or recover from a failure, the availability of implementations is not guaranteed anymore. Of course, this issue cannot happen when continuous queries are dened through an imperative language as the implementation is moved with the assignment of the operator on a processing unit.

Principle Algorithm selection is common in database systems. It corresponds to physical query plan denition [Garcia-Molina, 2008, Graefe, 1993] mapping operators from logical algebra (e.g., sort or join) to the physical algebra (e.g., heapsort, mergesort, hash join or nested loop join).

Denition 13. (Physical query plan) Let consider the set of tasks

T = {T 1 i , T 2 i , ..., T k i }. Let I = {I 1 , I 2 , .
.., I m } be the set of available implementations. A physical query plan QP : T → I is an injective application which associates an implementation I j ∈ I to each task T j i ∈ T .

According to Denition 13, the cardinality m of implementation set is equal or greater than than the cardinality n of operator set if we do not count equivalent tasks generated after operator parallelization as presented above. Moreover, it is important to note that in the case of operator parallelization, some implementations cannot be used in parallel or involve important overheads to merge results on partitions. Algorithm selection can be performed dynamically to adapt the implementation to input properties. In [Abadi et al., 2005], authors suggests a runtime mechanism having a static set of implementations for each operator and picking the fastest one according to input properties.

Benets for congestion management

Algorithm selection may increase processing latency of operators if there are a faster implementation than the current one available locally. The speedup may be absolute (e.g., quicksort over bubblesort) or relative (e.g., sort-merge join over nested-loop join) according to input properties.

Assumptions

As load balancing becomes relevant after operator parallelization, all assumptions mentioned for operator parallelization are valid for load balancing. In addition, load-aware strategies require a strict equivalence between tasks. Indeed, let consider a stateful operator computing average values of stock prices for multiple stock label. The operator computing average values can be naturally parallelized if disjoint subsets of stock labels are associated to each task. In such case, each task is only able to process a partition of stream elements so load imbalance cannot be corrected. Principle Denition 14. (Load balancing) Let consider a set of equivalent tasks T = {T 1 i , ..., T k i } and the set of their respective loads L = {L 1 , ..., L k } according to a load metric. For any nite subset of stream elements E = {e 1 , ..., e n }, balancing load consists in nding k partitions of E such as the standard deviation of loads in L is minimized at any time.

The choice of the strategy for load balancing and the metric load depends mainly on assumptions on stream properties. Load balancing strategies can be classied in two main categories [START_REF] Pearce | Quantifying the eectiveness of load balance algorithms[END_REF]]:

• Value-aware strategies [START_REF] Neumeyer | S4: Distributed stream computing platform[END_REF] route stream elements according to key values without consideration for balance between tasks.

• Load-aware strategies [START_REF] Rivetti | Online scheduling for shue grouping in distributed stream processing systems research paper[END_REF] tend to balance load between threads. They consider load balance in term of execution time or resource consumption (CPU, RAM).

Benets for congestion management

Load balancing reduces imbalance between equivalent tasks processing stream elements in parallel according to data parallelism. As the global throughput of such parallel region is directly impacted by the slowest operator, load balancing improves global throughput proportionally to the skew in data distribution. Without load balancing, operator parallelization does not improve signicantly the global throughput of an operator.

Load shedding

Motivation

Let consider a website suggesting articles to buy online. To suggest in priority interesting products according to user behavior, the website monitors clicks to identify points of interest. Monitoring logs are consumed in real time by a stream-based application identifying behavior of users and suggesting relevant products. The distributed infrastructure executing this application is calibrated to analyze logs generated by a standard trac on the website. When the festive season starts, the trac increases by orders of magnitude causing an overow in input of the stream-based application. In such case, the owner of the website is no more interested in suggesting products very relevant products to each users and accepts to degrade result quality. It avoids to add resources for this specic period through a limitation of the maximal input rate.

Assumptions

As mentioned in motivation of load shedding, this pattern degrades result quality so there is no assumption guaranteeing the preservation of result quality. The single assumption for load shedding concerns the control of the degradation of result quality. Actually, load shedding is enabled to avoid uncontrollable loss of stream elements. The load shedding method must provide a feedback on dropped elements to estimate the impact on nal results.

Principle

Denition 15. (Load shedding) Let consider a task T j i with an average throughput T P dened in stream elements processed per time unit and a potentially innite multiset S = {S 1 , S 2 , ..., } composed of nite sets arriving at each time unit. Load shedding consists in nding for each set S i ∈ S a subset S i such as the cardinality of S i remains lower or equal to T P .

Load shedding inverts the adaptation compared to patterns presented previously. Indeed, while other patterns adapt the DAG of operators and its execution on an infrastructure, load shedding adapts streams to available processing rates [START_REF] Tatbul | Staying t: Ecient load shedding techniques for distributed stream processing[END_REF]. In [START_REF] Tatbul | Load shedding in a data stream manager[END_REF], authors suggest load shedders which drop stream elements to satisfy some QoS metrics about overall latency and precision of nal results. Aurora* analyzes impact of operators on those metrics through their processing latencies and selectivity factors [Abadi et al., 2003[START_REF] Cherniack | Scalable Distributed Stream Processing[END_REF]. In [Arasu et al., 2004], multiple techniques for load shedding are suggested including load shheding through statistical approximation of aggregative queries. The distributed solution Borealis [Abadi et al., 2005, Tatbul et al., 2007, Ahmad et al., 2005] suggests several techniques to control quality degradation caused by load shedding including window-aware load shedding [START_REF] Tatbul | Window-aware load shedding for aggregation queries over data streams[END_REF].

Benets for congestion management

Load shedding avoids overow in input of tasks. So, it helps to maintain a maximal throughput when initial input rates exceed processing throughput. Load shedding can also be used before 2. Detection methods for congestion management operators generating multiple outputs per stream element in input like joins. It allows to control output rate and have an approximation of result quality.

To sum up, we presented a collection of common patterns adapting continuous query execution or streams in order to avoid congestion. They can tackle issues involved by some specic cases under certain assumptions on the execution context. Nevertheless, nding the right moment to trigger a reconguration is complex. On one hand, as adaptation patterns have overheads during and potentially after reconguration, it is not advisable to use them if they do not eectively compensate a degradation in the execution. On the other hand, waiting that eective congestion occurs has important eects on system stability and the recovering time (i.e., time elapsed between reconguration and an optimal throughput). To trigger reconguration with a high probability that it compensates eectively a degradation of the execution, some detection methods have been developed and integrated to DSMSs.

Detection methods for congestion management

As presented in section 1, some specic cases can be encountered while processing streams. The eectiveness of patterns presented above relies on the reactivity of the system. We distinguish two main categories of detection methods. On one hand, on user-demand methods [START_REF] Xu | Stela: Enabling stream processing systems to scale-in and scale-out on-demand[END_REF] rely on users to trigger a reconguration when a degradation of performance or result quality is observed. When user alerts the system, recent history of the behavior of operators, which may be an instantaneous snapshot, is considered to perform appropriate modications. On the other hand, some automatic methods [Xu et al., 2014, Gedik et al., 2014, Schneider et al., 2009] suggest monitoring mechanisms observing at runtime some metrics about behavior of operators and triggering automatically adequate modications to respect some QoS constraints.

In this section, we aim at giving an overview of detection techniques to manage congestion of operators. Each method is presented with its principle, advantages and limits illustrated by main variants.

On user-demand methods

Methods relying on user demand [Aniello et al., 2013, Xu and[START_REF] Xu | [END_REF] performs operator parallelization and scheduling when user triggers a reconguration. The aim is to nd an adapted conguration according to a recent history of operator behaviors. For each operator, metrics like input rate, throughput and processing latency are monitored at regular interval. When the user triggers a reconguration of a continuous query, the system takes these measures into account to decide which critical cases are encountered and which patterns are appropriate to satisfy QoS constraints.

In [Aniello et al., 2013], authors suggest an oine scheduler aiming at generating a low network trac between processing units. To do so, the scheduling algorithm has as inupt the workow representing a continuous query and returns a scheduling plan minimizing the number of edges between processing units. As the scheduler is oine, it is only when users submit or re-submit continuous queries on the system. In [START_REF] Xu | Stela: Enabling stream processing systems to scale-in and scale-out on-demand[END_REF], authors present an online algorithm enabling operator parallelization at runtime. This algorithm, named Stela, takes as input user action like add 2 processing units and a threshold discriminating critical operators from normal ones. This threshold is given by user and species the upper bound of the ratio input rate on processing rate. Stela returns the set of operators to parallelize in order to avoid congestion of critical operators.

Benets brought by approaches on user-demand mainly depend from the master user reactivity and expertise. By, master user, we refer to the user having credentials to recongure a given continuous query. As data streams may have uctuations in input rate and value distribution over time, a xed conguration may not deliver optimal performance over time. Moreover, a given conguration may lead a workow to congestion (e.g., a scheduling plan generating network bottlenecks).

Logically, limits of approaches on user demand are users. In [Aniello et al., 2013], as a scheduling plan is generated only on user demand, users have to re-submit continuous queries when there are some execution bottlenecks which could be eliminated through scheduling of operators. In addition, the oine scheduler does not have information about operator properties (selectivity factor or processing latency) so it is assumed that operators do not amplify or downplay a variation in input rate. For example, a cartesian product amplies signicantly a variation in input rate. This assumption is irrelevant for most stream-based applications ltering stream elements. In [START_REF] Xu | Stela: Enabling stream processing systems to scale-in and scale-out on-demand[END_REF], if a user adds only few processing units when there are massive overow, performance is not signicantly improved while reconguration involved important overheads. Moreover, users may trigger reconguration with an important oset after some operators start to accumulate stream elements on their input queues.

Automatic methods

Stream processing requires high reactivity to maintain treatments while facing critical uctuations in input rate and value distribution. To ease the management of continuous queries from user point of view, most elastic solutions [Abadi et al., 2003[START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF], Arasu et al., 2004, Abadi et al., 2005, Schneider et al., 2009, Gedik et al., 2014, Xu et al., 2014] integrate automatic and dynamic mechanisms to detect critical cases and trigger reconguration of the system according to metrics monitored continuously. These metrics are observed at dierent scopes of the execution:

• At infrastructure scope, some global metrics are observed like the network trac [Xu et al., 2014],

CPU load and memory usage on each processing unit [START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF].

• At operator scope, metrics describing the execution are observed. It concerns essentially input and output rate [START_REF] Schneider | Elastic scaling of data parallel operators in stream processing[END_REF], Gedik et al., 2014], processing latency [Abadi et al., 2005] and selectivity factors [Abadi et al., 2003, Arasu et al., 2004] of operators.

Automatic methods considering these metrics in input can be grouped two main classes [START_REF] Das | Model-based and model-free approaches to autonomic resource allocation[END_REF], Lorido-Botran et al., 2014]: reactive and proactive.

Reactive methods

Like on user demand methods, reactive methods rely on user expertise but not on its reactivity. User expertise is required before treatments to dene what is a normal state of the system. By state, we consider a set of metrics describing the execution of some continuous queries. There are two types of reactive methods: threshold-based and reinforcement learning-based approaches.

Detection methods for congestion management

Threshold-based

Threshold-based algorithms takes as input maximal and minimal thresholds for monitored metrics at infrastructure [START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF], Xu et al., 2014] and/or operator scope [Abadi et al., 2003, Arasu et al., 2004, Abadi et al., 2005]. These thresholds compose a policy dening conditions for the stability of the system. When the value of a monitored metric is under the minimal threshold or over the maximal threshold, some patterns for congestion management are applied. The objective is to maintain a stable state of the system.

Threshold-based algorithms ensure that a given policy is respected but they are strictly reactive. Actually, only an eective violation of the policy triggers a reconguration of the system. Depending on user expertise, the policy may allow temporary congestion of operators having an important impact on performance of the system and result quality. Moreover, some thresholds like input/process rate may be more critical for some operators like join which generate potentially multiple output per input element and have intrinsically a greater impact on result quality. Adjusting thresholds may be laborious according to the complexity of a continuous query and the execution support.

Reinforcement learning-based

Reinforcement learning-based algorithms (RL-based) aim at reaching a performance goal through interactions with the system but without a priori knowledge. In RL model, an agent receives as input a state describing the execution context (e.g., input rate and throughput of an operator) and applies an action a i from the set of all possible actions A = {a 1 , a 2 , ..., a n }.

According to the evolution of scalar metric, the agent receives a reward as feedback. Through a learning phase, the agent maps rewards to actions for each encountered states. The state-actionreward map is denoted the policy of the agent. Once the agent meets a known state, it applies the best action according to its policy and updated it according to the new reward.

In [START_REF] Heinze | Autoscaling techniques for elastic data stream processing[END_REF], Heinze et al., 2014a], authors present a distributed reinforcement learning solution, named FUGU, which maintains independent policies on each processing unit. This solution presents the advantage to be independent from workload as it is able to converge to a appropriate conguration. A problem is the convergence time which could be important if the workload varies often. An other limit of this approach is that it considers exclusively variations in input rate to associate parallelism degrees without taking distribution of values into account. Thus, if the distribution of values has an important impact on the processing latency and changes signicantly over time, the knowledge base built through training may be obsolete quickly.

Control-based

Control systems [De Matteis and Mencagli, 2016] aim at automating the management of online environments. There are three classes of reactive control-based systems [START_REF] Lorido-Botran | A review of auto-scaling techniques for elastic applications in cloud environments[END_REF]: open loop and feedback.

Open-loop approaches relies on a model of the system and a controllable metric which should not exceed some thresholds according to the model and the current state of the system. The controller does not receive feedback about the eectiveness of reconguration making this solution limited if the system diers from the model over time. For feedback controllers, the model of the system is replaced by an output metric associated to a desired value. Any signicant deviation from this value triggers a reconguration. For example, considering that the output metric is the throughput of an operator, the controller may increase the parallelism degree when the throughput decreases at constant input rate.

In cite [START_REF] Gedik | Elastic scaling for data stream processing[END_REF], authors suggest a feedback controller relying on a congestion metric which indicates if an operator is potentially critical. When the value of the congestion metric degrades (increase of input queue sizes), the algorithm triggers operator parallelization for the overloaded operator. To control if the reconguration is benecial to the execution, the evolution the throughput is observed. If the control metric evolves positively, the reconguration is repeated until the congestion metric reaches a satisfying value according to QoS constraints. Control-based algorithms oer more exibility than threshold-based approaches as it only require that users dene congestion and control metrics. The main limit of control-based approach is the accuracy of reconguration [START_REF] Gedik | Elastic scaling for data stream processing[END_REF]. Indeed, if the reconguration has a minor impact of the congestion metric, it will require several reconguration before stabilization.

Proactive methods

Proactive methods aim at anticipating the behavior of system to avoid congestion before a critical workload appear. Contrary to reactive methods, these methods might trigger unnecessary reconguration because of sudden variations of workload (e.g., an input rate increasing continuously for a signicant time and decreasing massively within a short period) but they react faster to specic cases leading to congestion. On one side, control-based and queuing-based approaches use a model to predict how the system reacts to modications of the workload. On the other side, time-series based approaches estimate the behavior of the system without model but from recent history.

Control-based

A third class of control-based systems, denoted feed-forward [START_REF] Lorido-Botran | A review of auto-scaling techniques for elastic applications in cloud environments[END_REF], try to anticipate deviation from the desired value in output. According to the model, they predict the behavior of the system with current inputs and reacts before a deviation appears. In most cases, feed-forward controllers are combined with feedback controllers to compensate prediction errors.

Queuing-based

Queuing-based algorithms use mathematical methods associated to te queuing theory as presented in [Kalashnikov, 2013] to model the execution of operators having pending queues as presented in section 1. To model an operator through queuing theory, several parameters are required:

• Stream elements are supposed to arrive at a mean rate λ and are enqueued before being processing.

• The operator has an average throughput µ for all possible values of stream elements.

• The inter-arrival time distribution A which corresponds to the distribution law describing the evolution of λ over time.

• The service time distribution B which corresponds to the distribution law describing the evolution of µ according the arrival rate λ.

• The number of operators C in the case multiple operators share a common pending queue (e.g., after parallelization of operators).

• The system capacity K which corresponds to queue length.

• The calling population N which is the size of the stream. In the case of innite streams, the queuing model is said open.

• The service discipline D which matches with the priority order in the execution context. In most systems, D is FIFO (First In First Out).

According to these parameters, a system is modeled according to the Kendall notation A/B/C/K/N/D or just A/B/C as K, N and D are optional as presented in [Kendall, 1953]. Actually, when not precised K and N are considered as innite and D is FIFO. A and B are generally equal to M, D or G. M refers to Poisson distribution, D for uniform distribution and G corresponds to any general distribution with known mean and variance.

The aim is to approximate the response time R for an incoming stream element according to the queuing model in Kendall notation. For example, the response time for a M/M/1 is R=

1 µ-λ .
In [START_REF] Jiang | Queueing analysis of relational operators for continuous data streams[END_REF], authors present models for Select-Project-Join operators according to queuing theory in a stream processing context. It allows to estimate response time of these operators to apply dynamically operator reordering and parallelization to respect some QoS constraints.

Time-series based

Time-series refer to sequence of values generally measured at regular time interval. Timeseries analysis can be used for two goals: forecasting future values of a sequence or nd repeating patterns to extrapolate future values. According to an estimation of future values, a decisionmaking policy denes if the system should be recongured and how.

Let focus rst on forecasting problem. From a recent history of monitored metrics, the aim is to predict resource usage or workload. Common forecasting techniques are Moving Average, Auto-regression, the combination of the two, denoted ARMA [START_REF] Li | Distribution of the residual autocorrelations in multivariate arma time series models[END_REF] and machine-learning techniques.

Moving Average assumes the next value y t+1 is the weighted average of the last n observed values x i as follow:

y t+1 = a 1 x 1 + a 2 x 2 + ... + a n x n (1)
such as the sum of weights a i is equal to 1. From this general denition, three variants exist:

• Simple Moving Average associates a weight 1 n to each measure x i . It corresponds to the arithmetic average.

• Weighted Moving Average associates dierent weights to measures. Commonly, the older is the stream element, the lower is the associated weight. So, it gives more weight to recent measures.

• Exponential smoothing assigns decreasing weights over time to measures according to an exponential distribution.

Auto-regression of order n relies on the same principle than Weighted Moving Average but weights are computed as auto-correlation coecients. The correlation between former forecasted values y t j with j < t + 1 is used to calculate weights for each timestamp.

Auto-regressive Moving Average or ARMA combines Weighted Moving Average with Autoregression. The next value is a weighted average of previous forecasted values and values of the recent history.

Machine learning techniques include statistical and neural network-based methods. Statistical methods corresponds to regression techniques. Regression aims at determining the polynomial function such that it minimizes the distance to each point composing the recent history. The particular case of polynomial function of order is denoted linear regression. This function is used to predict future values. Methods based on neural networks consider a group of neurons interconnected on several layers from input layer having some measures to output layer returning a result. In a stream processing context, the input layer contains one neuron for each measure in the recent history and the output layer contains one neuron for the expected value. The network is trained with initial weights chosen randomly. These weights are adapted through the neural network in order to nd a polynomial function which returns the expected value from measures.

To sum up, relying on user demand to manage congestion involves that it is assumed that the user monitors continuously states of the cluster and continuous queries. In addition, users should have an expertise to operate appropriate modications of the execution context at physical and logical layers. As many applications composed of several operators consume streams varying frequently, some automatic solutions have been proposed. Some of them react to eective situations leading to congestion while other anticipate the behavior of systems to identify needs of reconguration to respect QoS constraints.

Classication of distributed DSMSs

We have suggested a formalization of streams and continuous queries. In addition, we presented challenges involved by stream processing and described elastic mechanisms for congestion management. In this section, we aim at giving an overview of principal distributed DSMSs developed on last years. This collection is neither exhaustive nor a strict top-tier ranking but it covers most DSMSs in terms of variety according to classication criteria. Centralized DSMSs like STREAM [Arasu et al., 2004], Aurora [Abadi et al., 2003], Medusa [Cetintemel, 2003] and Nia-garaCQ [START_REF] Chen | Niagaracq: A scalable continuous query system for internet databases[END_REF] are detailed in former surveys [START_REF] Babcock | Models and issues in data stream systems[END_REF], Stephens, 1997]

Criteria of classication

As illustrated on Figure 8, we suggest a classication of DSMSs based on the paradigm used to represent continuous queries. It has a major impact on the expressiveness of the denition language and the eectiveness of data and task parallelisms aecting congestion management. As presented in chapter 2 section 2.1.4, two paradigms are commonly used to represent continuous queries: workow and MapReduce. In addition, a secondary criterion is the expressiveness of the language. Indeed, while processing streams, it is necessary that query languages support specic operators like windows to process streams with stateless and stateful operators. Then, detection method for congestion management denes the reactivity of the system and the conguration eort required to maintain respect QoS constraints at runtime. Finally, according to the stream classication suggested in chapter 2 section 1.2, we highlight which types of streams each DSMS is able to process eciently and detail which properties determine that choice. The open-source DSMS Apache Storm12 integrates a resource-aware scheduler since its version 1.x.x and is also called RStorm [Peng et al., 2015]. RStorm oers a high exibility for query denition and performance tuning. It allows users to specify processing requirements of each operator in terms of CPU and memory and guarantee minimal waste of resources according to these specications.

Query denition: RStorm relies on a high level language (Java, Scala, Python) for query denition. A continuous query is dened as a sequence of user-dened operators subscribing to sets of streams and generating zero, one or many streams. Each operator follows a template which corresponds to stateless or stateful operators. RStorm integrates a growing collection of templates corresponding to complex relational operators like Joins.

Query representation: A continuous query is represented as a workow of operators. RStorm considers two types of operators:

• Spouts are connectors to stream generating services like sensors, databases or a distributed messaging service like Apache Kafka 13 . Spouts are responsible to manage the lifecycle of stream elements through a workow. Each spout produces one or many streams.

• Bolts is an operator consuming one or many streams and producing a set of streams which may be empty. Bolts follow templates for stateless or stateful operators.

Each operator may or may not support data parallelism. However, for stateful operators, users take charge to declare and update states. RStorm serves as guarantee that states are moved when the operator is moved or restarted after node failure.

Window support: RStorm supports time-based and count-based sliding and tumbling windows through specic templates. Windowing templates take as parameters window size, an optional slide and a function describing business logic.

Congestion management: RStorm is able to adapt continuous queries at operator level exclusively. Indeed, operator scheduling is performed at the initialization of a continuous query according to processing requirements specied by users. RStorm relies on a greedy algorithm to nd a near-optimal scheduling plan which maximize resource usage on a minimal subset of available machines. Operator parallelization are triggered on user demand with the constraint that users assume that parallelized operators support data parallelism without changing the semantic of results (e.g., parallelization of a mean operator).

Target streams: As RStorm relies on user expertise and reactivity, it is basically designed to handle even steady streams involving rare modications at runtime. The absence of load shedding mechanism and the guarantee that each stream element is processed exactly once suppose that users set adequate conguration in terms of parallelism degrees and processing requirements according to input volumes.

Sonora

Sonora [Yang et al., 2012] is platform for mobile cloud computing. Sonora is designed to support the execution of distributed cloud services on mobile devices. Sonora integrates a stream engine able to perform fast failover after node failure as mobile devices supporting computations come in and out frequently from the cloud. In addition, Sonora has to be energy ecient to respect battery constraints of mobile devices.

Query denition: Sonora provides interfaces for continuous query denition in a high level language. These interfaces allow users to dene functions on streams without restriction. A continuous query is dened as a succession of operators subscribing to some streams and generating an output stream.

Query representation: Continuous queries are represented as workows of user-dened operators. Even if users dene stateless operators, Sonora may group and send stream elements into batches in order to save battery. Indeed, it allows to send a batch of stream elements and turn o the radio instead of keeping the radio active and send stream elements continuously as they arrive.

Window support: Sonora supports time-based and count-based sliding and tumbling windows. Windows are explicitly dened by users through the denition of timespans on streams.

Sonora computes incrementally results on sliding windows thanks to overlaps between consecutive iterations as discussed in chapter 2 section 2.1.1. In addition, Sonora computes incrementally mean, variance and Discrete Fast Fourier Transformation (DFFT) operators over sliding windows. Incremental computations for these operators remove redundant computations in order to save energy.

Congestion management: Sonora performs adaptation of continuous queries at operator and data level. Operator parallelization is managed through the storage system PacicA which splits dynamically streams into partitions as presented in [START_REF] Lin | Pacica: Replication in log-based distributed storage systems[END_REF]. It allows to perform operator parallelization while PacicA takes charge of splitting input streams and balancing the load at runtime. Sonora can also perform load shedding through a sampling mechanism. It allows Sonora to maintain a xed rate in input that the execution support (i.e., a mobile cloud) can handle.

Classication of distributed DSMSs

Target streams: Sonora is suitable to process even and uneven steady streams like sensors streams. As Sonora does not integrate mechanisms to detect congestion at operator scope, it is not adapted for uctuating streams. Adaptation mechanisms integrated in Sonora aims at recovering after failure more than facing uctuations in input streams. Moreover, the load shedding mechanism consists in turning a massive stream (steady or uctuating) into a steady stream with a maximal input rate.

Millwheel

Google Millwheel [Akidau et al., 2013] is a distributed DSMS designed to analyze data streams at Internet scale. Millwheel is fault-tolerant as the failure of any active machine does not aect results delivered to users. To guarantee fault-tolerance, operators communicate through the following pattern: An operator O 1 applies a function on a stream element, creates a checkpoint locally and returns a result to an operator O 2 . As soon as O 2 as processed the result, the checkpointing mechanism saves the state of O 2 and sends a acknowledgement message to O 1 to update its current state.

Query denition: Continuous queries are dened in Millwheel according to a programming model hiding parallelism and concurrency issues to developers. The programming model is designed to facilitate the denition of low latency streaming applications. Each continuous query is dened as sequence of user-dened operators like in RStorm, but with a main dierence on operator patterns. When bolts consume stream elements described by a set of attributes, operators in Millwheel consume stream elements linked to a key and users must implement a key extractor. Indeed, the business logic of an operator is applied exclusively on that key. For example, an operator computing the average price of a stock, the key extractor returns the price among the list of values describing the stock and apply mean function on the price.

Query representation: Millwheel represents continuous queries as workows of user-dened operators.

Window support: Millwheel supports time-based sliding and tumbling windows through its API. A specic operator pattern allows to dene a stateful operator over a computation window like RStorm.

Congestion management: Millwheel adapts continuous queries at operator and data levels.

It supports operator parallelization and load balancing through a threshold-based algorithm which detects lack of available resources in terms of CPU and memory. When an overload is detected by local monitors, an operator can be parallelized on another processing unit to share the load. Load balancing is performed by key grouping among tasks of an operator. For example, considering an operator which consumes keys in a range from 0 to 9 and has 2 tasks, the input stream in partitioned in two partitions, one receiving keys between 0 and 4 and another receiving keys between 5 and 9. It assumes that the distribution of keys over stream elements is uniform over time.

Target streams: Millwheel supports eciently even erratic streams in terms of volume as it relies on a threshold-based algorithm to detect and correct potential congestion of operators.

Nevertheless, its load balancing strategy based on key partitioning assumes that value distribution over keys is almost uniform at any time and processing latency associated to keys are pretty similar.

TelegraphCQ

TelegraphCQ [START_REF] Chandrasekaran | Telegraphcq: Continuous dataow processing[END_REF]] is a exible solution enabling fault-tolerant stream processing over PostgreSQL. Originally, it was designed for centralized multicore architecture but it can be extended to distributed infrastructure through an extension of the FluX module [START_REF] Chandrasekaran | Telegraphcq: Continuous dataow processing[END_REF].

Query denition: As TelegraphCQ is built over a PostgreSQL DBMS, it supports declarative expressions written in SQL and includes window statement. The denition is enriched with imperative expressions like for loops to apply treatments only on a nite number of window iterations.

Query representation: TelegraphCQ represents continuous queries as workows having a star topology and associated to a routing policy. Indeed, the heart of each continuous query is a module named Eddy [START_REF] Madden | Continuously adaptive continuous queries over streams[END_REF], Chandrasekaran et al., 2003] which receives the initial stream to process and route stream elements to operators according to its routing policy. Workows in TelegraphCQ are composed of common operators (Select, Project, Join) of the relational algebra.

Window support: TelegraphCQ supports multiple window semantics: time-based and countbased sliding and tumbling windows but also landmark windows [START_REF] Chandrasekaran | Telegraphcq: Continuous dataow processing[END_REF] which have a growing size over time. Indeed, landmark windows are only dened by a slide and consider all elements since the start of the execution. Each iteration of a landmark window is superset of previous iterations.

Congestion management: TelegraphCQ integrates adaptation mechanisms at workow, operator and data levels. To do so, TelegraphCQ relies on a threshold-based algorithm to detect potential congestion of operators. It is able to perform both operator reordering and load shedding through the Eddy module. Indeed, as the routing policy associated to the Eddy module denes the execution sequence of operators, a modication of this policy corresponds to operator reordering. The Eddy module can also discard some stream elements to reduce the workload in input of an operator. Operator parallelization and load balancing are supported in TelegraphCQ by a FluX module which is responsible to manage distribution of stream elements routed by the Eddy. FluX can add tasks of an operator and decides how inputs are distributed among them.

Target streams: TelegraphCQ has been designed to process eciently even and uneven erratic streams. Indeed, the design of TelegraphCQ is centered on dynamic Eddy modules to route streams and shed load anytime without reconguration cost but with permanent overheads [Deshpande, 2004]. So, it is assumed that these overheads are compensated by frequent adaptations due to uctuations in input rate and value distribution. Even if TelegraphCQ can also manage eciently periodic streams, there is no model or learning-based modules to optimize adaptation to such streams. Finally, steady streams are processed by TelegraphCQ with permanent overheads that are never compensated as soon as an appropriate conguration has been found.

Borealis

Borealis [Abadi et al., 2005, Ahmad et al., 2005] is a distributed DSMS which inherits the stream processing engine from Aurora and its management policy of distributed infrastructures from Medusa [START_REF] Balazinska | Load management and high availability in the medusa distributed stream processing engine[END_REF]. The main improvement brought by Borealis is the support of revisions at runtime. In a real-world context, some stream sources may send stream elements out-of-order or send incorrect values and correct them later. For example, in a context of nancial market analysis, some values may be out-of-date for some stock prices. A correction is sent after to update the value. Borealis integrates mechanisms which support revisions of results being computed to soften errors in a real-world context.

Query denition: Borealis inherits the graphical denition interface of Aurora and its boxesand-arrows query model introduced in [Abadi et al., 2003]. Users select operators from a predened set, named SQuAl [START_REF] Cherniack | Scalable Distributed Stream Processing[END_REF], and dene a graph which is considered as optimal. Indeed, even if Borealis integrates predened operators, there is no query optimization based on relational algebra. The set of operators supported by Borealis includes common operators like Select-Projection-Join and some aggregate operators as sort and union.

Query representation: Each continuous query is a graph where vertices are boxes which implements an operator. Boxes are linked by arrows which specify the routing policy of stream elements through the graph.

Window support: As Aurora, Borealis supports time-based and count-based sliding and tumbling windows through an Aggregate operator which takes as parameters a function, an order O for grouping, a window size and a window slide. For example, computing the average price on the last stream elements according to the chronological order hour each 30 minutes corresponds to an Aggregate operator taking as parameter the average function on the attribute price, the order On Time, a size 1 hour and a slide 30 minutes.

Congestion management: Borealis supports adaptation of continuous queries at each level: workow, operator, implementation and data. It relies on a feedback control-based algorithm as presented in section 2.2. When a deviation in throughput is detected, Borealis can trigger operator reordering, operator scheduling or load shedding at runtime. Operator reordering is based on a cost model between two operators. According to SQuAl algebra, if operators are commutative and selectivity factors measured online satisfy conditions presented in section 1, Borealis inverts operators. Operator scheduling relies on bottleneck detection. When stream elements are accumulated on input queues, Borealis looks on the processing unit if it is due to a lack of CPU or bandwidth. In both cases, Borealis moves the operator on a processing unit which is more available according to the lacking resource. If the scheduler cannot nd such processing unit, it enables load shedding through two phases. First, according to an algebraic analysis, Borealis may add Projection operators to reduce data volumes. If it is not feasible, load shedding is triggered on operators which aects the less result quality.

Target streams: Borealis has been designed to manage real-world even and uneven erratic streams and even support out-of-order and incorrect streams. While facing a signicant variation in input rate or value distribution, Borealis tries to adapt dynamically the execution of continuous queries at query and workow scopes and if it cannot nd a satisfying conguration, it sheds load to respect QoS constraints about performance. As TelegraphCQ, the absence of model and learning-based mechanisms denotes that Borealis has not been optimized for per-pattern streams.

Finally, Borealis supports algorithm selection through a mechanism similar to Eddy module. The SQuAl compiler provides multiple implementations of predened operators on each processing unit and depending of input properties, the implementation may be changed at runtime.

ESC

ESC [START_REF] Satzger | Esc: Towards an elastic stream computing platform for the cloud[END_REF] is cloud-based DSMS designed for real-time demands such as online data mining. The particularity of ESC is the capacity to support elasticity at infrastructure and workow levels. Indeed, ESC can attach and release VMs at runtime to adapt the number of congured machines to processing requirements.

Query denition: ESC oers a programming model based on key/value pairs as MapReduce model [START_REF] Dean | Mapreduce: Simplied data processing on large clusters[END_REF]. Users dene operators according to programming patterns which take as input a stream element or a set of stream elements. Operators are dened in a high denition language named Erlang [Wikström, 1994] supporting natively parallel and concurrent programming.

Query representation: Continuous queries are represented as workows of user-dened operators. In addition to query denition, users can dene rewriting rules for some operators. For example, a user-dened operator O i can be associated to a rewriting rule split→ O i →merge to trigger operator parallelization while needed. As mentioned above, ESC manages elasticity at physical and logical levels. At physical level, ESC integrates mechanisms to perform horizontal elasticity. After a physical scale-out, ESC can perform operator scheduling to avoid bottleneck due to a lack of CPU on active VMs. ESC relies on rewriting rules to perform operator parallelization as there are not algebraic properties on operators if users do not declare it explicitly.

Target streams: ESC centralizes the management of continuous queries and infrastructure in a single decision system. Combined to a reinforcement learning-based algorithm, ESC can add and delete processing units to execution support periodically. It makes ESC ecient to process even bounded streams after a learning phase. During that phase, ESC builds a knowledge base which covers the range of rates in input without shedding the load when data volumes become important. Intuitively, the convergence time to a complete and accurate knowledge base over any uneven or erratic streams is potentially innite as any variation may happen anytime.

System S

System S [START_REF] Amini | Spc: A distributed, scalable platform for data mining[END_REF], Jain et al., 2006] is a distributed DSMS developed at IBM supporting structured and unstructured data stream processing. It is fault-tolerant and integrates security mechanisms in a distributed context.

Query denition: System S relies on the language SPADE [START_REF] Gedik | Spade: The system s declarative stream processing engine[END_REF] for continuous query denition. This language oers a set of stream-oriented operators which are specic to a domain (e.g., signal processing or data mining) in addition to relational operators like Select-Project-Join. Users dene a continuous query as a sequence of SPADE operators consuming and generating specied streams. SPADE relies on a logical and physical algebra turning logical operators into sets of physical operators. Each operator in SPADE may be executed in parallel without specications from users.

Query representation: System S represents continuous queries as workows where vertices are Processing Elements (PE) linked by inner-streams. A PE is composed of a working receiving input streams and three types of threads taking charge of routing and processing stream elements but also adapting the throughput of the PE to variation of workload.

Window support: System S supports time-based sliding and tumbling windows [START_REF] Wu | Challenges and experience in prototyping a multi-modal stream analytic and monitoring application on system s[END_REF] through the use the operator Aggregate similar to the Aggregate operator included Borealis.

Windows can also be declared through the use of the operator Punctor [Biem et al., 2010] which apply a stateless operator on stream elements and inserts punctuations in output streams according to a predened interval.

Congestion management: System S integrates mechanisms to adapt continuous queries at operator and data levels. This DSMS is able to perform operator parallelization for each PE individually. Indeed, for each PE, the dispatch thread reevaluates periodically the number of worker threads to enable so it performs operator parallelization according to a reinforcementbased algorithm at PE scope. When an overload is detected, the dispatch thread increases the number of worker threads until there is no improvement of the throughput. When the workload decreases signicantly, the dispatch thread puts some worker threads into sleep. Worker threads are not deleted because it is cheaper to wake up a thread than creating it. Through a continuous monitoring of window contents, System S is able to perform load shedding when data volumes are critical in input of PEs running stateful operators.

Target streams: PEs manage parallelization individually according to a reinforcement learning algorithm. This design ts to even bounded streams as each PE builds its own knowledge base and is able to adapt quickly its parallelism degree according to input rate after a learning phase. Moreover, when some working threads are not necessary to handle the current workload, they are only put into sleep to speed up adaptations to future increases. Such mechanism is costly while processing steady streams as uctuations are rare. Processing uneven streams involve several modications of the knowledge base which lower its interest.

Infosphere

IBM Infosphere [Biem et al., 2010] is a distributed DSMS able to process static data and streams with the same processing engine. It is extensible as it allows to add domain specic operators implemented in a high level language like C++ in addition to predened operators oered by the declarative language SPADE [START_REF] Gedik | Spade: The system s declarative stream processing engine[END_REF].

Query denition: As System S, Infosphere relies on the declarative language SPADE for query denition. The declaration of continuous queries sits between a declarative expression like a CQL expression and an implementation in a high level language using stream-oriented APIs.

User-dened operators follow templates which allow or not data parallelism.

Query representation: SPADE expressions are turned into workows of PE as explained above. However, Infosphere classies PEs into 3 main categories:

• Source PEs are connected to stream generating services. Infosphere integrates several connectors to receive streams from GPS devices or databases.

• Sink PEs convert a stream into a relation stored in a le or database system. Generally, relations produced by a Sink are meant to be used by another system.

• All other PEs consume and generate streams or set of streams.

Window support: Infosphere supports time-based and tumbling windows exactly like System S as they both rely on the same query language.

Congestion management: Infosphere adapts continuous queries at operator level exclusively. Indeed, Infosphere supports operator parallelization and scheduling on user demand and automatically. Indeed, through SPADE primitives, parallelism degrees and assignment on processing units can be specied for each PE. In addition, a reinforcement learning-based algorithm allows Infosphere to analyze online properties of PEs as input and output volumes to collocate operators on same processing units in order to avoid network bottlenecks.

Target streams: As Infosphere turns continuous queries into graphs of PEs like System S, it ts also to even bounded streams but with the dierence that its initial conguration is dened by users.

Flink

Apache Flink [START_REF] Carbone | Apache ink: Stream and batch processing in a single engine[END_REF] is an open-source DSMS processing indierently static data and data streams as continuous sequences of stream elements running through fault-tolerant dataows.

Query denition: Flink provides several API for data manipulation and transformation.

These APIs include relational operators like Select-Project-Join or machine learning operators like k-mean clustering. Like Sonora, continuous queries are dened as sequences of user-dened operators. As some operators are user-dened, their semantics are hidden to Flink's optimizer.

Query representation: From the denition of a continuous query, Flink generates an optimized graph of stateless and stateful operators. In opposition to other workow systems, Flink may buer stream elements between two operators before emission. Indeed, as illustrated in [START_REF] Carbone | Apache ink: Stream and batch processing in a single engine[END_REF], two stateless operators processing stream elements in pipeline exchange stream element as soon as they are produced. These exchanges are performed concurrently and the throughput does not increase when the rate exceeds a certain threshold due to this concurrency. Buering stream elements and sending them into microbatches increase the throughput after this threshold.

Window support: Flink supports time-based and count-based sliding, tumbling and landmark windows. Windows are declared as options within stateful operators. Flink takes advantages of overlaps between consecutive iterations of a sliding window to compute incrementally results. This incremental processing model reduces signicantly computation latency of stateful operators over sliding windows.

Congestion management: Flink only supports adaptation of continuous queries at operator level. Flink performs resource-aware scheduling according to user specications about CPU and memory requirements for each operator like RStorm. As explained above, Flink relies on dynamic microbatching of stream elements between operators to improve throughput instead of shedding the load or parallelize operators at runtime.

Target streams: Flink does not integrate operator parallelization and load shedding at runtime. Nevertheless, Flink is able to maintain a near optimal throughput through dynamic microbatching. It makes it reliable for even erratic streams under the assumption that it exists an optimal size of microbatches which compensates an overow in input. Otherwise, Flink is appropriate for steady streams if the bandwidth supports microbatches. Target streams: C-MR aims even and uneven erratic streams as presented in [START_REF] Backman | C-mr: Continuously executing mapreduce workows on multi-core processors[END_REF].

RStorm

To absorb uctuations in input rate, C-MR relies on incremental computations over sliding windows as discussed above. According to this optimization, C-MR is able to process streams with short latency independently of input rate. The dynamic scheduling of operators allow C-MR to compensate critical overows in input rate.

iMR

In-situ MapReduce (iMR) [START_REF] Logothetis | In-situ mapreduce for log processing[END_REF] is designed to processing timestamped data without preliminary storing phase. It allows to process data with frequent updates (e.g., server logs) where they are produced.

Query denition: iMR extends the MapReduce framework with window denition like C-MR.

It allows to discretize timestamped data and apply MapReduce jobs to process stream elements in pipeline. The main dierence with C-MR is that iMR requires that data are stored on disk to replay them as a sequence of nite substreams. Nevertheless, an iMR application processes stream elements continuously and generates new results as soon as updates are stored.

Query representation: As C-MR, iMR represents a continuous as MapReduce jobs processing stream elements in pipeline. Nevertheless, iMR applications does not integrate Combine operators.

Window support: Time-based sliding and tumbling windows are supported by iMR. Like C-MR, iMR takes advantage of overlaps between consecutive iterations of a sliding window to process incrementally results. However, iMR does not require a Combine phase but discretize streams into panes [START_REF] Logothetis | In-situ mapreduce for log processing[END_REF]. Let consider a sliding window of size R and slide S, a pane is a subwindow of size R/S. Map operators compute results on panes instead of complete windows and results are updated incrementally during the Reduce phase.

Congestion management: iMR supports the adaptation of continuous queries at data level exclusively. iMR enables load shedding when the workload lead to a violation of QoS constraint about end-to-end latency. Indeed, the number of stream elements included in a pane requires a processing time greater than user specication, the pane is discarded in case of a violation of QoS constraints. A delity metric, denoted C 2 , reects the quality of results according to losses during computation.

Target streams: According to its load shedding policy, iMR processes even and uneven erratic streams as it is designed to handle any massive increase in input rate. But contrary to Sonora, iMR does not sample inputs to obtain a steady stream. Indeed, when iMR detects a critical workload, it discards entire logic substreams (i.e., elements included in a subwindow) until the global workload becomes acceptable.

Spark Streaming

Apache Spark Streaming [START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF]] is a stream processing engine built over Apache Spark 14 . It aims at bringing the eciency and simplicity of MapReduce paradigm to stream processing. Apache Spark outperforms other MapReduce implementations (Hadoop, Hive) because of its better memory management to perform most computation on main memory and avoid disk accesses.

Query denition: A continuous query is dened as a MapReduce job with window clauses. Target streams: Spark Streaming takes advantage of its threshold-based mechanism for congestion management to process even and uneven erratic streams. A signicant uctuation in input rate or value distribution may be compensated by reallocation of operators on processing units. The absence of reconguration traces like knowledge base shows that Spark Streaming is not specically optimized for bounded stream processing.

Samza

Apache Samza [Noghabi et al., 2017] is a distributed DSMS supporting stateful processing and recovering fast from failures. Samza relies on partitioned local states updated by a low-overhead mechanism. It allows states to scale to hundred of terabytes without increasing failover latency.

Query denition: As other MapReduce approaches, Samza relies on the key/value data model. To manipulate such data, it provides an API integrating three types of operators:

• One-to-one operators correspond to operators taking in input a stream and producing a single stream in output. One-to-one operators oered by Samza are map, lter, window and partition operators.

• Many-to-one operators take as input many streams in input and produce a single stream in output. Samza suggests join and merge operators for this type. The operator merge returns the union of all input streams.

• One-to-many operators correspond to partitioning operators and are user-dened.

With these operators, users can dene a continuous query as a sequence of operators subscribing and publishing some streams.

Query representation: Continuous queries are represented as Samza jobs. A job is a workow where vertices are one of the operator mentioned above.

Window support: Samza supports time-based tumbling windows through the operator window mentioned above. It takes as parameter a size and a function which could be user-dened.

Congestion management: Apache Samza integrates mechanisms to adapt continuous queries at operator level and can potentially adapt queries at data level. In addition to fast failover, Apache Samza supports operator parallelization and scheduling at runtime triggered by a thresholdbased monitoring module. Load balancing is performed by a user-dened partitioner.

Target streams: Like Spark Streaming, Samza adapts execution of operators according to a threshold-based algorithm so even and uneven erratic streams can be processed eciently. Indeed, Samza monitors continuously operators and adapts parallelism degrees and assignments on processing units accordingly.

S4

Yahoo Simple Scalable Streaming System (S4) [START_REF] Neumeyer | S4: Distributed stream computing platform[END_REF] is a general purpose stream processing system built for large clusters and supporting massively parallel applications. S4 oers a decentralized architecture where all nodes share same functions and responsibilities.

Query denition: S4 oers a simple programming interface based on the key/value data model. A continuous query is dened as a sequence of user-dened operators following a pattern.

A method processEvent species which keys are consumed by the operator and a method output describes keys produced the operator. These methods have stateful variations which require window specication.

Query representation: S4 represents continuous queries as workows of Processing Elements (PE) dierent from PE presented previously. In S4, a PE is the combination of a functionality which corresponds to the operator logic, the type of consumed events, the list of keyed attributes the operator consumes and optionally considered values of keyed attributes. A PE may also be associated to a Time-To-Live (TTL). If no stream elements enters a PE during its TTL, the PE is discarded.

Window support: Time-based sliding and tumbling are supported in S4 through parameters included in stateful operators. As iMR and C-MR, S4 takes advantage of sliding window denition to compute incrementally results.

Congestion management: S4 integrates automatic failover but does not integrate mechanisms to adapt continuous queries. Indeed, no mechanism can tackle the apparition of network and computation bottlenecks. Moreover, routes are dened statically. It is also assumed that key partitions dened by users do not involve major imbalance. Only the TTL may remove a PE receiving no stream elements.

Target streams: S4 is the less exible DSMS presented in this classication so it is assumed that input streams are steady and even. Even if the TTL allows the deactivation of idle PEs, it is supposed that the value distribution does not change often.

Dataow

Google Cloud Dataow15 , or just Dataow [START_REF] Akidau | The dataow model: A practical approach to balancing correctness, latency, and cost in massive-scale, unbounded, out-of-order data processing[END_REF], is MapReduce-based DSMS designed natively for stream processing. Dataow supports unordered data streams and provides multiple windowing schema. It results of the extension of FlumeJava API [START_REF] Chambers | Flumejava: easy, ecient data-parallel pipelines[END_REF] developed over Google Millwheel. FlumeJava is an API for the development of pipelines of MapReduce jobs. It facilitates the creation, the execution and the modication of such pipelines through a simple programming interface providing high-level operators mapped automatically on optimized MapReduce jobs.

Query denition: Dataow provides two primitive operators:

• ParDo is the generic parallel function. It receives in input a stream element or a collection of stream elements and apply a function DoFn on input elements. DoFn is dened either by users in a high level imperative language like Java either with a high level operator provide by FlumeJava API. For each stream element, it returns zero, one or many outputs.

• GroupByKey is a key grouping operator applying a function on collections of stream elements sharing a same key. Congestion management: Dataow only supports the adaptation of continuous queries at implementation level. Dataow performs implementation selection at runtime through Flume-Java. Indeed, while using predened operators, FlumeJava may have multiple implementations and select the most appropriate one according to execution features like input size.

Users dene continuous queries through

Target streams: The absence of load shedding and adaptation of operators limit the ability of Dataow to process even and uneven erratic streams. The algorithm selection is limited to compensate a substantial variation in input rate and value distribution for all operators. After a formalization of basic concepts about data streams, we introduced the problem of congestion. In a real-time context, resources may not be adapted for computations during the complete lifetime of a query. It requires to elastically adapt resources (i.e., infrastructure) and continuous queries to processing requirements over time. To adapt the global capacity of the cluster, we presented solutions for substituting or adding processing units at runtime. Substituting processing units allows to benet from more resources on a single processing unit but has many inconveniences. It requires to overprovision the execution support or to allow important migration overheads which is not acceptable for low-latency applications processing streams continuously.

Adding processing units at runtime is supported in [START_REF] Satzger | Esc: Towards an elastic stream computing platform for the cloud[END_REF] but is limited by the number of available machines and ts exclusively to cloud environments. Elasticity at logical layer can be applied independently of the execution support. Main patterns presented in section 1 allow removing congestion due to computation and network bottleneck. Moreover, they can improve signicantly performance of DSMSs (e.g., operator parallelization) when used at runtime. The triggering of such elastic techniques at runtime require a continuous monitoring of submitted queries coupled to a detection mechanism. This mechanism relies either on external intervention (users or tier application) or on an automatic method. Relying on external intervention limits the elasticity of DSMSs to user reactivity and expertise. Automatic solutions based on queuing theory are limited by the static model used to predict the behavior of DSMSs. Indeed, clusters executing continuous queries evolve over time (e.g., node failure [Yang et al., 2012]) so a static model is not appropriate in a stream processing context. Threshold-based approaches [START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF], Chandrasekaran et al., 2003, Noghabi et al., 2017] are exclusively reactive as they only trigger reconguration according to the current state of the system. Control-based [Abadi et al., 2005] and reinforcement learningbased approaches [START_REF] Amini | Spc: A distributed, scalable platform for data mining[END_REF], Biem et al., 2010] can anticipate congestion of operators through a continuous adaptation of workows to workload. Nevertheless, nding an optimal conguration may require an important number of interactions with the DSMS and induce reconguration overheads. We presented a wide variety of DSMSs and suggest a classication to highlight dierences in terms of target streams and congestion management between categories of DSMSs. It shows that dierent optimization techniques allow DSMSs to process streams elastically. For example, let consider a continuous query Q 1 consuming an uneven erratic stream S 1 . Adapting Q 1 consists in avoiding processing and network bottlenecks due to overload in input of operators while critical increases in input rate happen. Triggering operator parallelization and scheduling is an appropriate solution in that case. Moreover, as the distribution of values has an impact on processing latency too, adapting computations to uctuations in distribution of values consists in balancing the load to avoid processing bottlenecks due to imbalance between equivalent tasks. Nevertheless, there may be no satisfying solution based on load balancing if the parallelism degree is undersized according to the input rate. It shows that the eciency of a conguration may not only depend on a single pattern but a combination of patterns.

According to the classication of DSMSs suggested in the previous chapter, it appears, to our knowledge, that no DSMS can handle eciently any type of streams. It highlight the diculty to extract features of DSMSs which bring a benet without limiting global elasticity or degrading performance.

In this section, we focus on dierent adaptation levels enabling elastic stream processing through the identication of their roles, which triggers are used to detect reconguration needs and relations between these levels. It facilitates the identication of key aspects for elastic stream processing and their impacts on execution properties (i.e., processing latency and result quality). Finally, we present common strategies used in literature and discuss their eciency while facing dierent stream types. [START_REF] Arasu | The cql continuous query language : Semantic foundations and query execution[END_REF] The ORACL loop

Steps for query optimization

In the previous chapter, we expose some patterns commonly used to optimize the treatment of continuous queries presented in chapter 3 Table 1 (see page 26), a pattern may modify a continuous query at dierent scopes: the entire workow, a single operator or data lifecycle. Nevertheless, modications performed at each level may have an impact on other ones. In this chapter, we aim at specifying the dierent steps for query optimization. We present and detail their roles and the relation order between them. Indeed, optimizing a continuous query requires to follow a specic process composed of two main steps: the logical and the physical steps. The logical step aims at dening an optimal workow according to input streams and composed of operators potentially parallelized. The physical step aims at dening optimal assignment and customization of tasks on processing units. By customization, we mean the choice of the implementation and the selection of input data. For both logical and physical steps, we identify two substeps: the inter and the intra-operator optimization. It denes an optimization process composed of 4 steps performed in the following order:

• The logical inter-operator step aims at dening a near-optimal workow corresponding to the submitted continuous query and according to stream uctuations in terms of distribution of values.

• The logical intra-operator aims at dening appropriate parallelism degrees (i.e., number of tasks) for each operator belonging to the workow according to stream uctuations in input rate.

• The physical inter-operator step looks for an optimal assignment plan of all tasks composing a workow on available processing units. Criteria dening an optimal assignment plan depends on user objectives in terms of resource usage.

• The physical inter-operator step denes optimal implementations for tasks according to input stream properties (e.g., distribution of values) and manages data lifecycle.

Adaptation levels

Through the identication of these optimization steps, we suggest an abstract framework which formalizes the function of each step and existing dependencies between them according to query and stream characteristics.

As illustrated on Figure 1, these levels dene an adaptation loop, denoted ORACL loop, which is composed of steps Order (logical level and inter-operator step), Replicate (logical level and intra-operator step), Assign (physical level and inter-operator step) and Custom Locally (physical level and intra-operator step). Adaptation steps are revised according to a top-down model. For example, a revision at Replicate involves revisions of Assign and Custom Locally steps but not of Order step.

• The Order step performs an algebraic optimization of continuous queries. Let consider a continuous query Q applying a global function f on inputs to generate outputs. An optimal workow W of Q, from logical point of view, respects the following properties: • The Replicate step adapts parallelism degrees of operators in order to maintain processing rates greater or equal to input rate. Let consider an operator O i , with an input rate r i and an average processing latency lat i . As lat i can be dened in second per stream element, the inverse value 1 lat i corresponds to the processing rate pr i of O i . Maintaining pr i greater than r i is the major challenge of operator parallelization. Actually, if k tasks associated to an operator process stream elements in parallel, they virtually multiply pr i by k. Nevertheless, partitioning and transmission overheads limit the benet brought by parallelization. In addition, depending on the number of available processing units, increasing the parallelism degree of an operator over a certain threshold may create concurrency between tasks.

• The Assign step revises assignments of tasks on processing units, denoted scheduling plan, in order to adjust resource usage to processing requirements. As mentioned above, the denition of an optimal scheduling plan is guided by a targeted usage of resources. As most continuous queries are composed of heterogeneous operators in terms of processing requirements (e.g., stateless lters and stateful joins), usage of resources on each machine depends on the subset of operators assigned on it. An optimal scheduling plan should ensure that each operator benets of enough resources such as its processing rate is not limited by resources.

• Finally, the Custom Locally step aims at taking advantage of local implementations of operators in order to select an algorithm adapted to execution context. Indeed, the computational complexity of equivalent implementations of an operator may dier in term of time (e.g., nested loop join or hash join). Selecting the most appropriate implementation may increase signicantly the processing rate of a given operator.

Optimization strategies

As presented above, each step of the ORACL loop aims at producing a output which has specic properties. However, if the behavior of each step can be dened in a generic way, it exists several strategies to produce targeted output. The choice of the strategy depends on user objectives in terms of performance and resource usage but also on technical constraints (e.g., query language used to dene continuous queries). We provide a brief overview of common strategies used to implement each step and detail their impacts on outputs. The Order step considers continuous queries dened by users. As presented in chapter 2 section 2.1, continuous queries can be declarative expressions (e.g., CQL expressions), se-quences of operators declared in a high level language [Peng et al., 2015[START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF], Yang et al., 2012, Akidau et al., 2013] or graphs of connected operators dened through a graphical interface [Abadi et al., 2003, Abadi et al., 2005]. This step is only relevant for DSMSs relying on a declarative language. Moreover, a cost function is necessary to evaluate each possible workow associated to a continuous query.

Improvement through Order step

The generation of an optimized workow is totally dependent of the query language and possibilities of algebraic optimization. While using a declarative language, some approaches [Arasu et al., 2004, Chandrasekaran et al., 2003] use algebraic properties like traditional DBMS [Garcia-Molina, 2008]. Nevertheless, an optimal workow cannot be dened for an entire stream as its properties cannot be anticipate at the initialization of the continuous query. Only some choices will not be contradicted at runtime (e.g., projection of attributes before selective operators if there are commutative), other permutations of operators are performed at runtime according to operator properties (e.g., selectivity factor). Approaches using graphical interfaces for query definition [Abadi et al., 2003, Abadi et al., 2005] may use formal semantics of operators in order to permute safely operators. Approaches relying on imperative languages [Peng et al., 2015, Akidau et al., 2013, Noghabi et al., 2017, Akidau et al., 2015] cannot benet of such optimization as there is no explicit semantic nor algebraic properties (e.g., commutative operator) linked to operators. The parallelization of operators can be performed on user demand or automatically. Parallelization of operators on user demand [START_REF] Xu | Stela: Enabling stream processing systems to scale-in and scale-out on-demand[END_REF] relies on user expertise. Automatic parallelization of operators is triggered automatically at runtime to compensate a signicant gap between input and processing rate. Some approaches [START_REF] Schneider | Elastic scaling of data parallel operators in stream processing[END_REF], Gedik et al., 2014] rely on trial-and-error algorithms to explore parallelism degrees for each operator independently of all others and associate a rank of input rates managed eciently.

Improvement through Replicate step

A major risk of operator parallelization is that the frequency of recongurations becomes too important. Indeed, creating a thread on the y and balance partitions between new tasks involve important overheads. To limit this negative eect, authors [START_REF] Gedik | Spade: The system s declarative stream processing engine[END_REF], Biem et al., 2010] suggest to create threads once and instead of deleting them when they are not necessary, threads are only put into sleep.When additional working threads are necessary, the DSMS just wake up some sleeping threads. Another major inconvenient common to most solutions is the parallelization of operator independently of the eect on downstream operators. Indeed, modifying the parallelism degree of an operator may have an impact on its processing rate and throughput. Missing this relation between operators xed during Order step may cause instability due to contradictory reconguration.

Once parallelism degrees have been set for each operator, it is necessary to dene a strategy for load balancing between tasks. As presented in chapter 3 section 1, maintaining a balance between tasks of an operator is crucial to benet completely from data parallelism. For example, let consider an operator O i divided into tasks T 1 to T n . During a period of time ∆, each task T j receives in input a partition P j of global inputs P which may dier from other partitions in term of volume and distribution of values. According to this and as all tasks apply the function corresponding to O i , overall processing times may be dierent for each task. So, the global processing time of P by O i is at least the greatest of all processing times of its tasks. In this context, it is crucial to minimize imbalance in term of processing time between tasks.

To do so, we distinguish three categories of load balancing strategies:

• Round-Robin strategies [Abadi et al., 2003, Abadi et al., 2005, Chandrasekaran et al., 2003, Peng et al., 2015] distribute fair numbers of stream elements between tasks. This strategy is ecient to balance load under two assumptions. First, the input stream is even so partitions of same size involve theoretically same processing times. Secondly, all tasks has same available resources (CPU, RAM) to process stream elements.

• Key-based strategies [START_REF] Neumeyer | S4: Distributed stream computing platform[END_REF], Akidau et al., 2013, Akidau et al., 2015] associate a key set to each task and route stream elements accordingly. It assumes that input streams are even and key distribution is uniform over time. To remove the uniformity constraint, authors in [START_REF] Rivetti | Ecient key grouping for near-optimal load balancing in stream processing systems[END_REF] suggest an approach which builds balanced key groups associated to each task according to key distribution at runtime. This approach serves as guarantee that key grouping maintains balanced partitions in term of volume over time.

• Load-aware strategies aim at balancing processing times between tasks. In [START_REF] Rivetti | Online scheduling for shue grouping in distributed stream processing systems research paper[END_REF],

a load balancing strategy based on sketches associates average processing times to encountered keys in order to route dynamically stream elements on tasks able to process them with shortest delays. This solution has the advantage to be applicable on both even and uneven streams with a reduced footprint on memory.

Improvement through Assign step

The denition of the scheduling plan is guided by an objective function describing an optimized usage of resources. We distinguish three main objective functions for operator scheduling: • The rst objective function f equality can be denoted as equality-aware. They aim at balancing processing requirements evenly between all processing units. More formally, let consider n processing units M 1 , M 2 ,..., M n described by resource usages U 1 , U 2 ,..., U n . A optimal scheduling plan assigns operators such as:

f equality = argmin   1 n n i=1 U 2 i -Ū2   ( 1
)
where Ū is the mean value of all resource usages. It corresponds to minimize standard deviation of resource usage among processing units. Spreading computations uniformly over processing units has the advantage to spread the impact of an overload over all processing units. Commonly, this strategy is implemented through a Round-Robin distribution [Xu et al., 2014[START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF] of operators on slots or more generally processing units. Nevertheless, it assumes that all operators require same resources which may be wrong in real-world applications.

• Trac-aware objective function f traf f ic aims at minimizing the global network trac. As presented in [Xu et al., 2014], transmitting important volumes of data over network has a signicant impact on global processing latency. Indeed, considering a parallelized workow, edges dene transmissions of stream elements between operators, denoted inner-streams. These inner streams can be supported through shared memory or network involving serialization/deserialization and transmission overheads. Assigning operators on processing units such heaviest inner streams, in term of volume, are supported by shared memory improves signicantly overall latency and throughput. Formally, considering inner streams S 1 , S 2 ,..., S n belonging to a parallelized workow, an optimal scheduling plan is dened as follow:

f traf f ic = argmin n i=1 |S i | (2)
where |S i | is the volume of stream elements transmitted through the inner stream S i .

A side-eect of this strategy is to concentrate consecutive operators on same processing units. In comparison to equality-aware strategy, it limits signicantly the risk of network bottlenecks as heaviest inner-streams are supported by shared memory but it increases the risk of processing bottlenecks as operators manipulating greatest volumes of data are assigned on same resource.

• Resource-aware objective function f resource aims at minimizing combined usage of all resources (i.e., CPU, memory and bandwidth). This approach extends the trac-aware strategy through the consideration of CPU and memory constraints. In [Peng et al., 2015], authors suggest to consider static CPU and memory requirements for each operator and to assign operators such as network is minimized and each operator meets its resource requirements. It corresponds to a multi-dimensional multi-choice knapsack problem which is solvable through a heuristic algorithm with a polynomial complexity. The main dierence with f traf f ic is the consideration of explicit resource requirements which bound the minimal network trac. The choice of the algorithm may have a signicant impact on processing latency. The example of join algorithms is the most representative as equivalent implementations have dierent time complexity. Borealis [Abadi et al., 2005] takes advantage of its algebra to provide dierent implementations of its predened operators. Each processing unit is provisioned with all implementations and operators select at runtime which one ts the best to execution context.

Improvement through Custom Locally step

In [START_REF] Welsh | Seda: an architecture for wellconditioned, scalable internet services[END_REF], authors give a dierent semantic to algorithm selection as it corresponds to a controlled load shedding. Indeed, in case of overload, operators managed by SEDA can apply a degraded service. It corresponds to a implementation of the service integrating a sampling mechanism to reduce the workload. An other load shedding strategy has been introduced in [START_REF] Babcock | Load shedding for aggregation queries over data streams[END_REF] and developed in [START_REF] Tatbul | Staying t: Ecient load shedding techniques for distributed stream processing[END_REF]. It relies on the introduction of shedders in workows composed of aggregation operators (e.g., sum or count). The aimed shedding ratio of the workow, i.e. the ratio between the input rate before and after shedding, is used to compute where shedders with their respective shedding ratios should be placed such as the relative error due to shedding does not exceed a maximal threshold.

Orchestration of optimization 2.1 Adaptation triggers

To enable dynamic adaptation of continuous queries, a DSMS needs a monitoring module which observes continuously relevant metrics and optionally a set of parameters which discriminate normal execution of continuous queries. We describe here metrics observed at each level and optional parameters used to trigger each step of the ORACL loop.

Order step

As mentioned above, the Order step aims at optimizing continuous queries according to algebraic properties of operators. It corresponds to dynamic operator reordering (see 3 section 1). Reordering is based on evolution of selectivity factors. For reminder, considering two commutative operators O 1 and O 2 , if O 2 is more selective than O 1 and is executed downstream, permuting O 1 and O 2 reduces the volume of data exchanged between these operators without modifying the global semantic of the query. In addition to selectivity factors and volumes of data, DSMSs need algebraic properties of all operators composing the workow.

Replicate step

Detecting a need of operator parallelization relies on a continuous monitoring of input and processing rates. As explained above, modifying the number of tasks associated to an operator has an impact on its global processing rate. Adjusting parallelism degrees allows operators to process stream elements as soon as they arrive and prevent processing bottlenecks due to an accumulation of pending stream elements. In some approaches [START_REF] Schneider | Elastic scaling of data parallel operators in stream processing[END_REF], Gedik et al., 2014], throughput is observed instead of processing rate. It relies on the assumption that selectivity factor does not vary signicantly so for a given input rate, a maximal throughput could be identied and remain valid for the entire stream.

Assign step

Two event types can trigger reassignment of operators at runtime:

• Overload and underload of processing units. While a processing unit is overloaded according to some resources (CPU and RAM), it is necessary to move some operators assigned to this processing unit on others which have enough available resources. An overload means that usage of some resources have exceeded predened thresholds. This detection could be done according to a resource monitoring at dierent scopes. Some DSMS like Apache Storm or Apache Flink allow users to instantiate multiple processing units on a single machines which have declared resources. Depending on the resource sharing policy between processing units (i.e., if resources are exclusive to a processing unit or not), monitoring of resources is relevant only at machine or processing unit scope.

• Network trac. The appearance of network bottlenecks requires to revise the scheduling plan [Xu et al., 2014] in order to prevent an important degradation of the overall latency. Such bottlenecks are detected through an observation of inner-stream rates coupled with the current scheduling plan. For each inner-stream supported by network interface, if the rate is limited by the bandwidth, assignments of associated operators should be revised. More generally, authors show in [Xu et al., 2014] the interest to keep dynamically heaviest inner-streams on same machines.

Custom Locally step

Like the Order step, the Custom Locally step is only relevant for DSMSs based on a declarative language. Indeed, changing the local implementation of an operator depends on both inputs and implementation properties. For example, considering a common operator like join, two frequent implementations are nested-loop join and hash-join. Depending on the ratio between cardinalities of driving and probed relations, nested-loop implementation may be faster or slower than hash-join implementation. So, while changing the algorithm, specic metrics may be required according to implementation properties.

If implementations associated to an operator only suggest more or less degraded versions of the same algorithm (i.e., versions integrating a sampling or load shedding policy), selecting an other implementation at runtime relies on the ratio between input and processing rate. It is closed from the triggering condition for operator parallelization but it is commonly used when operator parallelization is not applicable (e.g., maximal usage of resources or absence of parallelization support at runtime).

Challenges and dependencies between adaptation levels

As presented on Figure 1, there is an execution order between steps . Considering an adaptation level, it results that choices performed at previous step have an impact on optimization possibilities. We present here dependencies be adaptation levels through a generic description of each intermediate output.

In output of the Order step, an optimal workow is dened. It xes data transmission between operators which has an impact on data volumes transferred during processing. Depending on these transmissions, more or less important volumes must be exchanged to execute the query dened by a user. Generating a non-optimal workow in output of this step may cause overloads on inner-streams which must be managed during next steps despite they could be avoided through algebraic optimization.

This optimized workow is used by the Replicate step to generate an optimal and parallelized workow. It is composed all operators to assign on processing units. It xes the parallelism degree for each operator which bounds theoretical processing rates. In addition, for parallel areas, the load balancing has been dened and should compensate uctuations in distribution of values while processing uneven streams. So, choices made during this phase dene which uctuations in input rate and distribution of values can be handled without reconguration. According to the current input rate, underestimating parallelism degrees for some operators may lead to processing bottlenecks which degrade the global latency. The single solution to manage an overload at xed parallelism degree is load shedding. So, it requires to degrade result quality to maintain acceptable latency according to user constraints. At the opposite, overestimating parallelism degrees add more operators to schedule without benet in terms of performance. Moreover, it degrades global latency because of partitioning and routing overheads.

According to a set of available processing units, a scheduling plan assigning each operator of the parallelized workow on processing units is generated. It denes the set of active processing units and the network trac between them. In case of eective overload (e.g., CPU overload) or deviation from the optimal scheduling plan, the Assign step is revised to nd to avoid node failure.

In the case of heterogeneous processing units in terms of available resources (e.g., processing units shared by multiple continuous queries) or available implementations of operators, assignments restricts opportunities of local customization.

Finally, when all operators are assigned, the Custom Locally step associated them to implementations. Stream elements can be routed to operators and computed according to selected implementations.

To sum up, the management of uctuations in input rate and value distribution can be done at dierent steps. Nevertheless, according to the order on adaptation levels, optimizing the execution of a continuous query early oers favors performance and result quality.

Discussion

The ORACL loop gathers key aspects for elastic stream processing and order them into steps. It highlights possibilities of elastic optimization while processing streams and points impacts of each step on global execution. For example, the Replicate step sets parallelism degrees of operators and load balancing strategy between tasks. It has a denes average processing rates of operators which have a direct impact on maximal input rate operators are able to absorb. Moreover, it shows that the global performance of a DSMS is the result of a sequence of optimization decisions. To obtain a full elastic stream processing, this sequence should be revised dynamically to maintain a near-optimal conguration according to uctuations in input rate and distribution of values.

Nevertheless, the ORACL loop is not fully applicable to any DSMS to enable a full elastic stream processing. For example, DSMS based on imperative denition languages cannot perform the Order step as operators are user-dened and do not have explicit algebraic properties (e.g., commutativity). In consequence, such DSMS rely on user expertise for the denition of optimal workows. Moreover, algorithm selection cannot be performed during Custom step due to the absence of dierent implementations. It limits signicantly optimization at query and operator scopes.

Concerning Assign step, several works [Aniello et al., 2013, Xu et al., 2014, Peng et al., 2015] suggested solutions using heuristic algorithm to obtain near-optimal scheduling plans according to an objective function as discussed above. The denition of a scheduling plan is mainly ofine [Aniello et al., 2013, Peng et al., 2015] and takes as inputs a set of operators, a set of available processing units and optional constraints specied by users. Then, this initial scheduling plan may be revised dynamically through migrations of operators at runtime to remove machine overloads when necessary [Xu et al., 2014].

It appears that Replicate step lacks strategies able to recongure eciently parallelism degrees of operators at runtime. In [START_REF] Xu | Stela: Enabling stream processing systems to scale-in and scale-out on-demand[END_REF], a solution named Stela is able to identify critical operators in a workow and according to available resources, adjust parallelism degrees in order to maximize global throughput. Nevertheless, this solution relies completely on user expertise and reactivity to add or delete resources when necessary. In several cases, end users cannot have such control on execution (e.g., execution of continuous queries on the cloud through a service provider). In [START_REF] Schneider | Elastic scaling of data parallel operators in stream processing[END_REF], Gedik et al., 2014], authors suggest solutions enabling automatic parallelization of operators at runtime over System S [START_REF] Gedik | Spade: The system s declarative stream processing engine[END_REF]. Their approaches relies on processing elements (see 3 section 3) to modify parallelism degree of each operator independently of all others. The parallelization strategy relies on reinforcement learning based only on input volumes to build a specic knowledge base for each processing element without consideration for modication performed upstream. So, the global convergence time may be innite in presence of erratic streams. Moreover, as parallelism degrees are exclusively associated to ranges of input rates, uneven streams may extend considerably the duration of the learning phase. Finally, all solutions for automatic parallelization estimate appropriate parallelism degrees according to a stable input stream, i.e., steady streams like GPS signals or temperature sensors in a smart building context. It means that in the case of an input stream increasing quickly, reconguration of parallelism degrees will be frequent and involve important overheads. Identifying evolution trend of input stream may avoid these overheads in most cases. Moreover, adapting the parallelism degree of an operator without taking upstream operators into account degrades the consistency and the stability of the DSMS. For example, while observing a signicant increase of input rate upstream a costly operator (e.g., join) it is interesting to adapt its parallelism degree in a proactive way so the impact of the overload is soften by the quick adaptation.

Concerning load balancing, load-aware strategy presented in [START_REF] Rivetti | Online scheduling for shue grouping in distributed stream processing systems research paper[END_REF] builds balanced stream partitions between tasks based on eective processing latency. So, it is independent of stream type according to distribution of values and application (i.e., even or uneven).

5

Preventive auto-parallelization approach for elastic stream processing Through the presentation of the ORACL loop and bibliography, we highlighted the need of automatic, reactive and accurate strategies for operator parallelization. Once on user demand approaches require user presence and expertise, the eciency automatic approaches based on reinforcement learning are limited by stream properties in terms of uctuations in input rate and distribution of values. To tackle this issue, we suggest an auto-parallelization strategy, named autoscale, which prevents operator congestion and limit degradation of result quality. This approach relies on an automatic and dynamic adaptation of resource consumption for each continuous query. This solution takes advantage of i) a metric estimating the activity level of operators in the near future ii) the autoscale approach which evaluates the need to modify parallelism degrees at operator and workow scope.

Execution context 1.Assumptions

According to the execution context presented in chapter 2 section 2.2, we assume that there are enough available resources to process all queries (H1).

We consider that the DSMS manages state migration when some tasks of operators are added or deleted (H2). Otherwise, it requires to manage states [START_REF] Hirzel | A catalog of stream processing optimizations[END_REF] as they allow the system to distinguish stream elements waiting in pending queues from ones being processed and waiting for the completion a current computation window. This management is out of the scope of this chapter.

Concerning the execution of each query, we consider that all operators can be processed in parallel by multiple tasks and scheduled potentially on dierent machines. Nevertheless, the global incoming load is divided evenly between tasks applying a same operator (H3).

These tasks are assigned on machines according to a scheduling strategy. We assume that the scheduling strategy revises periodically assignments of operators. Moreover, this strategy assigns at most one task for each operator on a given processing unit (H4) so there is no concurrency on resource usage (CPU,RAM) between two tasks of a same operator.

Finally, the auto-parallelization strategy is integrated to a system processing streams elastically according to the ORACL loop (see chapter 4 section 1). So, modications of operator ordering and scheduling are performed respectively before and after any modication of parallelism degree (H5).

Challenges

Each task applies a function dened by a user on each stream element in input. Depending on the time complexity of its function and available CPU, an operator can process, in average, a certain number of items per time unit. This number is called the capacity of the task. This capacity limits the input rate an operator can handle. According to his, a congestion may happen when the input rate is greater than the capacity. In order to limit the impact of a critical input rate, a solution consists in modifying the parallelism degree of the operator in order to distribute the incoming load between more tasks.

In order to prevent congestion, a DSMS should be able to detect when the rate in entry of an operator reaches or exceeds its capacity. Indeed, a detection based on resource consumption (CPU or memory) only allow to remedy an eective congestion. It is not satisfying because the quality of treatments is deteriorating before the system recongures itself. Even if no stream element is lost, the overall latency suers from the congestion of one or many operators.

Yet, it is not easy to decide judiciously when increasing (scale-out) or decreasing (scale-in) the parallelism degree of an operator. Actually, given a task, if its input rate exceeds its capacity, the associated operator tends to congestion. But the congestion is eective only if the input rate remains equal or higher than the capacity for a signicant time. Otherwise, a scale-out is triggered too early lead to the creation and assignment of one or many unnecessary tasks. Thus, it degrades the stability of the system and generates important reconguration overheads. It is crucial that a relevant parallelization strategy takes the stability of the system into account. Moreover, it is important that this strategy reduces the parallelism degree of underused operators. It ts global capacity to processing needs and, depending on the scheduling strategy, it allows to free unnecessary processing units which become available for other queries.

To sum up, the automatic and dynamic adaptation of capacities of operators requires that a DSMS is able to detect potential congestion before it becomes eective in most cases. Moreover, a relevant strategy should not overreact to sudden peaks of input rate. It degrades the stability and the performance of the DSMS because of reconguration overheads. Finally, the system should t capacities of operators to their processing needs in order to consume only necessary resources. The objective of an auto-parallelization strategy is then to detect when a reconguration is needed and to adjust accurately the parallelism degree. Finding a satisfying compromise between these issues is a major challenge for elastic stream processing. Indeed, with the growing popularity of pay-as-you-consume solutions (Amazon EC2, Microsoft Azure...) and the emergence of Green IT, it is crucial for the current generation of DSMS to take elasticity of treatments into account.

Overview of the autoscale approach

From these observations, let consider streams evolving progressively on a time interval ∆. Extracting knowledge from the recent history of stream uctuations through time-series analysis allows to estimate appropriate parallelism degrees of operators with an accuracy depending on the regression model (see chapter 3 section 2.2). Indeed, time-series analysis presents the advantage to forecast future states of a system without an a priori knowledge so parallelism degrees of operators can be adapted in constant time independently of stream rates encountered previously.

In the remainder of this chapter, we suggest an auto-parallelization strategy, named autoscale based on time-series analysis on recent history of operators as presented in section2.

These histories are built by a monitoring module which observes metrics describing the execution of each operator. A derived metric discriminating potential processing bottlenecks is computed from monitored ones. The computation of this metric and the interpretation of its possible values are detailed in section 3. Once a suggestion of reconguration is available for each operator, an algorithm checks the consistency of each suggested reconguration at workow scope. Checking the consistency at workow scope consists in identifying the set of necessary recongurations according to processing requirements. It aims at improving the stability of the system. Steps of this algorithm are presented in section 4 and we motivate its choice through highlighting examples. Then, we detail, in section 5 how we compute parallelism degrees according to globally consistent estimations of workloads. Finally we discuss limits and advantages of our approach in section 6 and expose an empirical study justifying the selected strategy for consistency checking.

Monitoring management

To identify potential congestion of operators through time-series analysis, we need to build recent histories of operators. These histories are composed of metrics describing execution states of operators. In this section, we suggest an attribute/value model to represent operators. Values are gathered over sliding monitoring windows presented below.

Operator model

We consider as operators, in the monitoring model, logical operators belonging to the optimized workow (see chapter 4 section 1). Thereafter, we describe operators at any timestamp t with three metrics: input load, pending queue size and average processing latency.

• The input load corresponds to the number of received stream elements.

• Pending queue size matches the number of stream elements pending in operator queue as illustrated on Figure 1.

• The average processing latency is the average processing time per stream element without distinction for each value appearing in the stream.

Figure 1: Operator model

For each metric, we consider the recent history. As logical operators may be applied by many tasks, each value belonging to a history is the aggregation of measured values for all tasks. For example, let consider an operator O i applied by tasks T 1 i and T 2 i , the number of stream elements in input of O i at a timestamp t 0 is the sum of input stream elements of T 1 i and T 2 i . We also consider the sum to aggregate pending queue size. Nevertheless for the average processing latency, we select the maximal value between average processing latency of T 1 i and T 2 i . So, we consider that the operator O i is limited by its slowest task, i.e. the task with the greatest processing latency.

Formalization

Let T = (O, V) be the topology of a continuous query represented as a direct acyclic graph where O is the set of operators and V the set of streams.

F i = {(F i j )} j∈N + . Each window F i is associated
with the operator O i , as illustrated on Figure 2, and is composed of iterations F i j . Each F i j is dened by a duration ∆ and groups measurements collected during this interval. These measurements are done according to a predened set of timestamps M i = {m i 1 , m i 2 , ..., m i n } n∈N + . For each operator O i , we collect measurements taking into account items received and processed in the interval [m i k-1 , m i k [ with k=1,...,n. It is worth noting that a master process (e.g. Nimbus for Storm or JobTracker for Hadoop) serves as guarantee that measurements are collected synchronously on each processing unit.

In order to compute relevant metrics from these monitored values, some constraints must be considered:

• The interval [m i k-1 , m i k
[ should be greater than the time required to pre-process and store measurements in a standard database management system, which is around a second. It reduces redundant measurements and massive monitoring overheads.

• The duration ∆ dening the size of the monitoring should be greater than all processing window sizes of stateful operators belonging to T . It ensures that all metrics presented in the remainder of this section can be computed for both stateless and stateful operators. Indeed, if a stateful operator computes results during a duration greater than ∆, it is impossible to analyze multiple measurements within a single window.

• A grace period must be considered after each reconguration triggered by autoscale.

Variations in input rates due to system stabilization are not considered during this period of time. In the remainder of this chapter, we consider a grace period of ∆ after each reconguration.

Let R i be the set, potentially innite, of stream elements received by operator O i . We consider R i,j as the subset of stream elements received by O i during the iteration F i j , and R i k the subset of elements received between [m i k-1 , m i k [. In the example presented on Figure 3, R i,j is the sum of measurements R i 0 to R i 4 , and R i,j+1 the sum of measurements R i 1 to R i 5 . In addition to the number of stream elements received, we collect the processing latency per stream element of the operator observed during F i j , denoted Lat F i j . This does not include the time an item may spend in pending queues.

Detection of reconguration needs

Now, that we have detailed how each operator belonging to a workow is monitored, we focus on how reconguration needs and opportunities are detected from observed metrics. An autoparallelization strategy prevents congestion without degradation in result quality only if it can anticipate appearance of processing bottlenecks. According to this requirement, it is necessary to estimate input load and capacity in near future for each operator. On the contrary, if the input load decreases signicantly, the capacity should be decreased accordingly to avoid waste of resources. In the remainder, we detail how autoscale detects reconguration needs through estimations of input load and capacity in near future.

Estimation of input load in near future

Let consider an operator O i observed at the start of an iteration F i j of the monitoring window F i . The global workload of O i during F i j , denoted Input i j corresponds to the number of stream elements received during F i j added to the total number of stream elements pending in input queues of tasks applying O i at the end of the previous iteration F i j-1 of the monitoring window. This workload is expressed in formula [START_REF] Arasu | The cql continuous query language : Semantic foundations and query execution[END_REF].

Input i j = |R i,j | + pending F i j-1 (1)
As illustrated on gure 3, the eective number of received elements |R i,j | cannot be computed exactly before the end of F i j . Nevertheless, to anticipate an eventual congestion of O i during F i j , it is possible to approximate |R i,j | at the end of F i j-1 .

Figure 3: Recent history F i j-1 at the start of iteration F i j

The estimation of |R i,j | can be performed through a linear regression model based on measurements collected during F i j-1 . Let f i j-1 be the ane function computed by linear regression as illustrated on gure 4. For each timestamp m i k corresponding to a future measurement in F i j , f i j-1 is applied to estimate the number of received stream elements R i k . So, the total number of stream elements received during F i j is estimated as in formula ( 2) by |EstimR i,j |:

|EstimR i,j | = m i k ∈M i f i j-1 (m i k ) (2)
It is worth noting that some regression models have been tested without improving signicantly the accuracy of estimations.

With the knowledge of |EstimR i,j |, it is possible to estimate the expected workload during F i j at the end of F i j-1 according to formula (3).

EstimInput F i j = |EstimR i,j | + pending F i j-1 (3)
According to formulas ( 1) and ( 2), the workload of each operator can be estimated with a maximal anticipation ∆, where ∆ is the size of each iteration of the monitoring window. The estimation of the workload is updated at each end of an iteration which corresponds to the acquisition of a new measurement. It allows to keep estimations close from eective workloads. To optimize the computation of the estimated workload, sums used for regression are updated incrementally instead of recalculating aggregates common to overlapping iterations of the monitoring window.

Estimation of processing capacity in near future

Now that an estimation of future workload has been computed for each operator, it is necessary to estimate the capacity in near future for each of them. Thus, it will be possible to determine if the capacity of an operator ts the expected workload during next iteration F i j . To do so, we aim at estimating the number of stream elements each operator should be able to process during F i j . The processing capacity of an operator O i depends on the average number of stream elements O i has been able to process during the previous iteration F i j-1 . This number can be computed from the average latency per stream element of O i during F i j-1 , denoted Lat F i j-1

. Thus the capacity Capacity F i j-1 is computed according to the formula (4).

Capacity F i j-1 = 1 Lat F i j-1 × ∆ × deg j-1 (O i ) (4)
As the processing latency may uctuate over time, we take the covariance between Capacity F i j-1

and previous iterations into account to estimate the expected capacity during F i j . The estimated capacity of O i is computed as follow:

EstimCapacity F i j = Capacity F i j-1 + i ( 5
)
where i is the covariance between the capacity during F i j-1 and the capacities observed during the previous iteration of the monitoring window.

Identication of potential congestion at operator scope

We have presented methods to estimate input workload and processing capacity in near future at operator scope. To identify if a modication of parallelism degree will be benecial to an operator, it is necessary to evaluate if the operator will be able to process its estimated workload. Intuitively, for a given operator O i , if the estimated workload exceeds the capacity, O i will accumulate stream elements on its pending queue and may become a processing bottleneck. At the contrary, if the capacity exceeds signicantly the estimated workload, it means that the parallelism degree of O i is oversized for future processing requirements and could be reduced to save resources.

Local estimation of activity level

To represent the balance between estimated input load and capacity, we suggest the notion of activity level. Denition 16. (Activity level) Let consider an operator O i applied by a set of tasks observed at the start of the iteration F i j of a monitoring window. The activity level of O i is the ratio between estimations of its input load and its processing capacity on F i j . Applied at operator, or local, scope, the Local Activity Level, denoted LAL, is dened in formula (6).

LAL F i j = EstimInput F i j EstimCapacity F i j (6)
The LAL is said local because it relies exclusively on operator history without consideration for upstream operators. It is computed at the end of each iteration of the monitoring window independently for all operators belonging to a workow. From the value of the LAL, a modication of parallelism degree can be suggested to t current capacity to estimated workload in near future.

Identication of activity states

Let θ min and θ max be two thresholds delimiting respectively a low and a high activity level, with θ min , θ max ∈ ]0;1]. For a given operator O i , modications of parallelism degrees are suggested according to the following policy:

• If LAL F i j
≤ θ min , the local activity of the operator is 'low' because operator capacity is at least 1 θ min greater than the estimated workload EstimInput F i j .

• If θ min < LAL F i j ≤ θ max , the local activity of the operator is 'medium' because the operator is able to process all items during F i j but EstimInput F i j is greater than θ min EstimCapacity F i j .

• If θ max < LAL F i j ≤ 1, the local activity of the operator is 'high' because the operator has just the capacity to process stream elements waiting to be processed during F i j .

• If LAL F i j > 1, the local activity of the operator is then 'critical' because the operator is not able to process EstimInput F i j with its estimated capacity EstimCapacity F i j during F i j .

Consistency at workow scope

We have determined processing requirements of each operator in near future according to their own histories. From activity states recommended locally for each operator, we could trigger scalein and scale-out without considering the upstream and downstream operators. Nevertheless, when the activity level is critical, a modication of the parallelism degree of an operator may aect the throughput of the operator. By consequence the input rate of downstream operators may be aect accordingly. If parallelism degrees of multiple operators are changed simultaneously, it may lead to inconsistent recongurations while considering a workow in its entirety. For example, let consider a workow W and let suppose that the system applies suggestions of reconguration computed locally.

As illustrated on gure 5, performing a scale-out increases the throughput of an operator so the input load of downstream operators will increase accordingly after reconguration. Considering that modications of parallelism degrees are performed simultaneously for all operators in order to minimize reconguration overheads, three inconsistent cases may occur after reconguration:

• If a scale-out is performed upstream and a scale-in is performed locally (see gure 6), it may cause a congestion because the input rate increases while the capacity decreases. Of course, it depends on amplitudes of scale-out and scale-in.

Figure 6: Potential inconsistent case 1

• In the same logic, if a scale-out is performed upstream and nothing is changed locally (see gure 7), the operator may have a high or critical activity level in near future and it is necessary to reevaluate if the current parallelism degree will handle the increasing load. • Finally, when scale-out is performed upstream and locally (see gure 8), and it is necessary to reevaluate if the expected capacity after reconguration will be appropriate to handle the input load. Indeed, a scale-out suggested locally may be based on a undersized estimation of local input load. By consequence, it may involve an additional scale-out later which degrades the stability of the system. It appears necessary to consider dependencies between operators while computing the set of recongurations to perform. Nevertheless, it requires to evaluate the impact of a reconguration on downstream operators.

Construction of the instantaneous graph of local activities

To tackle the consistency issue at workow scope, we suggest an instantaneous graph of local activities (IGLA) which allows to analyze the activity of an operator and take activities of upstream operators into account. The IGLA sums up metrics necessary to evaluate the impact of a reconguration on downstream operators. As illustrated on gure 9, an IGLA is an attributed graph where each vertex corresponds to an operator of the workow to recongure. Each vertex is associated to a vector of attribute/value pairs where each attribute corresponds to a metric computed locally as shown on gure 9. Indeed, from local estimations, we can infer which modications should be applied and detect inconsistencies as explained above.

Evaluation of reconguration impact

Let consider a monitored operator O i during an iteration F i j . We have measured the total number of stream elements it has processed, denoted processed F i j , and the total number of stream elements it has emitted during F i j , denoted output F i j

. Considering that O i can be selective (a lter, a join...), we compute its selectivity factor SF F i j during the iteration F i j according to formula [START_REF] Biem | Ibm infosphere streams for scalable, real-time, intelligent transportation services[END_REF]. It is worth noting that processed F i j is a measurement and not a theoretical estimation like Capacity F i j .

SF F i j = output F i j processed F i j (7)
According to the estimation of incoming load EstimInput F i j and the current capacity Capacity F i j of the operator O i , we dene the estimated number of processed elements during the next iteration, denoted EstimP rocessed F i j , according to formula [START_REF] Akidau | Millwheel : faulttolerant stream processing at internet scale[END_REF].

EstimP rocessed

F i j = min(EstimInput F i j , Capacity F i j × ∆) (8)
Indeed, an operator can at most process the number of items corresponding to its capacity per time unit multiplied by the duration of an iteration. With this estimation and the selectivity factor SF F i j , we estimate then number of items emitted by O i on F i j+1 , denoted EstimOuput F i j+1 thanks to the following formula:

EstimOutput F i j+1 = EstimP rocessed F i j × SF F i j (9)
According to this value, it is possible to have a complementary estimation of the incoming load of next operators. Actually, let consider a child operator O c receiving its inputs from a parent operator O p . The value EstimOutput F p j+1 is intrinsically dierent from EstimInput F c j because it is not based on items already received by O c as illustrated on Figure 10. Indeed, EstimOutput F p j+1 is computed from items received and processed by the previous operator, in this example, O p . Intuitively, it gives a greater anticipation of critical variations of the global input rate.

Consistency checking at workow scope

So, still considering an operator O c receiving its inputs from an operator O p , we have at disposal two distinct estimations of the incoming load of O c : its local estimation EstimInput F c j and the global estimation EstimOutput F p j+1 . The choice of the estimation to consider depends on which aspect the DSMS should favor.

Indeed, if the DSMS serves as guarantee that the capacity of each operator remains great enough to absorb its incoming load, the maximal value between EstimInput F c j and EstimOutput F p j+1 is considered to adjust the parallelism degree of O c . According to available estimations and H1, it ensures that each operator is able to process all incoming elements. Nevertheless, it prevents to perform a scale-in until local and global estimations conrms that it does not lead to a potential congestion.

In opposition, if the DSMS aims at using only necessary resources, the minimal value between EstimInput F c j and EstimOutput F p j+1 is used. According to that combination strategy, the DSMS decreases the capacity of operators as soon as it is locally or globally advisable. Yet, this strategy presents as drawback to degrade the stability of the system. Indeed, decreasing capacities of operators to save resources as soon as possible also means increasing them each time the incoming load increases signicantly.

For both combination strategies, we consider the globally consistent estimation as the result of a function combine which takes both estimations as input and returns the globally consistent estimation according to the DSMS' objective.

Considering a combination strategy, we consider that the consistent estimation of input volume is the result of a function combine. This function takes as input the estimation EstimInput F c j computed locally and the estimation EstimOutput F p j+1 computed upstream.

Global estimation of activity level

To evaluate accurately which operators should be revised, it is necessary to estimate and propagate the eect of each reconguration along the workow. Thus, autoscale is able to detect if a reconguration recommended locally worth being triggered according to recongurations performed upstream. Formally, we will override the value of the LAL F i j for an operator O i by an activity level taking into account estimations of inputs performed upstream. We compute this Global Activity Level, or GAL, according to formula [START_REF] Carbone | Apache ink : Stream and batch processing in a single engine[END_REF].

GAL F i j = combine(EstimInput F i j , EstimOutput F P i j+1
)

EstimCapacity F i j (10)
where

EstimOutput F P i j+1
is the sum of the estimated outputs of all parent operators of O i .

Once the LAL has been replaced by the GAL, autoscale can reconsider the activity state (low, medium, high and critical) of each operator and decide which reconguration should be triggered. ;

EstimInput F i j ← combine(EstimInput F i j , EstimP arentOutput); next ← degree j (O i ); if current > next then setScaleIn(IGLA, O i , next); end if if current == next then setN othing(IGLA, O i , current); end if if current < next then setScaleOut(IGLA, O i , next); end if checked ← checked ∪ {O i }; end if end for if operators = ∅ then checkConsistency(operators, IGLA); end if
As presented in Algorithm 1, the consistency checking algorithm explore the IGLA from sources. Estimations of input loads computed on source operators cannot be combined with estimations computed on upstream operators as there are entries of the workow. Then, the function children() look for all children operators of each source. If the function activity() returns that a child operator has a critical parent at local scale and the method unchecked() returns that it has not been checked already, we compute the globally consistent estimation of its incoming load and replace its local estimation. It propagates the eect of critical estimation to all operators processing stream elements emitted by a source. Then, we compare the current parallelism degree given by function currentDeg() and the adequate parallelism.

We can map each operator to a modication of its parallelism degree as presented in Table 1. This decision takes into account the global activity of a given operator and the evolution trend of its incoming load. As a reminder, the ane function f i j is computed with linear regression to estimate the load of an operator. This function allows to evaluate the evolution trend of the load according to its derivative value. If this value is strictly positive, the load is considered increasing. Otherwise the load is estimated as decreasing or constant.

θ min ≤ GAL F i j < θ max nothing nothing θ max ≤ GAL F i j < 1 nothing scale-out 1 ≤ GAL F i j scale-out scale-out
To sum up, autoscale estimates the activity at local and global scope for each operator as illustrated on Figure 10. At local scope, autoscale computes an estimation of the incoming load thanks to monitoring data on received items and pending queues. This incoming load is divided by the estimated capacity of the operator to give a value of its local activity level. To propagate local estimations to next operators, the estimated output is computed. It relies on an estimation of processed items and the selectivity factor of the operator. For children operators, this estimation is combined to their local estimation of the incoming load to help the DSMS to reach its objective as introduced above. For each operator requiring scale-in or scale-out, we have to evaluate an appropriate parallelism degree. As we consider that users do not have an a priori knowledge of stream uctuations, the system cannot be trained on relevant input rates to build workload→parallelism degree mappings upstream. So, we have to approximate the parallelism degree under the assumption that the total capacity of an operator is strictly proportional to the number of tasks. It assumes that overheads induced by input split and merge are negligible.

Let deg j-1 (O i ) be the parallelism degree of O i during the iteration F i j . Let maxP O i be the maximal parallelism degree of O i , we consider that its appropriate parallelism degree is dened according to formula [START_REF] Noghabi | Samza : stateful scalable stream processing at linkedin[END_REF]. It is worth noting that we consider a maximal parallelism maxP O i for O i because most DSMSs limit the number of tasks associated to a single operator. Moreover, when the parallelism degree of an operator a threshold, overheads involved by data routing and potential network transmissions balance the benet brought by the parallel execution.

deg j (O i ) = min(maxP O i , deg j-1 (O i ) + 1), if activity is 'high' min(maxP O i , deg j-1 (O i ) × GAL F i j ), otherwise (11)
We distinguish the specic case where an operator has a high activity and an increasing input rate. Indeed, the value of GAL F i j is smaller than 1, but a scale-out is recommended (see Table 1). In this case, we simply increment the parallelism degree of the operator by 1. In any other case, autoscale considers as the appropriate parallelism degree, the smallest parallelism degree greater than the current parallelism weighted by the value of GAL F i j .

Discussion

We have presented an auto-parallelization strategy estimating common metrics of the literature for each operator and in near future. It allows to anticipate potential congestion of operators instead of removing eective ones. In addition, we suggest an algorithm for consistency checking at workow scope in order to avoid inconsistent reconguration of operators.

Nevertheless, the time complexity of this algorithm is in O(V + E) where V is the set of operators and E the set of streams between operators, or inner streams. If the workow is composed of many operators highly connected, the computation of the global activity for each operator may add some overheads. So, to improve the anticipation of congestion, two alternative policies to congestion may be used: a policy considering exclusively local estimations for each operator, denoted LocalOnly and a policy checking the consistency considering only the type of reconguration performed upstream, denoted Straight policy.

scale-in and scale-out may be decided from estimated metrics at local scope (i.e., exclusively from operator history).

While the local activity has been computed, scale-in opportunities and scale-out needs can be identied as presented in Table 2 We can apply the straight policy for consistency checking which does not take the impact into account but only symbolic inconsistencies. By nominal, we mean scale-in, scale-out and nothing.

After deciding locally which reconguration suits to an operator, the decision matrix presented in Table 3 can be used to replace inconsistent decisions. We explore the IGLA according to a breadth-rst search, or BFS, from sources to nal operators. Considering the global decision matrix, we identify the consistent subset of actions to perform for each workow. For instance, if a scale-in is recommended at local scope but a scale-out has been validated upstream then the current parallelism degree of the operator is remains unchanged. It is interesting to notice that if a scale-in intervenes upstream current operator and nothing has been recommended locally, this policy prefers maintaining the current parallelism degree than decreasing it because it presents more risks to decrease parallelism degree before local and global recommendations conrm such reconguration. In comparison to the consistency checking algorithm of autoscale, it saves computations for global consistency as the selectivity factor and the combine function.

P P P P P P P P P P P P P P P P P P P P

Prevailing action upstream operator

Local suggestion for operator nothing scale-in scale-out nothing nothing scale-in scale-out scale-in nothing scale-in scale-out scale-out scale-out nothing scale-out Nevertheless, an issue occurs when the current operator has multiple parents with dierent suggestions of reconguration. To solve this issue, we suggest a relation order on actions. So, a scale-out prevails on a scale-in which prevails on a nothing action. To determine if a local action is globally consistent, we introduce the global decision matrix (see Table 3).

According to the relation order on possible actions, the Straight policy can determine the predominant action for all nodes upstream. Thus, if at most a scale-in is validated upstream the current node, the local action is validated otherwise it is replaced by a nothing or a scale-out in order to avoid foreseeable congestion in near future.

This policy increases the consistency of reconguration at workow scope but it assumes that each upstream scale-out causes a signicant increase of input rate which is not systematically the case. So, we suggest to enrich the IGLA with additional metrics in order to evaluate accurately the impact of each reconguration on downstream operators.

Empirical study of consistency checking

In order to evaluate the interest of the consistency checking algorithm integrated in autoscale, we test it in front of autoscale without consistency checking (LocalOnly) and autoscale with nominal consistency checking. The stream illustrated on Figure 11 has major increases in input rate followed by decreases to highlight adaptations performed by autoscale.

This stream is played in entry of the workow illustrated on Figure 12. This linear workow is composed of a source emitting the stream illustrated on Figure 11 to an operator (FastNonFilter) transmitting each stream element within a millisecond without ltering inputs. Outputs of this operator are sent to an operator (SlowNonFilterMid) applying heavy treatments without ltering. Each stream element is processed around 100ms. Outputs of the SlowNonFilter operator are consumed by an quick lter (FastFilter) which processes each input within a milliseconds but also lter its inputs. The operator FastFilter has a xed selectivity factor of 0.1. Finally, the operator SlowNonFilterEnd has the same properties than the operator SlowNonFilterMid.

This workow is interesting to evaluate consistency checking because the heterogeneity of processing latency and selectivity factors represent real-world complex applications. So, in front For each strategy for consistency checking, we observe modication of parallelism degree performed for each operator, the average end-to-end latency of the workow, the throughput and the number of stream elements processed over a predened threshold. In this case, we set this threshold to 30 seconds. takes only local estimations into account (LocalOnly), the parallelism degree of the operator SlowNonFilterMid is increased and decreased according to uctuations in input rate. Indeed, upstream operator FastNonFilter does not modify signicantly variations in input rate. The parallelism degree of the operator FastFilter remains unchanged according to local metrics. It can be explained by the fact that the input rate is limited by the throughput of the operator SlowNonFilterMid and the average processing latency of the FastFilter is greater or equal to the processing latency of upstream operators. Finally, the parallelism degree of the nal operator SlowNonFilterEnd is increased by 1 when the stream is at its maximal rate. It does not have same modications of parallelism degree than SlowNonFilterMid because the original input rate has been limited by SlowNonFilterMid processing rate and reduced by FastFilter which has a selectivity factor of 0.1.

When a nominal consistency checking strategy is applied (Straight policy), the operator SlowNonFilterMid is initially recongured according to local estimations but this reconguration implies a scale-out of the operator FastFilter according to the decision matrix presented in Table 3. The parallelism degree of FastFilter is adapted according to the linear projection on SlowNonFilterMid workload. It involves an additional reconguration which is not necessary as the maximal throughput SlowNonFilterMid can deliver with 14 tasks is smaller than the processing rate of FastFilter with a single task. So, the operator FastFilter is recongured each time SlowNonFilter requires a modication of parallelism degree. In the same logic, the nal operator SlowNonFilterEnd is recongured each time FastFilter is recongured.

When autoscale uses a consistency checking strategy computing the GAL for each operator (GALPolicy), we observe that the parallelism degree of SlowNonFilterMid is increased initially according to local metrics. But, at the second reconguration, we note that the parallelism degree is increased by 7 tasks instead of 3 because the estimation of inputs based on upstream operators is greater than the local estimation. The same phenomenon can be noticed on the third scale-out. Then, instead of taking local estimations into account to get down to 1 task, the parallelism degree of SlowNonFilterMid is decreased to 2 before being decreased to 1 later. As recommended by local estimations the parallelism degree of SlowNonFilterEnd is increased by 1 like it was without consistency checking. The impact of each sequence of scale-out/in can be evaluated in terms of performance and result quality at workow scope. The GALPolicy strategy improves both aspects compared to StraightPolicy and LocalOnly strategies. Concerning performance, GALPolicy maintains a smaller end-to-end latency because it does not trigger unnecessary scale-out like StraightPolicy and it anticipates predictable uctuations of workload at workow scope contrary to LocalOnly strategy. The throughput is signicantly greater when the input rate reaches its maximal value compared to StraightPolicy because of less reconguration and a better accuracy while estimating future workload. It can also be observed from out-of-time elements happening almost only with StraightPolicy because of consecutive recongurations of multiple operators which delays signicantly processing latency. In this testing simulation, the major improvement of GALPolicy compared to LocalOnly strategy is the combine method which avoid to underestimate processing requirements. So, the processing latency can decrease signicantly with GALPolicy because there are enough tasks to empty pending queues and process incoming stream elements without accumulating them for a long period of time.

Estimation of capacity

We justied empirically the interest of the consistency checking with GAL metric. Through the computation of the GAL, autoscale reconsiders the local estimation of input load. Nevertheless, the estimation of the capacity performed locally is only considered locally. Indeed, a local estimation of the capacity that suppose that two assumptions are true. First, the average processing latency of the operator will not change in near future. Such variations may be due to a sensitivity to values in input. To ensure that autoscale adapts accurately operators in front of even and uneven streams, the processing latency of each operator should remain as stable as possible. In the case of operators executed by multiple tasks, it can be done with a load-aware strategy for load balancing. Second, it is assumed that on an estimation period ∆, the operator will have as much available CPU time as it requires. Depending on the scheduling strategy and available processing units, some operators may be assigned on a same processing unit and share CPU time. To improve the accuracy of estimations, it is crucial to take concurrency into account for the computation of operator capacity. We presented the preventive auto-parallelization strategy autoscale which anticipates modication of parallelism degrees according to local and global estimations. The aim of this approach is to prevent congestion and limit increases of average processing latency due to accumulation of stream elements. So, it is assumed that each scale-out and scale-in respectively increase and decrease resources reserved for a given operator. For example, when autoscale estimates that an operator O i will have two times more stream elements to process in near future, its parallelism degree is at least multiplied by two to double available resources. Nevertheless, considering a scheduling strategy assigning tasks such as there could be concurrency for CPU and memory usage [Xu et al., 2014, Peng et al., 2015], this assumption may be wrong. Considering the operator O i , if its parallelism degree is multiplied by two but some tasks are assigned on processing units having less available resources than necessary, the processing capacity is not eectively doubled. Some solutions [Peng et al., 2015] oer the possibility to associate resource constraint for each operator. Still considering the operator O i , users can specify that a task of O i is assignable on a processing unit only if there are at least 20% of idle CPU time and 256Mb of free memory space. Then, the scheduling strategy denes the scheduling plan such as each task has at least required resources declared by users. It assumes two conditions:

• Users have a complete knowledge of relative time and space complexity of each operator.

This knowledge is particularly dicult to build if there are some user-dened operators without declared algebraic properties.

• Resources are fragmented such as if a task uses less resource than required, the dierence between required and used resources should be available at any time. In practice, idle resources may be used by other processes. The scheduling of active threads varies in according to CPU architecture and operating system.

So, such systems are designed to let users give indicative requirements for each operator with an accuracy depending totally of user expertise. While estimating processing capacities, the dierence between required and used resources by an operator aects signicantly the accuracy of the estimation. When there are concurrency on resource usage, taking eective resource usage into account is crucial to improve the accuracy of modication of parallelism degrees. In the remainder of this chapter, we consider DSMSs considering reservation of resources for each operator as described above.

Managing load between tasks in presence of uneven streams

To estimate the processing latency of an operator O i applied by many tasks, we consider the average latency of all tasks applying O i without evaluating the potential imbalance between tasks. Indeed, let consider a task T i 1 which processed values AAAB and another task T i 2 which processed values BBAB. Let consider that the key A takes 1 time unit to be processed and the key B takes 10 time units, the average latency of O i is 5.5 time units per stream element but the standard derivation is more than 2. This imbalance is negative for performance as it implies dierent processing requirements for task of a same operator. To tackle this issue, some load balancing strategies [START_REF] Rivetti | Ecient key grouping for near-optimal load balancing in stream processing systems[END_REF] have been developed. They aim at compensating such skew in value distribution through adaptive routing policy based on incoming values.

In the remainder of this chapter, we aim at dening an auto-parallelization strategy which takes concurrency on resource usage into account while estimating processing requirements of operators. This solution should not rely on static knowledge provided by users as it assumes user expertise and processing requirements may vary over time. In addition, this strategy should be paired with a load balancing strategy to serve as guarantee that value distribution does not lower benets brought by modication of parallelism degree.

2 Resource-aware auto-parallelization of operators

In this section, we present the auto-parallelization strategy autoscale+ which aims at tackling problems mentioned above. autoscale+ relies on the monitoring module of autoscale to get recent history of operators. In addition to these information, autoscale+ collects information about CPU usage for each task of operator. It allows to perform resource-aware parallelization of operators. In the remainder of this section, we detail the computation of input workload in near future at local and global scope within a single pass. Then, we present the evaluation of processing capacity taking potential concurrency on resource usage into account. Finally, we suggest a congurable adaptation policy oering the possibility to users to control the stability of the system.

Enhancement of workload estimation

As presented in chapter 5 section 3, autoscale+ aims at computing an estimation of the volume of stream elements to process in near future. autoscale+ reduces the computation time of workload estimation and increases the accuracy through the following improvements:

• Instead of estimating the activity level through a local and a global estimation, au-toscale+ explores each workow according to a breadth-rst search. For each operator O i , the value of |EstimR i,j | and EstimOutput F i j is computed according to formulas (2) and ( 9) presented in chapter 5 section 3. For sources, i.e. operators without upstream operators, the value of EstimInput F i j is computed according to formula (3) presented in chapter 5 section 3. For other operators, autoscale+, the value of EstimInput F i j is computed according to formula [START_REF] Arasu | The cql continuous query language : Semantic foundations and query execution[END_REF].

EstimInput F i j = combine(EstimR i,j , p∈par(O i ) EstimOutput F i j ) + pending F i j-1 (

Estimation of available resources

Now that we have estimated incoming and outcoming volumes for each operator, we need to approximate the maximal number of stream elements each operator can process during an iteration of duration ∆. We denote this number of stream elements the capacity of the operator. According to both estimations, then we are able to detect potential congestion. In an ideal case, the capacity is dened as follow:

IdealCapacity F i j = ∆ Lat F i j (2)
where Lat F i j is the processing latency. This processing latency does not take into account time spent in pending queue. Nevertheless, using this approximation means that each thread associated to the operator is able to maintain its execution during duration ∆ without being interrupted by other threads. In a distributed and multi-threaded environment, this assumption favors over-estimations of operator capacities. Each thread is associated to a resource constraint [Peng et al., 2015] and a current usage which could be greater, equal and smaller than the constraint. As CPU usage may vary suddenly overt time, we also consider a weighting factor α ∈ [0;1[. Thus, we underestimate lightly available CPU time to avoid fast overload. Let T X be a thread associated to a reservation constraint ResaCP U X . This thread is in concurrence with n other threads for the usage of a CPU C. We estimate that the usable CPU time by T X is dened according to formula (3).

U tilCP U (T X , C) = α × max(U sedCP U (T X , C), ResaCP U X ) + 1 n (100 - ∀T Y =T X U sedCP U (T Y , j)) (3)
Considering all threads T 1 i , T 2 i ,..., T m i applying an operator O i , we dene the global CPU time U tilCP U i for T 1 i , T 2 i ,..., T m i consuming resources of CPUs CP U (T 1 i ), CP U (T 1 i ),..., CP U (T m i ) as follow:

U tilCP U i = min x=1...x=m (U tilCP U (T x i , CP U (T x i )) (4)
We assume that an increase in input rate aects all threads executing the same function equally.

With this estimation, we can approximate usable CPU time with a greater precision according to current thread assignments. Thus, we can improve the denition of the capacity as follow:

Capacity F i j = ∆ Lat F i j × U tilCP U i (5)

Balance between processing requirements and resources

Now, we have an estimation of incoming volumes and accurate capacities for all operators, we can detect imbalance between processing requirements and resource usage. There are three possibilities: detecting a need of scale-out, a possibility of scale-in or doing nothing. While autoscale+ detects a need of scale-out, the system should recongure itself to avoid at least a degradation of performance due to the accumulation of stream elements on pending queues.

In the case of a scale-in, the system should benet from a decrease of parallelism degree in terms of performance and active resources. Nevertheless, performing a scale-in brings overheads that future benets does not systematically compensate. For example, if performing a scale-in reduces overall resource usage by 2% but increases massively the average processing latency until migrations of pending queues are completed, it does not worth for users wanting results with short latency or paying for all available resources without distinction between active and inactive ones. In such case, it appears relevant to let users dene which benet should bring a scale-in, in terms of resources saved, to compensate reconguration overheads.

Working interval

Now that we have an estimation of the incoming workload and the capacity taking resource usage into account, we can approximate the ideal parallelism degree, denoted idealK according to formula (6).

idealK = EstimInput F i j Capacity F i j ( 6
)
The problem is then to dene if it worth modifying the current parallelism degree. Indeed, if the parallelism degree idealK is lower than the current parallelism degree k but does not bring a minimal benet in terms of performance, triggering a scale-in involves reconguration overheads that are not compensated by a signicantly better performance.

To represent this notion of benet, we suggest a controllable working interval associated to each parallelism degree k. This interval has as upper bound the current parallelism degree k and as lower bound a parallelism degree min k which is function of the current parallelism degree k.

The value of min k is dened according to formula [START_REF] Biem | Ibm infosphere streams for scalable, real-time, intelligent transportation services[END_REF].

min k = β × k (7)
where β ∈ ]0;1] is a controllable parameter. If β is close from 0, it means that autoscale+ performs scale-in only when input volumes are very small compared to operator capacities. If

β is close from 1, autoscale+ performs scale-in as soon as a smaller parallelism degree k'

guaranties theoretically that input volumes can be processed. On the contrary, if idealK is smaller than min k , it means that k will be over-evaluated for the next iteration F i j+1 . Within this working interval, a possibility of scale-in is not considered as it does not save enough resources to compensate reconguration overheads.

More formally autoscale+ recommends to keep the current parallelism for an operator O i as long as the following condition is valid:

min k ≤ idealK ≤ k (8)
It is worth noting that the greater is k, the greater is the associated working interval. We opt for such property because the more there are tasks to merge into less tasks, the more it takes time to merge pending queues distributed over the cluster and re-route stream elements.

Computation of the appropriate parallelism degree

Thus, autoscale+ computes an appropriate parallelism degree k' according to formula [START_REF] Peng | Rstorm : Resource-aware scheduling in storm[END_REF].

argmin k ( EstimInput F i j ResCapacity F i j ≤ k) (9)
where ResCapacity F i j is the capacity of O i considering the CPU constraint ResCP U i . The capacity ResCapacity F i j is dened according to formula [START_REF] Carbone | Apache ink : Stream and batch processing in a single engine[END_REF].

ResCapacity F i j = ∆ Lat F i j × ResCP U i × α ( 10
)
where α ∈ ]0;1] is a parameter allowing autoscale+ to consider a relative margin between eective CPU usage and CPU reservation. It means that autoscale+ takes into account the fact that some threads may need more than their reservation at runtime. As β, the parameter α can be dened through several methods like empirical study, reinforcement learning or user expertise.

It is worth noting that the new parallelism degree is computed considering only the CPU requirement declared by users and not the last value of U tilCP U i . Indeed, a modication of parallelism degree will lead to a modication of thread assignments so U tilCP U i may change after reconguration of the system.

3 Load management 3.1 Auto-parallelization of operator with load imbalance Load imbalance may occur in a stream processing context. Let consider an operator O i applied by k tasks T i 1 to T i k . In addition, the average processing latency per stream element Lat i of O i is a function of the value v j in input. Each value v j belongs to an ordered set of values V ={v 1 , v 2 , ..., v m } and is associated to a specic processing latency Lat i (v j ) such as Lat i (v j ) < Lat i (v j+1 ). Let consider a stream sequence S of size q×k stream elements such as the distribution of values over S is uniform.

Figure 5: Worst and optimal cases of load balancing Now, let consider a load balancing strategy distributing stream elements in a Round-Robin fashion. As illustrated on gure 5, after the distribution of q×k stream elements, processing requirements in terms of computation of times are q×Lat(v 1 ) for T i 1 and q×Lat(v m ) for T i k in the worst case. Considering that the capacity of O i is computed taking the average processing latency into account, tasks associated to O i and having greater processing requirements will be under-provisioned in CPU time due to load imbalance. It causes more pending stream elements and increases future estimation of the workload. At the opposite, an optimal load balancing strategy would have distributed the same stream sequence of q×k such as each task has an average processing latency per stream element equals to the average latency Lat i (V ) with Lat i (V ) dened as follow:

Lat i (V ) = q × m j=1 Lat(v j ) m ( 11
)
By consequence, the imbalance due to load balancing may lead the auto-parallelization strategy to perform unnecessary scale-out which does not improve the performance but at the contrary bring additional overheads with reconguration. It appears then necessary to combine autoscale+ with a load balancing strategy.

Compatibility issues

Combining an auto-parallelization strategy with a load balancing strategy may raise some compatibility issues. Indeed, an auto-parallelization approach like autoscale+ relies on the fact that a congestion is due to an overload in input of all tasks associated to an operator, i.e., adding more tasks reduces eectively the workload of each task. In the specic case of load imbalance, a task may be overloaded while others process stream elements normally. This overload comes from an uneven distribution of stream elements according to their values. As some operators are sensitive to stream element values, the processing latency can signicantly vary depending on which value is read in input.

To tackle this issue, the auto-parallelization should rely on a load balancing strategy guaranteeing that the workload is evenly distributed between tasks of an operator with regards to the distribution of values. So, load balancing strategies relying of key grouping to build partitions [START_REF] Neumeyer | S4: Distributed stream computing platform[END_REF] are not ecient in this context. Indeed, they route statically stream elements to tasks according to their values without consideration for the volume of each partition. On the opposite, load balancing strategies balancing statically the number of stream elements in each partition may create imbalance if the operator is sensitive to values in input and the distribution of values changes over time.

To sum up, it appears necessary to combine autoscale+ with a load balancing strategy which takes into account the processing latency for each value appearing in the stream. Thus, the workload of each partition may be balanced at runtime according to the distribution of values.

Discussion

We presented the preventive auto-parallelization strategy autoscale+ which takes eective resource usage into account to modify parallelism degree of operators. For a given operator, autoscale+ assumes that the incoming load is evenly distributed over its tasks. This property is guaranteed by the resource-aware load balancing strategy OSG. In this section, we aim at evaluating benets brought by the combined approach through a comparison with static parallelism management.

Online Shue Grouping for resource-aware load balancing

As explained above, autoscale+ should modify parallelism degree of operators more accurately if it is combined with an optimal load balancing strategy. Indeed, the optimal load balancing strategy ensures that each task has the same workload. So, the estimation of the required capacity computed at operator scope ts to the input load of each task.

Computing an optimal routing policy of stream elements requires an a priori knowledge of Lat(v j ) for all v j ∈ V . Nevertheless, in a stream processing context, the set V is generally unknown at the beginning of treatments as the time complexity of O i . Moreover, the scheduling of tuples to tasks must be performed online i.e., the load balancing algorithm does not know the sequence of stream elements to schedule.

To solve this issue, we choose to associate autoscale+ to the resource-aware load balancing strategy OSG [START_REF] Rivetti | Online scheduling for shue grouping in distributed stream processing systems research paper[END_REF], for Online Shue Grouping. OSG performs online load balancing through the combination of two Count Min Sketches [START_REF] Cormode | An improved data stream summary: the count-min sketch and its applications[END_REF] and a Greedy Online Scheduler. OSG associates two matrices of controllable sizes to each operator O i . For each matrix, each row is mapped to an exclusive 2-universal hash function [START_REF] Carter | Universal classes of hash functions[END_REF]. Each time a stream element t is read in input, the Count Min Sketch algorithm updates the occurrence of t in the rst matrix and the completion time in the second matrix. Thus, according to both sketches, OSG is able to return the average completion time of any value seen previously. This estimation is approximated with a bounded error as analyzed in [START_REF] Rivetti | Online scheduling for shue grouping in distributed stream processing systems research paper[END_REF]. A Greedy Online Scheduler uses these estimations to route an incoming stream element to the task able to process it with the shortest delay. This delay is computed as the time needed to execute all pending stream elements. More details about OSG and theoretical analysis can be nd in [START_REF] Rivetti | Online scheduling for shue grouping in distributed stream processing systems research paper[END_REF].

Empirical study of the combination autoscale+ and OSG

We implemented autoscale+ and OSG over the stream engine Apache Storm 1.0.216 . We perform tests on a the workow illustrated on gure 6. This workow is composed of several operators with various selectivity factors and average processing latency. The spout (OpinionSource) emits stream elements concerning opinions submitted by users about a topic. Each opinion is described by information on the user, like its age and code representing its location, the topic and user opinion. Stream elements are sent to a bolt (CategoryDispatcher) ltering unnecessary attributes and depending on the branch downstream. In addition, it lters stream elements concerning a predened list of irrelevant topics. A branch starts with a bolt (CityNormalizer) retrieving information on user location from the code. This bolt has the exact same properties than the sensitive bolt of the simple sensitive topology. Indeed, depending on the code, retrieving information on the city takes more or less time. It allows us to compare the impact of workow structure and complexity on bolt behavior and dynamic adaptation of its parallelism degree. Then, a bolt (CityAnalyzer) extracts relevant subgroups according to opinion and location. The other branch starting from the bolt CategoryDispatcher performs similar treatments in order to dene subgroups on user opinion and age. Finally, the Persister takes in input descriptions of subgroups and persists them in a storage le system. In a rst time, we focus on benets brought by the association of autoscale+ and OSG introduced in [START_REF] Rivetti | Online scheduling for shue grouping in distributed stream processing systems research paper[END_REF]. We observed the behavior of the workow presented on gure 6 while receiving the synthetic streams in function of following congurations:

• The default conguration dening statically parallelism degree of operators for the complete lifetime of the workow and balancing the load of each operator stream elements according to a Round-Robin (shue) policy.

• The same conguration but using OSG as load balancing strategy.

• autoscale+ as the auto-parallelization strategy and the shue policy for load balancing.

• autoscale+ as the auto-parallelization strategy and OSG for load balancing.

For each conguration we observe the average processing latency and the throughput of the workow to evaluate the performance. Concerning result quality, we dene a processing latency threshold discriminating nal results computed within a satisfying end-to-end latency. Parameters α and β have been dened through an empirical study. Indeed, we have evaluated the impact of these parameters on scale-in and scale-out and choose these values as the most appropriate ones.

Evaluation with uniform distribution

While processing the synthetic stream with the dierent congurations, we notice major dierences between static management of parallelism degrees and autoscale+. Indeed, congurations using autoscale+ maintain treatments while facing critical uctuations in input rate as illustrated on gure 8. So, result quality is improved with 94.8% of all stream elements processed under the time threshold against 26.1% with the static management of parallelism degree. OSG strategies for load balancing. Concerning performance, it is worth noting that OSG keeps latency 39% lower than Round-Robin strategy after reconguration. Load balancing between tasks of the critical operator is improved faster with OSG. Actually, when OSG detects new tasks, it routes new stream elements in priority to thee new tasks. At the opposite, the Round-Robin strategy keeps an imbalance between old and new tasks for a longer duration.

To highlight this imbalance, we focus on congurations using autoscale+ with the Round-Robin strategy and with OSG. It is worth noting that the standard deviation of CPU usage of all tasks of the critical operator is in average 4.4% with OSG and 5.6% even with a uniform distribution of values. Moreover, the ll ratio of pending queues is in average 26% lower with OSG than with the Round-Robin strategy. The lower ll ratio of pending queue has an impact on input workload and estimations performed by autoscale+. So, with OSG, autoscale+ estimates more accurately parallelism degree to handle maximal load and delivers better performance.

Evaluation with biased distribution

We repeat the same experiment but this time, we play in input the biased version of the synthetic stream described in introduction of experimental study. For remainder, uctuations in input rate remain unchanged but the distribution of values follows a Zipf law. Each value in the stream is associated to a specic processing time going from 10 to 90 milliseconds.

Between static congurations and congurations using autoscale+, we observe similar behaviors mentioned above. So, adaptations performed by autoscale+ allows the system to process 94.6% of all stream elements against 26.8% in average for static congurations. For congurations using autoscale+, we observe same behaviors too. Nevertheless, the bias in data distribution has an impact on load balancing. Indeed, compensating load imbalance with such stream is harder than with a uniform distribution of values. So, the standard deviation of CPU usage of all tasks is increased to 6.8% with the Round Robin strategy and reduced to 3.8% with OSG. Nevertheless, the ll ratio of pending queues is in average 11% less important with OSG than the Round-Robin strategy.

In conclusion, the auto-parallelization strategy autoscale+ allows to adapt parallelism degrees of operator before the host of a critical operator is overloaded. It maintains treatments while facing critical uctuations in input rate. The association with the load strategy OSG reduces the imbalance between tasks of an operator. It has two eects: decrease the ll ratio of pending queues which improves the end-to-end latency of the workow. Then, it reduces workload estimations performed by autoscale+ which results in better performance in front of important input rate.

Limits of autoscale+

We presented the approach autoscale+ which is able to anticipate the congestion of operators.

The estimation of workload in near future is accurate under the assumption that the evolution trend of stream rate does not vary signicantly in near future. Indeed, if autoscale+ detects a signicant increase input rate for an operator, it may trigger a scale-out. If the input decreases deeply after the analysis performed by autoscale+, a scale-in can be triggered and involve In this section, we summarize solutions for parallelization, load balancing and scheduling of operators that we experiment in the remainder of this chapter (see Figure 1). Each solution solution has selected for its popularity over existing solutions and is described briey with its specic features.

For the parallelization of operators, we consider ve approaches:

• The static approach oers only the opportunity to set parallelism degree of operators on user demand. To prevent the congestion of operators with this approach, users must monitor continuously states of operators and trigger scale-in or scale-out manually. Moreover, users must have an expertise in order to determine which parallelism degrees satisfy current processing requirements.

• The approach autoscale presented in chapter 5 prevents the congestion of operators according to an estimation of processing requirements in near future. • An incremental (incr) strategy observes the input rate and the throughput of each operator. Considering two thresholds min and max, the INC strategy increases the parallelism degree by one if the ratio input rate on throughput exceeds max and a scale-in if this ratio is lower than min.

• A reinforcement learning-based (rlearn) strategy mapping input rates to appropriate parallelism degrees at runtime. The RL strategy is initialized with a knowledge base covering stream uctuations in input rate. It assumes that users can train the system with representative uctuations which is not always the case in practice.

In order to balance the load within operators, we consider two solutions:

• The shue grouping (shue) which corresponds to a Round-Robin distribution of stream elements to tasks (see chapter 4). This solution aims at balancing the number of stream elements processed by each task of an operator without consideration for the distribution of values.

• The OSG (osg) approach presented in chapter 6 which estimates the processing time necessary to compute each value of stream element. Thus OSG can balance the load between tasks of an operator even if the distribution of values varies signicantly over time.

Finally, we consider a single scheduling strategy named resource-aware scheduling and presented in [Peng et al., 2015]. This strategy takes in input constraints on resource availability for each operator (CPU and memory) and returns a near-optimal scheduling plan which minimize the consumption of resources. All those approaches have been implemented in the DSMS Apache Storm due to the lack of open-source and extensible solutions integrating at least one the solutions mentioned above. Moreover, we have chosen this solution over other ecient SPEs such as Apache Spark Streaming [START_REF] Zaharia | Discretized streams: A fault-tolerant model for scalable stream processing[END_REF] due to data management. Yet, Spark Streaming systematically groups stream elements into batches, called Resilient Distributed Datasets (RDD). Nevertheless, if RDD sizes are large compared to incoming volumes, detection of congestion and over-consumption of resources is delayed. Thus, RDD size must be managed dynamically in addition to parallelism degrees. Compared to Apache Flink, Storm also provides technical support to implement an auto-parallelization strategy without aecting the core of the system. In addition, at the beginning of developments, Apache Flink was not released yet and lack documentation about its design and performance. To summarize, operators, named components in Storm terminology, belong to one of two categories: spouts or bolts. A spout is a connector to a raw stream source and represents an entry of a topology. It distributes stream elements to components to which it is connected and can process ltering operations if required. Bolts consume stream elements from any component and compute a result for each element received (stateless bolt) or for a set of stream elements (windowed bolt).

Each component is executed in parallel by executors. An executor is an instance of an operator. Each executor is assigned to a processing unit by the scheduler (see gure 2). The number of executors for a given spout/bolt is revised at runtime only at user request [START_REF] Xu | Stela: Enabling stream processing systems to scale-in and scale-out on-demand[END_REF].

Concerning the execution support, Storm relies on two types of processing nodes: Nimbus and supervisor. The Nimbus acts as a JobTracker for Hadoop . As illustrated on gure 2, each supervisor manages a pool of workers, i.e. processing units, and monitors executors assigned on them. dene a near-optimal scheduling plan. To perform auto-parallelization according to monitored metrics, autoscale and autoscale+ rely on three modules:

Implementation of autoscale

• Component Monitor: this module is in charge of monitoring continuous queries at operator scope and triggering recongurations when necessary. To collect all measurements at runtime, a client listens to internal metrics gathered by the Nimbus node. This client is integrated within a sub-module named the Nimbus listener. These raw data are pre-processed and stored in a MySQL database through a Statistic Storage Manager sub-module. The Statistic Storage Manager is also in charge of providing methods to group measurements by iterations of the monitoring window. Measurements are analyzed by the Scaling Monitor sub-module which includes the auto-parallelization strategy presented in chapter 5 or 6 depending on the version.

• Assignment Monitor: this module is in charge of collecting information on the execution support and assignments of operators on processing units. Through this module, autoscale can extract the scheduling plan and compute the parallelism degree of each operator, which resources are available on each processing unit and the amount of CPU and memory resources each task requires.

• Topology Explorer: this module is in charge to facilitate the exploration of each submitted queries. Indeed, each time a new continuous query is submitted, this module builds static knowledge on its structure for faster exploration. For example, it provides methods to identify entry and output operators of a topology but also methods to identify the children and parents operators of a given operator.

Evaluation of autoscale

In this section, we present the an experimental evaluation of autoscale. Due to the lack of open-source implementations of auto-parallelization strategies compatible with Storm, we focus our eorts on the comparison of autoscale with the static approach (see section 1) natively integrated in Storm. We evaluate the benet brought by Storm depending on the expertise level of the user on a micro-benchmark.

Experimental protocol

Execution support

Our test cluster is composed of 7 VMs. Each VM has at its disposal a dual-core CPU Intel(R) Xeon(R) E5-2620 running at 2.00GHz, 4Gb of RAM and 40Gb of hard disk space. A machine runs the Nimbus daemon and is dedicated to cluster coordination. Each supervisor manages 4 workers. On the Nimbus host, a MySQL database is also deployed in order to store historical data as illustrated above.

Workows and streams

To validate our approach, we choose to study its impact on three elementary topologies: a linear, a diamond and a star topology. Each elementary topology is composed of two types of bolts: intermediate bolts with low latency and sink bolts with high latency. In this section, we choose to present only some results relative to the linear, diamond, star and a complex topology. More detailed results are also available on our website 17 . Moreover, implementation, datasets and topologies can be downloaded for reproductibility. We built a 3-step synthetic stream with the following characteristics: 1) distribution with a small standard derivation 2) signicant increase and decrease in load. Indeed, as illustrated on gure 4, input load is constant at a low rate. Load increases progressively before stabilizing at a high rate. Finally, rates decrease markedly until it reaches the initial low rate. We also add small and irregular uctuations in order to simulate uctuations in a real stream. To comply with good Storm practices, we implemented the replay of out-of-time stream elements.

Then, we apply a 5-step stream with sudden input rate peaks (see gure 4) to test the reactivity of our approach. This second stream moves from a low input to a very high one without a progressive transition as presented above. The input rate decreases suddenly before increasing again.

Criteria of evaluation

We summarize the main experimental parameters in table 3 We collect all measurements each 10 seconds and group them in windows of 60 seconds. autoscale considers that the activity level of an operator is low if it is lower than 0.3 and high or critical if it exceeds 0.8 (see chapter 5 section 3). A stream element is considered as out of time, or obsolete, if it has spent 30 seconds or more within the topology. This timeout takes time spent in pending queues and network latency into account in addition to processing times within operators. Finally, the consistency checking of reconguration at workow scope (see chapter 5) considers the max as the combine function.

For each conguration, we measured the global latency of the topology (performance) and the number of dephased stream elements (result quality). Concerning system reactivity and the usage of resources, we observed parallelism degrees of each bolt.

Results on the microbenchmark 3.2.1 Application on the linear topology

We compare autoscale to the native scheduler of Apache Storm according to two congurations. We summarize experimental congurations for the linear topology in table 2: With the conguration ConfMin, the initial number of executors per bolt corresponds to minimal degrees (see table 2). Intuitively, the conguration ConfMin is adapted to small incoming loads but cannot handle large ones. With the conguration ConfExpt, initial numbers of executors correspond to expert degrees (see table 2). Expert degrees have been chosen with full knowledge of stream variation and latency of operators. This conguration can in fact handle the maximal load without wasting resources. With ConfMin, we observe that the incoming load cannot be handled, thus leading to the complete congestion of the topology. Indeed, the topology is not able to process stream elements completely. As soon as congestion occurs, new stream elements emitted by the spout are dephased and replayed indenitely until a user intervenes (see gure 5a). On the contrary, our autoparallelization strategy increases dynamically and automatically the parallelism degree of critical operators in order to adjust their capacities to future incoming loads. When the stream rate decreases, the parallelism degree decreases accordingly. It also prevents overusing resources that are no longer necessary.

With ConfExpt (see Figure 5b), we start with a conguration able to handle large loads. Nevertheless, this conguration overuses resources when the stream rate is low. It corresponds to the start and the end of the synthetic stream. Our auto-parallelization strategy reduces the parallelism degree when operators do not need large capacities. In this case, just as with ConfMin, the parallelism degree is adapted dynamically. Thus, autoscale achieves equivalent performance with approximately 37.5% less CPU and memory resources. The signicant increase in topology latency with autoscale is due to a scale-in from three to one supervisor, which re-routes multiple stream elements and implies this signicant overhead. Unsurprisingly, we can see on Figure 6, that with ConfMin, Storm is unable to handle the sudden increase in input rate and that topology is completely congested. Even the decrease in input rate is not enough to restore normal operator activity. Indeed, due to replay of out-of-time stream elements more and more emissions are carried out by the spout, with the result that pending queues remain full. On the contrary, the autoscale approach reacts in multiple stages to adapt the capacity of each operator to uctuations in input rate. Even the intermediate bolt, which has a very low latency, performs a scale-out as a precaution thanks to the global context. As a result of this adaptation, operators can consume their respective pending queues fast enough to benet from the decrease in input rate. autoscale adapts dynamically the parallelism degree. Finally, a reconguration is performed as soon as a new peak appears.

Application on the diamond topology

We reused same congurations for the diamond topology (see table 2).

For the conguration ConfMin, we observe that the default scheduler of Storm is not able to avoid the congestion as illustrated on gure 7a. As exposed for the linear topology, operators accumulate stream elements on their pending without being able to respect the maximal timeout.

So, stream elements are replayed indenitely which leads to complete congestion. autoscale detects a congestion risk before it becomes eective and performs scale-out on sink operator in order to adapt dynamically its parallelism degree.

With ConfExpt, autoscale decreases the parallelism of the sink operator after a short time because the input rate does not require an important parallelism degree. Then, autoscale When sudden increases of input rate are applied on the diamond topology, the default scheduler is not able to absorb the input stream and operators are completely congested. So, autoscale is able to performs scale-out to absorb the input stream almost as fast as it arrives. So, when the input rate decreases deeply between two peaks, autoscale performs scale-in to t dynamically and automatically capacity of operators to their processing needs.

Application on the star topology

For the star topology, we used parameters presented in table 3 to dene congurations ConfMin and ConfExpt. It is important to note that the average processing latency of sink operators is signicantly than it is for linear and diamond topologies. It relies on the fact that a star topology contains multiple sink operators so they receive their input load is divided by a factor of 3 in comparison to linear and star topologies. Concerning the star topology, a similar behavior is observed. It is worth noting that because of the structure of the star topology, autoscale is able to detect the risk of congestion (see gure 8a) before it actually happens. This can be done according to the awareness of the global context oered by autoscale. Finally, autoscale reduces resource usage around 24.6% as illustrated on gure 8b. It is less than linear and diamond because there are more slow operators. Indeed, slow operators tend naturally to be more often in high or critical activity than fast ones.

Results on advertising topology

We test our approach on an advertising topology mainly inspired from a topology used in [Peng et al., 2015] and available on Github18 to validate our approach in a real context. We essentially modify the source to be able to reproduce the same stream with dierent congurations and add two operators (ip projection and ip processor) to obtain a complex topology. Moreover, we apply the 3-step input stream illustrated on gure 4. This topology takes as input, logs representing an event linked to an advertisement on a web page. Each log is rst deserialized before being transmitted to an event lter. Two projection operators receive stream elements from this lter, one looking for user IP addresses and the other for information on the ad. A join with a static dataset is performed to link the ad to a promotion campaign. Finally, IP and campaign processors increase users and campaign counts to update a remote database. The main interest of this topology is the signicant selectivity of a lter operator (see gure 9), as this implies that a large increase in input rate will have a minor impact on nal operators even if they have large latency in comparison with other operators. We observe that even if the topology is not congested, the autoscale approach performs some scale-outs in order to adapt operator capacity to their respective input rates as illustrated on gure 10a. This is due to the combine strategy (see chapter 5 section 4), which takes into account the maximum between local and global estimations as the globally consistent one. Therefore, when a slow operator begins to accumulate some stream elements on its pending queue, the autoscale approach performs a scale-out to avoid congestion. Nevertheless, autoscale performs a similar throughput even if there are some unnecessary recongurations in one case.

We can estimate overheads induced by autoscale to 12% in comparison to actual needs in terms of CPU and memory requirements.

If a user bases his/her choice of parallelism degree exclusively on latencies, he/she will start the topology with some unnecessary executors (see gure 10b). The autoscale approach performs scale-in to t capacities of operators to their respective processing needs. It is important to notice that the dynamic adaptation made by autoscale, combined with the scheduler, allows all treatments to be collected on a single supervisor. With autoscale, Storm is able to handle biggest amount of data without generating network trac, which is a large overhead factor, as explained in [Xu et al., 2014], and using 50% less resources.

4 Evaluation of autoscale+ with osg

In this section, we evaluate the behavior of DABS on a dierent micro-benchmark composed of topologies sensitive and insensitive to stream element values. As a reminder, a topology is sensitive to stream element values if the processing latency is function of the value read in input.

As we have demonstrated the eectiveness of autoscale compared to the native solution of Storm, we suggest here implementations of INC and RL strategies (see section 1) to evaluate the performance of DABS in front of common auto-parallelization strategies integrated in other DSMSs. We assume that these implementations are not as optimized as they are in other systems but they have characteristic behaviors that we want to observe ad analyze.

Experimental protocol

We used the same cluster described in section 3.1 but extended the number of processing units to 10. The module managing the distribution of stream elements between executors implements the CustomStreamGrouping interface of Storm API. We also deploy a MySQL database on Nimbus to store monitoring data. We summarize main experimental parameters in table 4. To validate our approach, we demonstrate its eectiveness on three topologies. The simple insensitive topology (see gure 11a) composed of a spout (Source) emitting stream elements without ltering them. These stream elements are processed by a bolt (InsensitiveBolt) applying a function with a time complexity independent of the value read in input. So, streams are all even when played in input of this topology. Finally, a bolt (FinalizeBolt) ends the computation of each stream element by sending a termination signal to Storm monitor.

The simple sensitive topology (see gure 11b) has the same structure as the simple insensitive topology but the function applied by the intermediate bolt (SensitiveBolt) has a time complexity which depends directly from the value read in input.

The complex sensitive topology (see gure 11c) is composed of several operators with various selectivity factors and average processing latency. The spout (OpinionSource) emits stream elements concerning opinions submitted by users about a topic. Each opinion is described by information on the user, like its age and code representing its location, the topic and user opinion. Stream elements are sent to a bolt (CategoryDispatcher) ltering unnecessary attributes and depending on the branch downstream. In addition, it lters stream elements concerning a predened list of irrelevant topics. A branch starts with a bolt (SensitiveBolt) retrieving information on user location from the code. This bolt has the exact same properties than the sensitive bolt of the simple sensitive topology. Indeed, depending on the code, retrieving information on the city takes more or less time. It allows us to compare the impact of workow structure and complexity on bolt behavior and dynamic adaptation of its parallelism degree. Then, a bolt (CityAnalyzer) extracts relevant subgroups according to opinion and location. The other branch starting from the bolt CategoryDispatcher performs similar treatments in order to dene subgroups on user opinion and age. Finally, the Persister takes in input descriptions of subgroups and persists them in a storage le system. As illustrated on Figures 11a and11b, we build two synthetic streams with following common features: 1) at least one critical increase in input rate leading the system to congestion with a minimal (one executor per operator) and static conguration 2) decrease of input rate to evaluate the elasticity of the system. For each stream, we can set the distribution law. It can be uniform over all possibles values or biased according to a zipf law with a predened skew. These streams allow us to determine which impact has DABS while facing critical uctuations in both input rate and value distribution.

Results on simple insensitive topology 4.2.1 Simple insensitive topology in front of the progressive stream

As presented above, the simple insensitive topology has an average processing latency independent of input values. It exclusively depends of the volume of stream elements to process. For this reason, we decide to enable only autoscale+ for parallelism management and let the default grouping solution of Storm, denoted shue grouping, route stream elements to executors.

As presented above, the reinforcement learning strategy relies on a knowledge base such as it can associate a parallelism degree to an input rate. While processing the progressive stream (see gure 11a), the reinforcement learning strategy increases the parallelism degree of the operator InsensitiveBolt (see gure 11a). We notice that the parallelism degree decreases Figure 11: Simple insensitive topology with progressive stream signicantly as soon as the peak in input rate decreases. Nevertheless, most stream elements are just delayed by the processing rate and pending in input queue. So, the parallelism degree must be increased to avoid congestion. These modications have a major impact on average processing latency and result quality. Indeed, reconguring the system while large volumes of data are running between operators causes increases of average processing latency exceeding the maximal threshold. It is due to reconguration overheads including migrations of pending queues and activation/deactivation of tasks on machines. It has also an impact on result quality because 17% stream elements cannot be processed under the maximal threshold.

In comparison, the incremental strategy increases continuously the parallelism degree of the operator as long as long the workload exceeds the processing rate. By workload, we refer to the sum of incoming and pending stream elements. Even if the parallelism degree is increased, it cannot reach a suitable value to handle maximal uctuations in input rate. It implies large increases of average processing latency causing 29% losses of stream elements over the complete execution. In addition, in terms of resource usage, the incremental strategy requests 64% more active processing units than the reinforcement learning and 18% more than autoscale+.

While using autoscale+, Storm is able to anticipate suitable parallelism degrees over the complete execution. Even if autoscale+ tends to overestimate the required parallelism degree due to regression, reconguration overheads are compensate by benets on processing latency. Actually, the average processing latency remains stable over the complete execution reducing losses to 7%. It can also be observed on throughput as autoscale+ is able to maintain a throughput close to input rate with short time shift.

Simple insensitive topology in front of the erratic stream

We applied the erratic stream (see gure 11b page 117) in input of the simple insensitive topology to test the reactivity of each auto-parallelization strategy while facing sudden and large peaks in input rate. The reinforcement learning strategy increases and decreases the parallelism degree of the operator InsensitiveBolt according to two main peaks corresponding essentially to the increase in input rate happening at the end of the erratic stream. Indeed, brief increases in input rate does not imply important scale-out. They do not increase signicantly the average input rate on recent history so the parallelism degree does not require large increase of its parallelism degree. Nevertheless, the sudden accumulation of a large number of stream elements on pending queues increases the average processing latency. The impact on result quality remains negligible with only 0.6% of stream elements lost over the complete execution.

The incremental strategy benets from the short duration of peaks in input rate. Indeed, as the incremental strategy over-provisions the operator, resources necessary to handle brief increases in input rate are available. So, the average processing latency increases signicantly only when the input rate remains high for a long duration like it is happening at the end of the erratic stream. Losses of stream elements are reduced to 19% over the complete execution but the usage of processing units remains 85% higher than the reinforcement learning strategy and 4% higher than autoscale+.

Concerning autoscale+, the appearance of sudden increases in input rate has an impact on the regression model used to anticipate processing requirements. So, the parallelism degree of the operator is increased sooner causing a degradation of the average processing latency as the the input rate decreases immediately. autoscale+ overestimates processing requirements as the immediate decrease in input rate cannot be predicted. This overestimation allows to maintain a throughput close to the input rate and process the entire stream with 18% losses. As critical increase and decrease of input rate are sudden and brief, they cannot be anticipated and aect the processing latency before autoscale+ recongures the system. We present now same results with the simple sensitive topology. Input streams follow same uctuations in input rate but the distribution of values is biased. It follows a zipf distribution with a skew of 1,5 as used in [START_REF] Rivetti | Online scheduling for shue grouping in distributed stream processing systems research paper[END_REF]. To compensate the skew in value distribution, we used the OSG grouping solution for all congurations. So, auto-parallelization strategies are not penalized by imbalance between executors of same operator. As observed with the simple insensitive topology, autoscale+ anticipates processing requirements and is able to maintain a smaller processing latency while the stream is at its maximal rate. Nevertheless, the reinforcement learning strategy is able to decrease signicantly the processing latency when the input rate decreases. The incremental strategy is penalized by OSG because the frequent modication of parallelism degree forces OSG to reevaluate its routing policy. Indeed, as OSG tracks the load of each task associated to an operator, the frequent modication of parallelism degree involves frequent updates of the monitoring structure.

In terms of result quality, the incremental strategy is able to keep loss of stream elements at 17% but it requires 94% more active processing units than reinforcement learning strategy and 83% more than DABS (autoscale+ with OSG). The reinforcement learning strategy reacts to increase in input rate and cannot prevent 19% of losses while autoscale+ looses 20% of stream elements during the complete execution.

Concerning throughput, all solutions deliver close performance even if autoscale+ remains the auto-parallelization strategy keeping the smaller time shift between uctuation in input rate and throughput. As discussed above, the erratic stream has two short uctuations in input rate spaced by a signicant decrease. We notice that DABS is signicantly more accurate than autoscale+ alone. Indeed, the average processing latency remains low except two punctual increases during the rst two peaks in input rate and before the last increase in input rate which lasts longer (see gure 11b). DABS reduces losses of stream elements to 4.8% even if sudden increases cannot be anticipate. It is due to OSG which delivers fast routing when slightly overprovisionned in executors, what DABS does through regression. It makes the major dierence with the reinforcement learning strategy which provides only the suitable number of executors to avoid congestion. The incremental strategy maintains a low processing latency and delivers a throughput close to the input rate but it still uses considerably more resources to complete the treatment of the entire stream. Moreover, losses are more important than reinforcement learning solution and DABS and go up to 20% over the complete execution. It is also interesting to notice that DABS maintains a lower time shit between uctuations in input rate and throughput than the reinforcement learning solution. So, even if stream elements arrive at important rates, preventive reconguration performed by DABS does not delay their treatment. There are two signicant dierences between this experimental setup and ones presented above:

• The operator becoming critical is not directly connected to the source. Stream elements emitted by the source are ltered and transformed by other operators before reaching the operator. Its inputs may not present same uctuations than original ones.

• Operators processing stream elements upstream critical operators can be analyzed to evaluate consistency at workow scope. Indeed, simple topologies composed of a single operator cannot benet from analysis at workow scope because there is no complementary estimation to compare with local estimations.

While playing the progressive stream in input of the complex topology, DABS anticipates the rst scale-out compared to autoscale+ with the simple topology. It also maintains an important parallelism degree longer than autoscale+ does for the simple topology. According to these modications of parallelism degree, DABS maintains a lower processing latency than the reinforcement learning strategy and requests 30% less active resources than incremental strategy for equivalent performance.

In terms of result quality, DABS benets from the analysis at workow scope to improve anticipation of scale-out and to reduce losses to 10%. In comparison, reinforcement learning strategy reduces losses only to 27% and the incremental strategy to 18%. It is due to the accumulation of stream elements which has a greater impact of processing latency when multiple operators are aected downstream.

Except a time shift when the stream reaches its maximal input rate, all strategies maintain a throughput close to input rate. DABS is slightly more reactive than other auto-parallelization strategies and deliver a maximal throughput sooner. The complex structure of the topology makes an important dierence in the treatment of the erratic stream. For all auto-parallelization strategies, we can observe that brief but large uctuations in input rate do not aect signicantly the operator. These uctuations are absorbed by the processing rate of upstream operators and inter-operator transmissions.

In this context, DABS performs all scale-out before any scale-in. It allows the system to maintain a low latency even when it faces critical variations in input rate. The incremental strategy also keep a low processing latency but uses 36% more active processing units to achieve treatments. On its side, the reinforcement learning strategy keeps reacting to average input rate to adjust parallelism degree. It results in an inconsistent scale-in at workow scope which is contradicted afterwards. It has a major impact on processing latency and result quality. Actually, when DABS and the incremental strategy maintains losses at 13% and 19% respectively, the reinforcement learning strategy looses 26% of stream elements at the end of the observation.

Discussion

To sum up, we presented the evaluation of autoscale and autoscale+ approaches for autoparallelization of operators. They both present the advantage to avoid congestion of operators through an automatic of the parallelization. It appears that autoscale may involve some reconguration overheads when it overestimates the workload in near future. Nevertheless, it consistently reduces the usage of resources at runtime for equivalent performance.

Moreover, the experimental evaluation highlights the ability of autoscale+ to maintain a low processing latency compared to auto-parallelization strategies commonly used in DSMSs. In most cases, autoscale+ looses less stream elements than other auto-parallelization strategies while facing critical uctuations in input rate. The combination with the load balancing strategy osg improves the performance of the system while processing streams with high skew in data distribution.

validates or invalidates suggested local reconguration requirement according to modications validated upstream. Indeed, autoscale computes a global activity metric relying on the processing capacity estimated locally and a combination of input workloads estimated locally and upstream. It aims at triggering a consistent set of recongurations, including anticipated scale-out of operators. It improves system stability by avoiding antagonist scale-in and scale-out.

The execution model considered by autoscale does not take concurrency for resource usage and load imbalance into account. To bridge this gap, we extend autoscale to autoscale+ through the following improvements:

• The monitoring module collects the CPU time allocated to each task associated to an operator during the recent history. It allows to evaluate which CPU is eectively available for each task and take the concurrency between tasks into account.

• From these new metrics, we suggest a resource-aware variant of our global activity metric. This variant evaluates the processing capacity of a given operator according to available resources of machines running its associated tasks. So, each scale-in/out is triggered accurately with regards to resources of processing units.

• We modify the consistency checking at workow scope to perform both local and global estimations in one-pass. On complex topologies with several operators, it reduces the computation time of the set of recongurations globally consistent.

• We associate autoscale+ to the load balancing strategy OSG [START_REF] Rivetti | Online scheduling for shue grouping in distributed stream processing systems research paper[END_REF] such as the accuracy of scale-in/out is not biased by a skew in data distribution. Thus, this combined solution is able to adapt dynamically and automatically workows while facing critical uctuations in input rate and data distribution at the same time.

Experimental studies of autoscale and autoscale+ show respectively that consistency checking at workow improves signicantly the stability of the system and the association with a load balancing strategy serves as guarantee that even and uneven streams are processed similarly.

Concerning autoscale, experiments on a micro-benchmark shows that autoscale avoids congestion of workows when the input rate becomes critical. According to our experiments, autoscale is able to save up to 37,5% less CPU and memory resources and at least 12% for equivalent performance. While computing several continuous queries on the same cluster, it improves signicantly the scalability of the system.

Improvements included in autoscale+ let users dene more accurately their priorities in terms of performance and resource usage. With an appropriate conguration, autoscale+ is able to consume up to 18% less resources than a reactive auto-parallelization strategy based on reinforcement learning. Moreover, autoscale+ is able to limit losses to 7% while receiving streams at critical input rates.

Combining autoscale+ with the load balancing strategy osg signicantly improves the performance and usage of resources of the DSMS. Indeed, even when a biased stream in terms of distribution of values is received at rates changing suddenly, losses are limited to 10% and resource usage is reduced by 30% compared to common solutions for elastic stream processing implemented in the DSMS Apache Storm. We develop some auto-parallelization strategies as custom modules integrated into Apache Storm, including autoscale and autoscale+ and demonstrate the eectiveness of autoscale and autoscale+ approach in order to manage congestion of operators. We develop microbenchmarks in order to evaluate benets brought by these solutions while processing a variety of continuous queries over dierent streams. We give the possibility to users to reuse these workows and streams for further developments and works.

Perspectives

Contributions suggested during this research can be improved and extended in several directions to reach a fully elastic stream processing with a better synergy between adaptation levels presented in chapter 4.

Exploiting user prole

In chapter 5, we present a combine strategy which takes local and global workload estimations into account to return the considered estimation. We can oer the possibility to users to prioritize performance (i.e., performing scale-out as soon as it can prevent any increase of the end-to-end latency) or resource usage (i.e., performing scale-out only when congestion cannot be avoided and performing scale-in as soon as possible). Indeed, depending on the critical aspect of a continuous query and the nancial cost of computations, users may not have same priorities. The problem is then to integrate user prole within the auto-parallelization strategy. It raises several issues such as the identication of metrics describing user prole and the integration of these metrics in the decision algorithm.

Generalizing the resource-aware scheduler

As studied in chapter 6, autoscale+ takes advantage of the load balancing strategy OSG to perform accurate scale-in/out in presence of bias in data distribution. In the same idea, when a task suers from a lack of resources on its host (e.g., the scheduling strategy assigned several complex operators on the same machine), imbalance between processing capacities appears because of concurrency between threads. To compensate such kind of imbalance, it would be interesting to associate autoscale+ with a resource-aware scheduling strategy. Such strategies [Peng et al., 2015, Aniello et al., 2013] have already been investigated in the literature but they take as input static user constraints dening resource requirements for each operator. Even if it is assumed that users are experts knowing the time and space complexity of each operator, it remains particularly dicult to estimate resource requirements such as the ratio of allocated CPU time and memory space. Moreover, it induces that there is a minimal subset of available machines which will stay active even if the input stream is signicantly lower than the expected input rate (e.g., a road monitoring stream at dierent hours of a day). Instead of considering static user constraints, it would be interesting to consider user preferences which can be modied at runtime depending on eective resource usages. For example, let consider an operator associated to a preference of 20% on CPU usage. If this operator requires in average 50% at runtime, the scheduler will consider that each task applying this operator requires 50% instead of user preference. The main issue consists in the evaluation of processing requirements of each operator at runtime as it depends on several parameters like the type of the operator (stateless or stateful) and its sensitivity to values of stream elements. Another challenge is to evaluate when it is benecial to move an operator from a processing unit to another as resource usage (e.g., CPU usage) may vary quickly.

Introducing operator model for auto-parallelization

Auto-parallelization strategies suggested in this research focus on adapting the processing capacity to the estimated workload without considering the type of the operator (stateless or stateful). As stateful operators process stream elements micro-batch by micro-batch, evaluating the accurately workload requires to distinguish stream elements being processed and stream elements pending in input queue until the next micro-batch. More generally, as several DSMS support declarative languages for query denition, it could be interesting to exploit algebraic properties of predened operators (selectivity factor, time complexity) to improve the regression model used by autoscale. The problem is then to dene for each class of operators, specic features rening the regression model in order to anticipate congestion with a better accuracy.

Auto-parallelization on limited resources

During this research, we assumed that users submit their continuous queries to a cluster managed by a service provider. With the democratization of small single-board computers (e.g., Rapsberry Pi), establishing a cluster of multiple processing units located at dierent places becomes a viable solution according to the economical aspect. It allows to process data closer from stream sources [START_REF] Logothetis | In-situ mapreduce for log processing[END_REF] and have a control on distributed execution. Nevertheless, such solution is limited by the performance of each processing unit. Depending on the input rate and the complexity of continuous queries, treatments may be limited by resources [Yang et al., 2012]. To perform elastic stream processing, it is then necessary to prioritize some operators. The prioritization of scale-out has been studied in [START_REF] Xu | Stela: Enabling stream processing systems to scale-in and scale-out on-demand[END_REF] and has been presented as the impact of an operator on the throughput of a workow. Depending on the application, some operators may have a great impact on results in terms of quality without generating the greatest volume of outputs. It appears interesting to rene the notion of impact of an operator in a stream processing context. Thus, autoscale+ can be extended with a prioritization of operators combined to the resource-aware activity metric in order to maximize the result quality under resource constraints. It could also be extended to the preemption of resources by some operators even if it implies reducing reserved resources of other operators. In such context, autoscale+ could also be combined with a load shedding strategy to discard judiciously stream elements of non-priority operators. Dans le contexte de cette thèse, qui s'est réalisée dans le cadre du projet ANR Socioplug (ANR-13-INFR-0003), nous considérons une plateforme collaborative de traitement de flux de données. Chaque utilisateur peut soumettre des requêtes continues et contribue aux ressources de traitement de la plateforme. Cependant, chaque unité de traitement mise à disposition pour le traitement des requêtes dispose de ressources limitées en termes de processeur et de mémoire ce qui peut engendrer la congestion du système en fonction des variations des flux en entrée. Le problème est alors de savoir comment adapter dynamiquement les ressources utilisées par chaque requête continue par rapport aux besoins de traitement ? Cela soulève plusieurs défis : i) comment détecter un besoin de reconfiguration ? ii) quand reconfigurer le système pour éviter sa congestion ? iii) comment éviter des reconfigurations n'ajustant pas l'usage des ressources aux besoins des traitements ?

Durant ces travaux de thèse, nous nous sommes intéressés aux différentes étapes de traitement d'une requête continue sur une infrastructure distribuée. De cette analyse, nous avons pu identifier les limites de l'existant et les mécanismes permettant d'adapter dynamiquement les ressources utilisées pour l'exécution d'une requête continue. Nous avons focalisé nos efforts sur la gestion automatique de la parallélisation des opérateurs composant le plan d'exécution d'une requête. Nous proposons une approche originale basée sur l'observation des opérateurs et une estimation des besoins de traitement dans un futur proche. Ainsi, nous pouvons augmenter (scale-out) ou diminuer (scale-in) le niveau de parallélisme des opérateurs composant une requête continue de manière proactive afin d'ajuster les ressources utilisées aux besoins des traitements. Par rapport à une configuration statique définie par un expert, nous montrons qu'il est possible à la fois d'éviter la congestion du système dans certains cas ou de la retarder dans les cas les plus critiques. Nous montrons également qu'il est possible de réduire significativement la consommation de ressources tout en maintenant une performance et une qualité des résultats équivalentes.

Nous proposons également de combiner cette approche avec des mécanismes complémentaires tels que l'équilibrage de charge pour l'adaptation dynamique de requêtes continues. Ces différents travaux ont été implémentés et validés dans un SGFD largement utilisé avec différents jeux de tests reproductibles.

MOTS-CLÉS : flux de données, requête continue, traitement distribué, adaptation dynamique

  de parallélisme des opérateurs pour fournir les ressources processeur et mémoire nécessaires aux traitements. De plus, le placement des tâches sur les unités de traitement peut impacter le débit entre tâches successives. Au niveau de l'implémentation, le choix d'une implémentation au détriment d'une autre peut avoir un impact important sur la latence de traitement et donc sur le débit de traitement. En fonction de la distribution des valeurs et de la fonction appliquée par l'opérateur, il peut être nécessaire d'équilibrer la charge de données à traiter entre les diérentes tâches. Cela consiste à partitionner les données entrantes entre toutes les tâches associées à un même opérateur. Enn, lorsque le volume de données est trop important, des stratégies d'échantillonnage permettent de réduire le débit d'entrée tout en maitrisant la dégradation de la qualité des résultats. À chaque niveau, les mécanismes élastiques peuvent être déclenchés soit sur demande de l'utilisateur, soit de manière automatique. Le déclenchement manuel requiert une expertise et une observation constante de la part de l'utilisateur an d'améliorer les performances d'une requête continue. Le déclenchement automatique permet d'ajuster les ressources allouées aux besoins des traitements, toutefois, la réactivité de la méthode dépend des métriques considérées pour identier des opérateurs congestionnés. An de mettre en évidence les diérences entre les méthodes de détection d'opérateurs congestionnés, nous proposons de les regrouper en catégories. Ces catégories sont basées sur la capacité d'anticipation de ces méthodes et sur l'algorithme de prise de décision, i.e. quand une augmentation ou une diminution du degré de parallélisme doit être eectuée. 2.3 Classication des SGFD Suite à l'étude bibliographique, nous avons identié une grande variété de langages de requêtes et de méthodes de gestion d'opérateurs congestionnés. Nous suggérons alors une classication de SGFD basée sur la représentation des requêtes continues (graphes d'opérateurs ou jobs MapReduce), le type de langage de requêtes supporté et la gestion des opérateurs congestionnés. Pour chacun des systèmes sélectionnés, nous analysons chaque critère et expliquons comment une requête est gérée de sa déclaration à la gestion d'opérateurs congestionnés pendant l'exécution. Les SGFD analysés et intégrés dans cette classication ont été sélectionné pour leur performance et leur représentativité des solutions existantes. Cette classication met en évidence la grande diversité des solutions mais également l'absence de solutions prenant en compte tous les aspects nécessaires à la gestion dynamique d'opérateurs congestionnés. 3 Traitement élastique de ux de données Diérents facteurs interviennent dans l'apparition d'une congestion. En eet, entre la requête continue dénie par l'utilisateur et les tâches distribuées sur un ensemble d'unités de traitement interconnectées, plusieurs phases d'optimisation sont appliquées. Les choix eectués par le SGFD durant chacune de ces phases peut causer l'apparition d'un goulot d'étranglement au niveau du réseau ou des traitements.

3. 4

 4 Optimisation locale Certains SGFD proposent un ensemble d'opérateurs logiques pour la dénition de requêtes continues. Pour chaque opérateur logique, plusieurs implémentations peuvent être disponibles an d'adapter localement les traitements en fonction du contexte d'exécution. Par exemple, pour un opérateur de jointure, un SGFD peut disposer d'une implémentation basée sur un algorithme de hachage et une autre implémentation sur des boucles liées. Selon la taille des entrées une implémentation est plus performante que l'autre.

4. 2

 2 Analyse des recongurations à l'échelle du graphe d'opérateurs 4.3 Évaluation expérimentale d'autoscale An d'évaluer l'approche autoscale, nous avons choisi de la comparer avec le comportement natif de la solution Apache Storm selon diérentes congurations. Apache Storm permet aux utilisateurs de dénir le degré de parallélisme de chaque opérateur à l'initialisation de la requête continue. Par défaut, ces degrés initiaux ne sont pas modiés par Storm. Nous avons proposé un micro-benchmark composé de quatre requêtes continues ou topologies : Trois topologies simples : linéaire, diamant et en étoile. Ces topologies sont dites élémentaires car la majorité des topologies peuvent être découpées selon ces motifs. Chacune de ces topologies est composée d'une ou plusieurs sources, d'au moins un opérateur à forte latence de traitement et d'au moins un opérateur à faible latence de traitement. Une topologie complexe inspirée par un cas d'usage réel. Cette topologie possède plusieurs opérateurs à forte et faible latence de traitement.

4 3 -

 3 step and 5-step streams . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 5 Experimental results for the Linear topology . . . . . . . . . . . . . . . . . . . . . 6 Comparison between Storm (Default) and autoscale for the Linear topology in front of the 5-step stream with ConfMin. . . . . . . . . . . . . . . . . . . . . . . . 7 Experimental results for the Diamond topology . . . . . . . . . . . . . . . . . . . 8 Experimental results for the Star topology . . . . . . . . . . . . . . . . . . . . . . 9 Advertising topology for stream benchmarking . . . . . . . . . . . . . . . . . . . . 10 Experimental results for the Advertising topology . . . . . . . . . . . . . . . . . . 11 Simple insensitive topology with progressive stream . . . . . . . . . . . . . . . . . 12 Simple insensitive topology with erratic stream . . . . . . . . . . . . . . . . . . . 13 Simple sensitive topology with progressive stream . . . . . . . . . . . . . . . . . . 14 Simple sensitive topology with erratic stream . . . . . . . . . . . . . . . . . . . . 15 Complex sensitive topology with progressive stream . . . . . . . . . . . . . . . . . 16 Complex sensitive topology with erratic stream . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1 2 Problem Statement . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 3 3 Contributions and Organization . . . . . . . . . . . . . . . . . . . . . . 5 1 Stream Processing

Figure 1 :

 1 Figure 1: A stream with variations in input rate and distribution of values

Figure 2 :

 2 Figure 2: Stream types according to rate

Figure 3 :

 3 Figure 3: Two iterations of a computation window

Figure 4 :

 4 Figure 4: A MapReduce job

Figure 1 :

 1 Figure 1: Operator reordering

  Figure 2: Operator parallelization

Figure 3 :

 3 Figure 3: Task scheduling

Figure 4 :

 4 Figure 4: Algorithm selection

  application receiving vehicle speeds observed by radars. An operator O 1 takes as input speeds violating the limit with the associate registration license. O 1 queries a remote database depending on registration license to extract driver information. Each remote database has a specic communication latency. If the load in input of O 1 becomes critical, it can be 1. Congestion management

Figure 5 :

 5 Figure 5: Load balancing

Figure 6 :

 6 Figure 6: Load shedding

Figure 7 :

 7 Figure 7: Classication of distributed DSMSs

  Window support: Basically, ESC supports count-based tumbling windows. The order is based on arrival time making stateful operators not deterministic as window content may vary randomly. The management of other types of windows is delegated to user implementation. Congestion management: ESC supports exclusively adaptation of continuous queries at operator level. Congestion is managed by a reinforcement learning-based algorithm in ESC. An autonomic manager implementing a Monitor-Analyze-Plan-Execute (MAPE) loop which enriches a knowledge base through interactions with ESC platform.

  Query representation: A continuous query is represented as a DAG of MapReduce jobs running in pipeline. C-MR benets from optimization of MapReduce jobs like sorting and merging between Map and Reduce phases. Note that Map and Reduce phases are executed asynchronously in C-MR to process a microbatch of stream elements as soon as it arrives. Window support: C-MR supports time-based sliding and tumbling windows. Windows are declared directly as parameters of Map and Reduce operators. They are implemented through punctuations (see section chapter 2 2.1.1) to maintain chronological order within streams. C-MR takes advantage of sliding windows to compute incrementally results like Flink. Intermediate results are computed by Combine operators which keep updated results over sliding windows.Congestion management: C-MR supports the adaptation of continuous queries at operator level. C-MR performs operator scheduling according to anities between operators. As a continuous is represented by a DAG of MapReduce jobs processing stream elements in pipeline, C-MR collocate operators exchanging highest amount of data to limit network bottlenecks.

  Dataow API as sequences of ParDo/GroupByKey pairs. Query representation: Continuous queries are represented as pipelines of MapReduce jobs composed of ParDo and GroupByKey operators. According to query denition, Dataow can merge ParDo functions on a single operator if they share same keys.Window support: To apply windowing on stream elements, Dataow enriches the key/value data model with two additional metadata: event time and window. Event time corresponds to the arrival date of a stream element and window refers to the window it belongs for GroupByKey operators.

Figure 8 :

 8 Figure 8: Classication of distributed DSMSs

4 A

 4 generic framework for elastic stream processing: a global picture

Figure 1 :

 1 Figure 1: ORACL loop

Figure 2 :

 2 Figure 2: Order step

Figure 3 :

 3 Figure 3: Replicate step

Figure 4 :

 4 Figure 4: Assign step

Figure 5 :

 5 Figure 5: Custom Locally step

Figure 2 :

 2 Figure 2: Monitoring window

Figure 4 :

 4 Figure 4: Estimated number of received stream elements over F i j

Figure 5 :

 5 Figure 5: An example of inconsistent reconguration at workow scope

Figure 7 :

 7 Figure 7: Potential inconsistent case 2

Figure 8 :

 8 Figure 8: Potential inconsistent case 3

Figure 9 :

 9 Figure 9: An example of IGLA

Figure 10 :

 10 Figure 10: Estimations at local and global scope

Figure 11 :

 11 Figure 11: Stream uctuations in input rate

Figure 12 :

 12 Figure 12: Heterogeneous workow

Figure 13 :

 13 Figure 13: Modication of parallelism degree

Figure 14 :

 14 Figure 14: Impact of consistency checking strategy on performance and quality

  1) where par(O i ) returns all parent operators of O i • To improve the accuracy of the regression model, autoscale+ applies linear, logarithmic and exponential regression models and selects the model tting the best to the previous iteration of the window. It implies light computation overheads but allow to detect characteristic stream uctuations with improved accuracy.

Figure 1 :

 1 Figure 1: Metrics

Figure 2 :

 2 Figure 2: Usable CPU for threads on one core

Figure 3 :

 3 Figure 3: Working interval

Figure 4 :

 4 Figure 4: Modication of parallelism degree

Figure 6 :

 6 Figure 6: Complex sensitive topology

Figure 7 :

 7 Figure 7: Fluctuations in input rate of synthetic streams

Figure 8 :

 8 Figure 8: Comparison between static parallelization of operators and autoscale+

Figure 9 :

 9 Figure 9: Comparison between Round-Robin and OSG

Figure 10 :

 10 Figure 10: Comparison between static parallelization of operators and autoscale+

Figure 11 :

 11 Figure 11: Comparison between Round-Robin and OSG

  reconguration overheads. Discussion . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 124 1 Overview of solutions

Figure 1 :

 1 Figure 1: Experimented solutions

2

  Design and implementation of autoscale and autoscale+2.1 Overview of Apache StormApache Storm is an open-source SPE, allowing users to dene continuous queries as graphs of operators, called topologies. Users dene each operator in a high-level programming language such as Java, Python or Clojure.

Figure 2 :

 2 Figure 2: Storm architecture

Figure 3 :

 3 Figure 3: autoscale architecture

Figure 4 : 3 -

 43 Figure 4: 3-step and 5-step streams

  (a) Comparison between Storm (Default) and autoscale for the Linear topology in front of the 3-step stream with ConfMin.

  (b) Comparison between Storm (Default) and autoscale for the Linear topology in front of the 3-step stream with ConfExpt.

Figure 5 :

 5 Figure 5: Experimental results for the Linear topology

Figure 6 :

 6 Figure 6: Comparison between Storm (Default) and autoscale for the Linear topology in front of the 5-step stream with ConfMin.

  (a) Comparison between Storm (Default) and autoscale for the Diamond topology in front of the 3-step stream with ConfMin. (b) Comparison between Storm (Default) and autoscale for the Diamond topology in front of the 3-step stream with ConfExpt.

Figure 7 :

 7 Figure 7: Experimental results for the Diamond topology

  (a) Comparison between Storm (Default) and autoscale for the Star topology in front of the 3-step stream with ConfMin. (b) Comparison between Storm (Default) and autoscale for the Star topology in front of the 3-step stream with ConfExpt.

Figure 8 :

 8 Figure 8: Experimental results for the Star topology

Figure 9 :

 9 Figure 9: Advertising topology for stream benchmarking

Figure 10 :

 10 Figure 10: Experimental results for the Advertising topology

Figure 12 :

 12 Figure 12: Simple insensitive topology with erratic stream

4. 3

 3 Results on simple sensitive topology 4.3.1 Simple sensitive topology in front of the progressive stream

Figure 13 :

 13 Figure 13: Simple sensitive topology with progressive stream

4. 3 . 2

 32 Simple sensitive topology in front of the erratic stream

Figure 14 :

 14 Figure 14: Simple sensitive topology with erratic stream

4. 4

 4 Results on complex sensitive topology 4.4.1 Complex sensitive topology in front of the progressive streamAfter testing the behavior of auto-parallelization strategies in front of simple topologies, we applied same biased streams in input of a complex topology (see gure 11c). This topology is more representative of real-world continuous queries. It includes common operators as lters on values and attributes, joins with static bases and also user-dened functions from expert domains like data mining.

Figure 15 :

 15 Figure 15: Complex sensitive topology with progressive stream

4. 4 . 2

 42 Complex sensitive topology in front of the erratic stream

Figure 16 :

 16 Figure 16: Complex sensitive topology with erratic stream

  'UNIVERSITE DE LYON OPEREE AU SEIN DE L'INSA LYON NOM : KOTTO KOMBI DATE de SOUTENANCE : 29/06/2018 (avec précision du nom de jeune fille, le cas échéant) Prénoms : Roland Olivier TITRE : Distributed query processing over fluctuating streams NATURE : Doctorat Numéro d'ordre : 2018LYSEI050 Ecole doctorale : INFOMATHS Spécialité :Informatique RESUME : Le traitement de flux de données est au coeur des problématiques actuelles liées au Big Data. Face à de grandes quantités de données (Volume) accessibles de manière éphémère (Vélocité), des solutions spécifiques tels que les systèmes de gestion de flux de données (SGFD) ont été développés. Ces SGFD prennent en entrée des flux et des requêtes continues pour générer de nouveaux résultats aussi longtemps que des données arrivent en entrée. Dans certains domaines, les flux considérés ont des débits qui varient en termes de nombre de données produites par unité de temps ou en termes de distribution de valeurs des données. Ces variations peuvent impacter fortement les besoins en ressources nécessaires au traitement des requêtes continues.

1

  The ORACL loop . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1.1 Steps for query optimization . . . . . . . . . . . . . . . . . . . . . . . . . . 1.2 Adaptation levels . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1.3 Optimization strategies . . . . . . . . . . . . . . . . . . . . . . . . . . . . 2 Orchestration of optimization . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 2.1 Adaptation triggers . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 2.2 Challenges and dependencies between adaptation levels . . . . . . . . . . 3 Discussion . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . Enhancement of workload estimation . . . . . . . . . . . . . . . . . . . . . 2.2 Estimation of available resources . . . . . . . . . . . . . . . . . . . . . . . 2.3 Balance between processing requirements and resources . . . . . . . . . . 3 Load management . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . Compatibility issues . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 4 Discussion . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . Empirical study of the combination autoscale+ and OSG . . . . . . . . Limits of autoscale+ . . . . . . . . . . . . . . . . . . . . . . . . . . . . variations in input rate and distribution of values . . . . . . . . . 2 Stream types according to rate . . . . . . . . . . . . . . . . . . . . . . . . . . . . 3 Two iterations of a computation window . . . . . . . . . . . . . . . . . . . . . . . Estimated number of received stream elements over F i j . . . . . . . . . . . . . . . 5 An example of inconsistent reconguration at workow scope . . . . . . . . . . . 6 Potential inconsistent case 1 . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 7 Potential inconsistent case 2 . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 8 Potential inconsistent case 3 . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 9 An example of IGLA . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 10 Estimations at local and global scope . . . . . . . . . . . . . . . . . . . . . . . . . 11 Stream uctuations in input rate . . . . . . . . . . . . . . . . . . . . . . . . . . . 12 Heterogeneous workow . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 13 Modication of parallelism degree . . . . . . . . . . . . . . . . . . . . . . . . . . . 14 Impact of consistency checking strategy on performance and quality . . . . . . . Working interval . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 4 Modication of parallelism degree . . . . . . . . . . . . . . . . . . . . . . . . . . . 5 Worst and optimal cases of load balancing . . . . . . . . . . . . . . . . . . . . . . 6 Complex sensitive topology . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 7 Fluctuations in input rate of synthetic streams . . . . . . . . . . . . . . . . . . . 8 Comparison between static parallelization of operators and autoscale+ . . . . 9 Comparison between Round-Robin and OSG . . . . . . . . . . . . . . . . . . . . 10 Comparison between static parallelization of operators and autoscale+ . . . . 11 Comparison between Round-Robin and OSG . . . . . . . . . . . . . . . . . . . . 1 Experimented solutions . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 2 Storm architecture . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 3 autoscale architecture . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .

		2.1
		3.1 3.2	Auto-parallelization of operator with load imbalance . . . . . . . . . . . . List of Figures
		4.1	Online Shue Grouping for resource-aware load balancing . . . . . . . . .
	1	4.2 A stream with
		4.3
	Chapter 5 Preventive auto-parallelization approach for elastic stream process-ing Chapter 7 Experiments

1 Execution context . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1.1 Assumptions . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1.2 Challenges . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1.3 Overview of the autoscale approach . . . . . . . . . . . . . . . . . . . . 1 Overview of solutions . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 2 Design and implementation of autoscale and autoscale+ . . . . . . . . . . . 2.1 Overview of Apache Storm . . . . . . . . . . . . . . . . . . . . . . . . . . 2.2 Implementation of autoscale . . . . . . . . . . . . . . . . . . . . . . . . 3 Evaluation of autoscale . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 3.1 Experimental protocol . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 3.2 Results on the microbenchmark . . . . . . . . . . . . . . . . . . . . . . . . 3.3 Results on advertising topology . . . . . . . . . . . . . . . . . . . . . . . . 4 Evaluation of autoscale+ with osg . . . . . . . . . . . . . . . . . . . . . . . . 4.1 Experimental protocol . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 4.2 Results on simple insensitive topology . . . . . . . . . . . . . . . . . . . . 4.3 Results on simple sensitive topology . . . . . . . . . . . . . . . . . . . . . 4.4 Results on complex sensitive topology . . . . . . . . . . . . . . . . . . . . 5 Discussion . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 4 A MapReduce job . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 5 Distributed stream processing . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1 Operator reordering . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 2 Operator parallelization . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 3 Task scheduling . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 4 Algorithm selection . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 5 Load balancing . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 6 Load shedding . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 7 Classication of distributed DSMSs . . . . . . . . . . . . . . . . . . . . . . . . . . 8 Classication of distributed DSMSs . . . . . . . . . . . . . . . . . . . . . . . . . . 1 ORACL loop . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 2 Order step . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 3 Replicate step . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 4 Assign step . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 5 Custom Locally step . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1 Operator model . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 2 Monitoring window . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 3 Recent history F i j-1 at the start of iteration F i j . . . . . . . . . . . . . . . . . . . 4 1 Metrics . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 2 Usable CPU for threads on one core . . . . . . . . . . . . . . . . . . . . . . . . . ix

3

  State of the art

Contents 1 Congestion management . . . . . . . . . . . . . . . . . . . . . . . . . . 25 1.1 Principle . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 25 1.2 Adaptation at workow level . . . . . . . . . . . . . . . . . . . . . . . . 26 1.3 Adaptation at operator level . . . . . . . . . . . . . . . . . . . . . . . . 28 1.4 Adaptation at implementation level . . . . . . . . . . . . . . . . . . . . 31 1.5 Adaptation at data level . . . . . . . . . . . . . . . . . . . . . . . . . . . 32 2 Detection methods for congestion management . . . . . . . . . . . . 35 2.1 On user-demand methods . . . . . . . . . . . . . . . . . . . . . . . . . . 35 2.2 Automatic methods . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 36 3 Classication of distributed DSMSs . . . . . . . . . . . . . . . . . . . 40 3.1 Criteria of classication . . . . . . . . . . . . . . . . . . . . . . . . . . . 40 3.2 Workow-based solutions . . . . . . . . . . . . . . . . . . . . . . . . . . 41 3.3 MapReduce-based solutions . . . . . . . . . . . . . . . . . . . . . . . . . 51 4 Discussion . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 57

Table 1 :

 1 ].

	Level	Pattern	Required parallelism
	Workow	Operator reordering	task parallelism
	Operator	Operator parallelization	data parallelism
	Operator	Task scheduling	none
	Implementation	Algorithm selection	none
	Data	Load balancing	data parallelism
	Data	Load shedding	none

Patterns for congestion management

As presented in Table

  1. Congestion management price values for each stock label. Splitting such stateful operator consists in partitioning stock labels among some tasks so states are completely disjoint. Principle Denition 11. (Operator parallelization) An operator O i is parallelization if it is replaced by a set of equivalent tasks {T 1 i ,...,T k i } processing stream elements in parallel. Each task receives in input a partition of O i inputs.

Table 2 :

 2 Summary of workow-based DSMSs 50 3.3 MapReduce-based solutions All solutions based on the MapReduce framework (see chapter 2 section 2.1.4) support the key/value data model with possible extensions. By default, we consider in the section that all approaches represent continuous queries as pipelines of MapReduce jobs except if mentioned dierently. Backman et al., 2012] is a DSMS enabling the pipeline execution of MapReduce jobs over unbounded data streams. C-MR extends the MapReduce framework to support window denition and ensures order preservation between parallel nodes processing partitions of a global input stream. Query denition: C-MR extends MapReduce framework with window denition. Continuous queries are dened as standard MapReduce jobs (see chapter 2 section 2.1.4). Map and Reduce operators are enriched with windowing parameters (size and slide).

	3.3.1 C-MR
	Continuous-MapReduce(C-MR) [

  Continuous queries correspond to MapReduce jobs processing stream elements in pipeline like C-MR and iMR applications. Data are grouped and sent in Resilient Distributed Datasets (RDDs)[Zaharia et al., 2012a]. RDDs rely on the same principle presented for Flink. As it assumed that Spark Streaming processes huge volume of data, global throughput is signicantly improved if data are buered before being sent between operators. It optimizes the execution of Spark applications to t most requirements induced by a stream context. Operator scheduling is performed according to a threshold-based algorithm which detects overload in entry of operators and move them in order to avoid computation bottlenecks on some processing units. It does not take network overheads into account.

	Query representation: Window support: Spark Streaming supports time-based and count-based sliding and tum-
	bling windows. Windows are declared through Spark Streaming API potentially for each op-
	erator. Size of RDDs is adapted to window size and slide in order to perform incremental
	computations.
	Congestion management: Spark Streaming only supports adaptation of continuous queries
	at operator level.
	Spark Streaming benets from Spark APIs to query relational data or use domain specic oper-
	ators (e.g., machine learning methods). Each operator Map and Reduce is declared with input
	and output streams.

  Overview of the autoscale approach . . . . . . . . . . . . . . . . . . . Empirical study of consistency checking . . . . . . . . . . . . . . . . . . 6.2 Estimation of capacity . . . . . . . . . . . . . . . . . . . . . . . . . . . .

Contents 1 Execution context . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1.1 Assumptions . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1.2 Challenges . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1.3 2 Monitoring management . . . . . . . . . . . . . . . . . . . . . . . . . . 2.1 Operator model . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 2.2 Formalization . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 3 Detection of reconguration needs . . . . . . . . . . . . . . . . . . . . 3.1 Estimation of input load in near future . . . . . . . . . . . . . . . . . . 3.2 Estimation of processing capacity in near future . . . . . . . . . . . . . 3.3 Identication of potential congestion at operator scope . . . . . . . . . . 4 Consistency at workow scope . . . . . . . . . . . . . . . . . . . . . . 4.1 Construction of the instantaneous graph of local activities . . . . . . . . 4.2 Evaluation of reconguration impact . . . . . . . . . . . . . . . . . . . . 4.3 Consistency checking at workow scope . . . . . . . . . . . . . . . . . . 5 Quantication of reconguration . . . . . . . . . . . . . . . . . . . . . 6 Discussion . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 6.1

Table 1 :

 1 Decision Matrix for IGLA computation

	H H H H H H Operator activity GAL F i j < θ min Evolution trend of inputs Decreasing or H H H constant H H H H scale-in	Increasing nothing

  .

	H H H H H Evolution H H H Operator H H H activity trend in input rate H H Decreasing or stable	Low activity scale-in	Medium activity nothing	High activity nothing	Critical activity scale-out
	Increasing	nothing	nothing	scale-out scale-out

Table 2 :

 2 Local decision matrix for reconguration evaluation

Table 3 :

 3 Decision matrix for global consistency

  Motivation . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 91 Limits of autoscale+ . . . . . . . . . . . . . . . . . . . . . . . . . . . 104

			6
		Preventive auto-parallelization
			enhancements
	Contents
	1
		1.1	Impact of thread concurrency on operator capacity . . . . . . . . . . . . 91
		1.2	Managing load between tasks in presence of uneven streams . . . . . . . 92
	2	Resource-aware auto-parallelization of operators . . . . . . . . . . . 92
		2.1	Enhancement of workload estimation . . . . . . . . . . . . . . . . . . . . 93
		2.2	Estimation of available resources . . . . . . . . . . . . . . . . . . . . . . 94
		2.3	Balance between processing requirements and resources . . . . . . . . . 95
	3	Load management . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 97
		3.1	Auto-parallelization of operator with load imbalance . . . . . . . . . . . 97
		3.2	Compatibility issues . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 98
	4	Discussion . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 98
		4.1 4.2	Online Shue Grouping for resource-aware load balancing . . . . . . . . 99 Empirical study of the combination autoscale+ and OSG . . . . . . . 99
		4.3
	1 Motivation
	1.1 Impact of thread concurrency on operator capacity

Table 1 :

 1 Parameters for autoscale+ and OSG Finally, we analyze the eect on resource usage through the cumulative CPU usage of the critical operator. Parameters for autoscale+ and OSG congurations are summarized in table 1:

	Size of monitoring window (in s) 90
	Monitoring frequency (in s)	90
	α (autoscale+)	0.3
	β (autoscale+)	0.7
	θ (osg)	0.05
	(osg)	0.05

  .1.3.

	Table 1: Control parameters
	window size	60s
	monitoring frequency	10s
	θ min (autoscale)	0.3
	θ max (autoscale)	0.8
	processing timeout (autoscale) 30s
	combine strategy (autoscale) max

Table 2 :

 2

	Conguration of operators for Linear topology
		intermediate	sink
	average latency	2ms	80ms
	min degree	1	1
	expert degree	1	8
	max degree	8	8
	CPU reservation	20.0	80.0
	memory reservation	256Mb	512Mb

Table 3

 3

	: Conguration of operators for Star topology
		intermediate	sink
	average latency	2ms	240ms
	min degree	1	1
	expert degree	1	8
	max degree	8	8
	CPU reservation	20.0	30.0
	memory reservation	256Mb	512Mb

Table 4 :

 4 Main

	parameters
	window size	90s
	monitoring frequency	10s
	processing timeout	30s
	α (autoscale+)	0.3
	β (autoscale+)	0.8
	combine strategy (autoscale+) max
	θ (osg)	0.05
	(osg)	0.05

Cette thèse est accessible à l'adresse : http://theses.insa-lyon.fr/publication/2018LYSEI050/these.pdf © [R. Kotto Kombi],[2018], INSA Lyon, tous droits réservés

Grâce à la métrique d'activité, il est alors possible de déterminer si un opérateur nécessite une augmentation, une diminution ou la conservation de son degré de parallélisme courant. Toutefois, une prise de décision basée sur des estimations locales peut s'avérer inappropriée du point de vue de l'état global d'une topologie.En eet, la modication du degré de parallélisme d'un opérateur peut impacter fortement la quantité de données en entrée des opérateurs en aval de celui-ci. Il serait donc importun de prendre la décision locale de réduire le degré de parallélisme d'un opérateur suite à l'observation d'une faible activité si le degré de parallélisme de l'opérateur en amont est fortement modié.Cette analyse globale favorise la stabilité du système en évitant de déclencher des recongurations qui seront contredites à court terme mais également en évitant des recongurations antagonistes qui dégradant les performances du SGFD au lieu de les améliorer. Enn, cette analyse globale permet au SGFD de disposer de deux estimations de l'activité d'un opérateur et de choisir soit une stabilité accrue en ne modiant le degré de parallélisme que lorsque les deux estimations le suggèrent soit une réactivité accrue en modiant le degré de parallélisme dès la détection d'un risque de congestion.

http://socioplug.univ-nantes.fr/index.php/SocioPlug_Project

http://storm.apache.org/

http://infolab.stanford.edu/stream/sqr/

http://www.nasdaq.com/quotes/real-time.aspx

http://sqlstream.com

https://spark.apache.org/docs/latest/sql-programming-guide.html

https://spark.apache.org/ 16 Cette thèse est accessible à l'adresse : http://theses.insa-lyon.fr/publication/2018LYSEI050/these.pdf © [R. Kotto Kombi], [2018], INSA Lyon, tous droits réservés

Execution context motivated by ANR project Socioplug (ANR-13-INFR-0003), http://socioplug.univnantes.fr/index.php/SocioPlug_Project

https://aws.amazon.com/fr/ec2/

https://www.vmware.com/fr/products/esxi-and-esx.html

https://zookeeper.apache.org/

http://storm.apache.org/

https://kafka.apache.org/

https://spark.apache.org/ Cette thèse est accessible à l'adresse : http://theses.insa-lyon.fr/publication/2018LYSEI050/these.pdf © [R. Kotto Kombi], [2018], INSA Lyon, tous droits réservés

https://cloud.google.com/dataow Cette thèse est accessible à l'adresse : http://theses.insa-lyon.fr/publication/2018LYSEI050/these.pdf © [R. Kotto Kombi], [2018], INSA Lyon, tous droits réservés

https://storm.apache.org/ Cette thèse est accessible à l'adresse : http://theses.insa-lyon.fr/publication/2018LYSEI050/these.pdf © [R. Kotto Kombi], [2018], INSA Lyon, tous droits réservés

https://perso.liris.cnrs.fr/roland.kotto-kombi/autoscale/v2/ Cette thèse est accessible à l'adresse : http://theses.insa-lyon.fr/publication/2018LYSEI050/these.pdf © [R. Kotto Kombi], [2018], INSA Lyon, tous droits réservés

https://github.com/yahoo/streaming-benchmarks/ Cette thèse est accessible à l'adresse : http://theses.insa-lyon.fr/publication/2018LYSEI050/these.pdf © [R. Kotto Kombi], [2018], INSA Lyon, tous droits réservés

Acknowledgments

I would like to express my sincere gratitude to my supervisors Philippe Lamarre and Nicolas Lumineau. During these last years, they have pushed me, with great benevolence, to develop and exploit my own abilities. They devoted more time than I could expect to teach me the rigor of scientic research. Working with them was one of the most valuable experience for my own progression as a scientist and a person.

I want to thank warmly all members of the BD team. Mohand-Saïd Hacid, Jean-Marc Petit, Emmanuel Coquery and Romuald Thion for their precious advises and support during my Master degree and my thesis. I also thank all other members for welcoming me into this research team. I would also like to thank all members of the Socioplug project, especially Yann Busnel, for their valuable feedback on my works. Moreover, I want to thank Yves Caniou for our enriching discussions and Marc Plantevit for initiating me to research and supervising my rst works.

I would like to express my gratitude to Franck Morvan and Bernd Amann for devoting time to review this thesis and make valuable comments about my works. I also want to thank Hala Skaf-Molli and Bruno Defude for their participation to the presentation.

I thank my parents for their constant condence and support during my studies and to push me to do what I like. I thank my sisters and my brothers-in-law for their help and advises during those years. I address a special thank to my girlfriend, for her immeasurable thoughtfulness and support even when I was working several sunday nights in a row. I thank all my precious friends for their kindness and to make me explain my works regularly. It would have never be possible without all of them around me.

Finally, I thank all members of the University Claude Bernard and of the INSA de Lyon I have met for making those places such great workplaces. i

The focus of this research is congestion management. As presented in chapter 3, congestion may be due to dierent reasons. For a given operator belonging to a continuous query, an input rate signicantly greater than the processing rate induce an accumulation of stream elements on the input queue. This accumulation increases the time spent by each stream element in the input queue up to saturation. In such case, exploiting data parallelism can be a solution. Indeed, duplicating an operator increases its overall processing rate.

To answer this problem, users may set maximal parallelism degree for each operator. It ensures that all available resources are exploited to process streams. Nevertheless, this solution is costly as it requires to consume resources on all available machines. In a Green IT context and from a economical point of view, this solution implies major waste of resources. Thus, we suggest the auto-parallelization strategy autoscale which performs scale-in and scale-out according to uctuations in input rate. It adapts usage of resources to processing requirements avoiding congestion of operators and overconsumption of resources at the same time. Each reconguration is triggered after the following steps:

• At operator scope, a monitoring module collects metrics (input rate and processing latency) describing the execution of the operator. From the recent history, autoscale estimates the input workload and the processing capacity in near future. According to these estimations, we suggest a metric of local activity used to detect recongurations requirements in a proactive way.

• From these estimations of processing requirements in near future, autoscale identies which operators may benet from scale-in or scale-out according to metrics computed locally.

• We propose an algorithm checking the consistency of reconguration requirements at workow scope. Through an exploration of the workow from sources to sinks, the algorithm ABSTRACT :

In a Big Data context, stream processing has become a very active research domain. In order to manage ephemeral data (Velocity) arriving at important rates (Volume), some specific solutions, denoted data stream management systems (DSMSs), have been developed. DSMSs take as inputs some queries, called continuous queries,defined on a set of data streams. A continuous query generates new results as long as new data arrive in input. In many application domains, data streams have input rates and distribution of values which change over time. These variations may impact significantly processing requirements for each continuous query.

This thesis takes place in the ANR project Socioplug (ANR-13-INFR-0003). In this context, we consider a collaborative platform for stream processing. Each user can submit multiple continuous queries and contributes to the execution support of the platform. However, as each processing unit supporting treatments has limited resources in terms of CPU and memory, a significant increase in input rate may cause the congestion of the system. The problem is then how to adjust dynamically resource usage to processing requirements for each continuous query ? It raises several challenges : i) how to detect a need of reconfiguration ? ii) when reconfiguring the system to avoid its congestion at runtime ? iii) how to avoid reconfigurations that do not improves the performance of the system ?

This thesis takes place in the ANR project Socioplug (ANR-13-INFR-0003). In this context, we consider a collaborative platform for stream processing. Each user can submit multiple continuous queries and contributes to the execution support of the platform. However, as each processing unit supporting treatments has limited resources in terms of CPU and memory, a significant increase in input rate may cause the congestion of the system. The problem is then how to adjust dynamically resource usage to processing requirements for each continuous query ? It raises several challenges : i) how to detect a need of reconfiguration ? ii) when reconfiguring the system to avoid its congestion at runtime ? iii) how to avoid reconfigurations that do not improves the performance of the system ?

In this work, we are interested by the different processing steps involved in the treatment of a continuous query over a distributed infrastructure. From this global analysis, we extract mechanisms enabling dynamic adaptation of resource usage for each continuous query. We focus on automatic parallelization, or auto-parallelization, of operators composing the execution plan of a continuous query. We suggest an original approach based on the monitoring of operators and an estimation of processing requirements in near future. Thus, we can increase (scale-out), or decrease (scale-in) the parallelism degree of operators in a proactive many such as resource usage fits to processing requirements dynamically. Compared to a static configuration defined by an expert, we show that it is possible to avoid the congestion of the system in many cases or to delay it in most critical cases. Moreover, we show that resource usage can be reduced significantly while delivering equivalent throughput and result quality.

We suggest also to combine this approach with complementary mechanisms for dynamic adaptation of continuous queries at runtime. These differents approaches have been implemented within a widely used DSMS and have been tested over multiple and reproductible micro-benchmarks.