Nicoló Rivetti
email: nicolo.rivetti@univ-nantes.fr

Emmanuelle Anceaume
email: emmanuelle.anceaume@irisa.fr

Yann Busnel
email: yann.busnel@ensai.fr

Leonardo Querzoni

Bruno Sericola
email: bruno.sericola@inria.fr

Proactive Online Scheduling for Shuffle Grouping in Distributed Stream Processing Systems

Shuffle grouping is a technique used by stream processing frameworks to share input load among parallel instances of stateless operators. With shuffle grouping each tuple of a stream can be assigned to any available operator instance, independently from any previous assignment. A common approach to implement shuffle grouping is to adopt a round robin policy, a simple solution that fares well as long as the tuple execution time is constant. However, such assumption rarely holds in real cases where execution time strongly depends on tuple content. As a consequence, parallel stateless operators within stream processing applications may experience unpredictable unbalance that, in the end, causes undesirable increase in tuple completion times. In this paper we propose Proactive Online Shuffle Grouping (POSG), a novel approach to shuffle grouping aimed at reducing the overall tuple completion time. POSG estimates the execution time of each tuple, enabling a proactive and online scheduling of input load to the target operator instances. Sketches are used to efficiently store the otherwise large amount of information required to schedule incoming load. We provide a probabilistic analysis and illustrate, through both simulations and a running prototype, its impact on stream processing applications.

I. INTRODUCTION

Stream processing systems are today gaining momentum as a tool to perform analytics on continuous data streams. Their ability to produce analysis results with sub-second latencies, coupled with their scalability, makes them the preferred choice for many big data companies. A stream processing application is commonly modeled as a direct acyclic graph where data operators, represented by vertices, are interconnected by streams of tuples containing data to be analyzed, the directed edges. Scalability is usually attained at the deployment phase where each data operator can be parallelized using multiple instances, each of which will handle a subset of the tuples conveyed by the operator's ingoing stream. The strategy used to route tuples in a stream toward available instances of the receiving operator is embodied in a so-called grouping function.

Operator parallelization is straightforward for stateless operators, i.e., data operators whose output is only function of the current tuple in input. In this case, in fact, the grouping function is free to assign the next tuple in the input stream, to any available instance of the receiving operator (contrarily to statefull operators, where this choice is typically constrained). Such grouping functions are often called shuffle grouping.

Shuffle grouping implementations are designed to balance as much as possible the load on the receiving operator instances as this increases the system efficiency in available resource usage. Notable implementations [START_REF]Apache Storm[END_REF] are based on a simple Round-Robin assignment strategy that guarantees each operator instance will receive the same number of input tuples. This approach is effective as long as the time taken by each operator instance to process a single tuple (tuple execution time) is the same for any incoming tuple. In this case, all parallel instances of the same operator will experience over time, on average, the same load. However, such assumption (i.e., same execution time for all tuples of a stream) does not hold for many practical use cases. The tuple execution time, in fact, may depend on the tuple content itself. This is often the case whenever the receiving operator implements a logic with branches where only a subset of the incoming tuples travels through each single branch. If the computation associated with each branch generates different loads, then the execution time will change from tuple to tuple. As a practical example consider an operator that works on a stream of input tweets and that enriches them with historical data extracted from a database, where this historical data is added only to tweets that contain specific hashtags: only tuples that get enriched require an access to the database, an operation that typically introduces non negligible latencies at execution time. In this case shuffle grouping implemented with Round-Robin, may produce imbalance between the operator instances, and this typically causes an increase in the time needed for a tuple to be completely processed by the application (tuple completion time) as some tuple may end-up being queued on some overloaded operator instances, while other instances are available for immediate processing.

In this paper we introduce Proactive Online Shuffle Grouping (POSG) a novel approach to shuffle grouping that aims at reducing tuple completion times by accurately scheduling incoming tuples on available operator instances in order to avoid imbalances. To the best of our knowledge POSG is the first solution that explicitly addresses and solves the above stated problem.

The idea at the basis of POSG is simple: by measuring the amount needed by operator instances to process each tuple we can schedule incoming tuples minimizing the completion time. However, making such idea works in practice in a streaming setting is not trivial. In particular, POSG makes use of sketches to efficiently keep track of the huge amount of information related to tuple execution times and then applies a greedy online multiprocessor scheduling algorithm to assign tuples to operator instances at runtime. The status of each instance is monitored in a smart way in order to detect possible changes in the input load distribution and coherently adapt the scheduling. As a result, POSG provides an important performance gain in terms of tuple completion times with respect to Round-Robin for all those settings where tuple processing times are not constant, but rather depend on the tuple content.

In summary, we provide the following contributions:

• we introduce POSG the first solution for shuffle grouping that explicitly addresses the problem of parallel operator instances imbalance under loads characterized by non-uniform tuple execution times; POSG schedules tuple on target operator instances online, with minimal resource usage; it works at runtime and is able to continuously adapt to changes in the input load; • We study the two components of our solution: (i) showing that the scheduling algorithm efficiently approximate the optimal one and (ii) providing some error bounds as well as a probabilistic analysis of the accuracy of the tuple execution time tracking algorithm; • we evaluate POSG's sensibility to both the load characteristic and its configuration parameters with an extensive simulation-based evaluation that points the scenarios where POSG is expected to provide its best performance; • we evaluate POSG's performance by integrating a prototype implementation with the Apache Storm stream processing framework on Microsoft Azure platform and running it on a real dataset. After this introduction, the paper starts by defining a system model and stating the problem we intend to attack (Section II); it then introduces POSG (Section III) and shows the results of our probabilistic analysis (Section IV); results from our experimental campaign are reported in Section V and are followed by a discussion of the related works (Section VI); finally, Section VII concludes the paper.

II. SYSTEM MODEL

We consider a distributed stream processing system (SPS) deployed on a cluster where several computing nodes exchange data through messages sent over a network. The SPS executes a stream processing application represented by a topology: a direct acyclic graph interconnecting operators, represented by nodes, with data streams (DS), represented by edges. Each topology contains at least a source, i.e., an operator connected only through outbound DSs, and a sink, i.e., an operator connected only through inbound DSs. Each operator O can be parallelized by creating

k independent instances O 0 , • • • , O k-1 of it and by partitioning its inbound stream O in in k sub- streams O in 0 , • • • , O in k-1 .
Tuples are assigned to sub-streams with a grouping function. Several grouping strategies are available, but in this work we restrict our analysis to shuffle grouping where each incoming tuple can be assigned to any sub-stream. Data injected by the source is encapsulated in units called tuples and each data stream is an unbounded sequence of tuples. Without loss of generality, here we assume that each tuple t is a finite set of key/value pairs that can be customized to represent complex data structures. To simplify the discussion, in the rest of this work we deal with streams of unary tuples with a single non negative integer value. For the sake of clarity, and without loss of generality, here we restrict our model to a topology with an operator S (scheduler) which schedules the tuples of a DS O in consumed by the instances

O 0 , • • • , O k-1 of operator O.
We denote as w t,i the execution time of tuple t on operator instance O i . Abusing the notation, we may omit the instance identifier subscript. We assume that the execution time w t,i depends on the content of the tuple t: w t,i = g i (t), where g i is an unknown function that may be different for each operator instance. We simplify the model assuming that the functions g i (t) depend on a single fixed and known attribute value of t. The probability distribution of such attribute values, as well as the functions g i (t) are unknown and may change over time. However, we assume that subsequent changes are interleaved by a large enough time frame such that an algorithm may have a reasonable amount of time to adapt.

Let l(i) be the completion time of the i-th tuple of the stream, i.e., the time it takes for the i-th tuple from the moment it is injected by the source in the topology, till the moment the last operator concludes processing it. Then we can define the average completion time as:

L = i∈[m] l(i)/m.
The general goal we target in this work is to minimize the average tuple completion time L. Such metric is fundamentally driven by three factors: (i) tuple execution times at operator instances, (ii) network latencies and (iii) queuing delays. More in detail, we aim at reducing queuing delays at parallel operator instances that receive input tuples through shuffle grouping.

Typical implementation of shuffle grouping are based on Round-Robin scheduling. However, this tuple to DS sub-streams assignment strategy may introduce additional queuing delays when the execution time of input tuples is not constant. For instance, let a 0 , b 1 , a 2 be a stream with an inter tuple arrival delay of 1s, where a and b are tuples with the following execution time: w a = 10s and w b = 1s. Scheduling this stream with Round-Robin on k = 2 operator instances would assign a 0 and a 2 to instance 1 and b 1 to instance 2, with a cumulated completion time equal to 10 + 1 + 10 + (10 -2) = 29s (i.e., 8s wasted for the additional queuing delay of a 2 ). A better schedule would be to assign a 0 to instance 1, while b 1 and a 2 to instance 2. In this case, the cumulated completion time equals to 10 + 1 + 10 = 21s (i.e., no queuing delay).

III. PROACTIVE ONLINE SHUFFLE GROUPING

Proactive Online Shuffle Grouping is a shuffle grouping implementation based on a simple, yet effective idea: if we assume to know the execution time w t,i of each tuple t on the available operator instances O 0 , • • • , O k-1 , we can schedule the execution of incoming tuples on such instances with the aim of minimizing the average per tuple completion time at the operator instances. However, the value of w t,i is generally unknown. A common solution to this problem is to build a cost model for tuple execution and then use it to proactively schedule incoming load. However building an accurate cost model usually requires a large amount of a priori knowledge on the system. Furthermore, once a model has been built, it can be hard to handle changes in the system or input stream characteristics at runtime. Another common alternative is to periodically collect at the scheduler the load of the operator instances. However, this solution only allows for reactive scheduling, where input tuples are scheduled on the basis of a previous, possibly stale, load state of the operator instances; reactive scheduling typically impose a periodic overhead even if the load distribution imposed by input tuples does not change over time.

To overcome all these issues, POSG takes decision based on the estimation C i of the execution time assigned to instance i: C i = ∀t∈O in i w t,i . In order to do so, POSG computes an estimation ŵt,i of the execution time w t,i of each tuple t on each operator instance i. Then, POSG can also compute the sum of the estimated execution times of the tuples assigned to an instance i, i.e., C i = ∀t∈O in i ŵt,i , which in turn is the estimation of C i . A greedy scheduling algorithm (Section III-A) is then fed with estimations for all the available operator instances. To enable this approach, each operator instance builds a sketch (i.e., a memory efficient data structure) that will track the execution time of the tuples it process. When a change in the stream or instance(s) characteristics affects the tuples execution times on some instances, the concerned instance(s) will forward an updated sketch to the scheduler which will than be able to (again) correctly estimate the tuples execution times. This solution does not require any a priori knowledge on the stream or system, and is designed to continuously adapt to changes in the input distribution or on the instances load characteristics. In addition, this solution is proactive, namely its goal is to avoid unbalance through scheduling, rather than detecting the unbalance and then attempting to correct it.

A. Background

Data Streaming model -We present the data stream model [START_REF] Muthukrishnan | Data Streams: Algorithms and Applications[END_REF], under which we analyze our algorithms and derive bounds. A stream is an unbounded sequence of elements σ = t 1 , . . . , t m called tuples or items, which are drawn from a large universe [n] = {1, . . . , n}. The unknown size (or length) of the stream is denoted by m. We denote by p t the unknown probability of occurrence of item t in the stream and with f t the unknown frequency1 of item t, i.e., the number of occurrences of t in the stream of size m.

2-Universal

Hash Functions -Our algorithm uses hash functions randomly picked from a 2-universal hash functions family. A collection H of hash functions h : {1, . . . , n} → {0, . . . , c} is said to be 2-universal if for every

two different items x, y ∈ [n], for all h ∈ H, P{h(x) = h(y)} ≤ 1 c
, which is the probability of collision obtained if the hash function assigned truly random values to any x ∈ [c]. Carter and Wegman [START_REF] Carter | Universal classes of hash functions[END_REF] provide an efficient method to build large families of hash functions approximating the 2-universality property.

Count Min sketch algorithm -Cormode and Muthukrishnan have introduced in [START_REF] Cormode | An improved data stream summary: The count-min sketch and its applications[END_REF] the Count Min sketch that provides, for each item t in the input stream an (ε, δ)-additive-approximation ft of the frequency f t . The Count Min sketch consists of a two dimensional matrix F of size r × c, where r = log 1 δ and c = e ε . Each row is associated with a different 2-universal hash function h

i : [n] → [c]. When the Count Min algorithm reads sample t from the input stream, it updates each row: ∀i ∈ [r], F[i, h i (t)] ← F[i, h i (t)] + 1.
Thus, the cell value is the sum of the frequencies of all the items mapped to that cell. Upon request of f t estimation, the algorithm returns the smallest cell value among the cell associated with t:

ft = min i∈[r] {F[i, h i (t)]}.
Fed with a stream of m items, the space complexity of this algorithm is O( 1 ε log 1 δ (log m + log n)) bits, while update and query time complexities are O(log 1/δ). The Count Min algorithm guarantees that the following bound holds on the estimation accuracy for each iteam read from the input stream:

P{| ft -f t |≥ ε(m -f t )} ≤ δ, while f t ≤ ft is always true.
This algorithm can be easily generalized to provide (ε, δ)-additive-approximation of point queries Qt on stream of udpates, i.e., a stream where each item t carries a positive integer update value v t . When the Count Min algorithm reads the pair t, v t from the input stream, the update routine changes as follows: ∀i ∈

[r], F[i, h i (t)] ← F[i, h i (t)] + v t .
Greedy Online Scheduler -A classical problem in the load balancing literature is to schedule independent tasks on identical machines minimizing the makespan, i.e., the Multiprocessor Scheduling problem. In this paper we adapt this problem to our setting, i.e., to schedule online independent tasks on non uniform machines aiming to minimizing the average per task completion time L. Online scheduling means that the scheduler does not know in advance the sequence of tasks it has to schedule. The Greedy Online Scheduler algorithm assigns the currently submitted tasks to the less loaded available machine. In Section IV-A we prove that this algorithm closely approximates the optimal algorithm.

B. POSG design

Each operator instance op maintains two Count Min sketch matrices (Figure 1.A): the first one, denoted as F op , tracks the tuple frequencies f t,op ; the second, denoted as W op , tracks the tuples cumulated execution times W t = w t,op ×f t,op . Both Count Min matrices share the same sizes and hash functions. The latter is the generalized version of the Count Min presented in Section III-A where the update value is the tuple execution time when processed by the instance (i.e., v t = w t,op ) . The operator instance will update (Listing III.1) both matrices after each tuple execution.

The operator instances are modelled as a finite state machines (Figure 2) with two states: START and STABI-LIZING. The START state lasts until instance op has executed N tuples, where N is a user defined window size for i = 1 to c do 8:

c 1 2 3 4 r 2 1 FO 2 c 1 2 3 4 WO 2 FO 2 , WO 2 O2 FO 1 , WO 1 O1 POSG C = [ CO 1 , CO 2 ] FO 1 , WO 1
Fop[i, hi(t)] ← Fop[i, hi(t)] + 1 9: Wop[i, hi(t)] ← Wop[i, hi(t)] + l 10:
end for 11: end function parameter. The transition to the STABILIZING state (Figure 2.A) triggers the creation of a new snapshot S op . A snapshot is a matrix of size r × c where ∀i ∈

[r], j ∈ [c] : S op [i, j] = W op [i, j]/F op [i, j].
We say that the F and W matrices are stable when the relative error η op between the previous snapshot and the current one is smaller than µ: if

η op = ∀i,j |S op [i, j] - Wop[i,j] Fop[i,j])| ∀i,j S op [i, j] ≤ µ (1)
is satisfied. Then, each time instance op has executed N tuples, it checks whether Equation 1 is satisfied. (i) If not, then S op is updated (Figure 2.B). (ii) Otherwise the operator instance sends the F op and W op matrices to the scheduler (Figure 1.B), resets them and moves back to the START state (Figure 2

.C).

There is a delay between any change in the stream or operator instances characteristics and when the scheduler receives the updated F op and W op matrices from the affected operator instance(s). This introduces a skew in the cumulated execution times estimated by the scheduler. In order to compensate for this skew, we introduce a synchronization mechanism that kicks in whenever the scheduler receives a new pair of matrices from any operator instance. Notice also that there is an initial transient phase in which the scheduler has not yet received any information from operator instances. This means that in this first phase it has no information on the tuples execution times and is forced to use the Round-Robin policy. This mechanism is thus also needed to initialize the estimated cumulated execution times when the Round-Robin phase ends. The scheduler (Figure 1.C) maintains the estimated cumulated execution time for each instance, in a vector C of size k, and the set of pairs of matrices: { F op , W op }, initially empty. The scheduler is modelled as a finite state machine (Figure 3) with four states: ROUND ROBIN, SEND ALL, WAIT ALL and RUN.

The ROUND ROBIN state is the initial state in which scheduling is performed with the Round-Robin policy. In this state, the scheduler collects the F op and W op matrices sent by the operator instances (Figure 3.A). After receiving the two matrices from each instance (Figure 3.B), the scheduler is able to estimate the execution time for each submitted tuple and moves into the SEND ALL state. SEND ALL state goal is to send the synchronization requests towards the k instances. To reduce overhead, requests are piggy backed (Figure 1.D) with outgoing tuples applying the Round-Robin policy for the next k tuples: the i-th tuple is assigned to operator instance i mod k. On the other hand, the estimated cumulated execution time vector C is updated with the tuple estimated execution time using the UPDATE C function (Listing III.2). When all the requests have been sent (Figure 3.C), the scheduler moves into the WAIT ALL state. This state collects the synchronization replies from the operator instances (Figure 3.D). Operator instance op's reply (Figure 1.E) contains the difference ∆ op between the instance cumulated execution time C op and the scheduler estimation C[op]. In the WAIT ALL state, scheduling is performed as in the RUN state, using both the SUBMIT and the UPDATE C functions (Listing III.2). When all the replies for the current epoch have been collected, synchronization is performed and the scheduler moves in the RUN state (Figure 3.E). In the RUN state, the scheduler assigns the input tuple applying the Greedy Online Scheduler algorithm, i.e., assigns the tuple to the operator instance with the least estimated cumulated execution time (SUBMIT function, Listing III.2). Then it increments the target instance estimated cumulated execution time with the estimated tuple execution time (UPDATE C function, Listing III.2). Finally, in any state except ROUND ROBIN, receiving an updated pair of matrices F op and W op moves the scheduler into the SEND ALL state (Figure 3.F).

Theorem 3.1 (Time complexity of POSG): For each tuple read from the input stream, the time complexity of POSG for each instance is O(log 1/δ). For each tuple submitted to the scheduler, POSG time complexity is O(k + log 1/δ).

Proof: By Listing III.1, for each tuple read from the input stream, the algorithm increments an entry per row of both the F op and W op matrices. Since each has log 1/δ rows, the resulting update time complexity is O(log 1/δ) By Listing III.2, for each submitted tuple, the scheduler has to retrieve the index with the smalled value in the vector C of size k, and to retrieve the estimated execution time for the submitted tuple. This operation requires to read entry per row of both the F op and W op matrices. Since each has log 1/δ rows, the resulting update time complexity is O(k + log 1/δ) Listing III.2: Scheduler: submit t and update C. δ × e ε of counters of size log m. In addition, it also stores a hash function with a domain of size n. Then the space complexity of POSG on each operator instance is O 1 ε log 1 δ (log m + log n) . The scheduler stores the same matrices, one for each instance, as well as a vector of size k. Then the space complexity of POSG on the scheduler is O k 1 ε log 1 δ (log m + log n) .

IV. THEORETICAL ANALYSIS

This section provide the analysis of the quality of the scheduling performed by POSG in two steps. First we study the Greedy Online Scheduler algorithm approximation. Then, in Section IV-B we provide a probabilistic analysis of the mechanism that POSG uses to estimate the tuple execution times.

A. Online Greedy Scheduler

We suppose that tuples cannot be preempted, that is tuples must be processed in an uninterrupted fashion on the operator instance it has been scheduled on. As mentioned, we assume that the processing time w tj is known for each tuple t j . Finally, given our system model, the problem of minimizing the average completion time L can be reduced to the following problem (in terms of makespan):

Problem 4.1: Given k identical operator instances, and a sequence of tuples σ = t 1 , . . . , t m that arrive online from the input stream. Find an online scheduling algorithm that minimizes the makespan of the schedule produced by the online algorithm when fed with σ.

Let OPT be the schedule algorithm that minimizes the makespan over all possible sequences σ, and C σ OP T denote the makespan of the schedule produced by the OPT algorithm fed by sequence σ. Notice that finding C σ OP T is an NP-hard problem. We will show that the Greedy Online Scheduler (GOS) algorithm defined in Section III-A builds a schedule that is within some factor of the lower bound of the quality of the optimal scheduling algorithm OPT. Let us denote by C σ GOS the makespan of the schedule produced by the greedy algorithm fed with σ. Theorem 4.2: For any σ, we have

C σ GOS ≤ (2 -1/k)C σ OP T .
Proof: Let i ∈ [k] be the instance on which the last tuple t j is executed. By construction of the algorithm, when tuple t j starts its execution on instance i, all the other instances are busy, otherwise t j would have been executed on another instance. Thus when tuple t j starts its execution on instance i, each of the k instances must have been allocated a load at least equivalent to ( m =1 w tw tj )/k. Thus we have,

C σ GOS -w tj ≤ m =1 w t -w tj k C σ GOS ≤ m =1 w t k + w tj (1 - 1 k ) (2)
Now, it is easy to see that

C σ OP T ≥ m j=1 w tj k , (3)
otherwise the total load processed by all the machines in the schedule produced by the OPT algorithm would be strictly less than m j=1 w tj , leading to a contradiction. We also trivially have

C σ OP T ≥ max w t . (4)
Thus combining relations (2), (3), and (4), we have

C σ GOS ≤ C σ OP T + C σ OP T (1 - 1 k ) = (2 - 1 k )C σ OP T (5)
that concludes the proof. This lower bound is tight, that is there are sequences of tuples for which the Greedy Online Scheduler algorithm produces a schedule whose completion time is exactly equal to (2 -1/k) times the completion time of the optimal scheduling algorithm [START_REF] Gusfield | Bound the naive multiple machine scheduling with release times deadlines[END_REF].

Consider the example of a(a-1) tuples whose processing time is equal to w max /k and one task with a processing time equal to w max . Suppose that the a(a -1) tuples are scheduled first and then the longest one. Then the greedy algorithm will exhibit a makespan equal to (m -1)w max /k + w max = w max (2 -1/k) while the OPT scheduling will lead to a makespan equal to w max .

B. Execution Time Estimation

POSG uses two matrices, F and W, to estimate the execution time w t of each tuple submitted to the scheduler. To simplify the discussion, we consider a single operator instance.

From the count Min algorithm, and for any v ∈ {1, . . . , n}, we have for a given hash function h i ,

C v (m) = n u=1 f u 1 {hi(u)=hi(v)} = f v + n u=1,u =v f u 1 {hi(u)=hi(v)} .
and

W v (m) = f v w v + n u=1,u =v f u w u 1 {hi(u)=hi(v)} ,
where

C v (m) = F[v][h v (m)] and W v (m) = W[v][h v (m)].
Let us denote respectively by w min and w max the minimum and the maximum execution time of the items. We have trivially

w min ≤ W v (m) C v (m) ≤ w max .
In the following we write respectively C v and W v instead of C v (m) and W v (m), to simplify the writing. For any i = 0, . . . , n -1, we denote by U i (v) the set which elements are the subsets {1, . . . , n} \ {v} whose size is equal to i, that is

U i (v) = {A ⊆ {1, . . . , n} \ {v} | |A| = i}.
We have U 0 (v) = {∅}.

For any v = 1, . . . , n, i = 0, . . . , n -1 and A ∈ U i (v), we introduce the event B(v, i, A) defined by

B(v, i, A) = {h u = h v , ∀u ∈ A and h u = h v , ∀u ∈ {1, . . . , n} \ (A ∪ {v})} .
From the independence of the h u , we have

Pr{B(v, i, A)} = 1 k i 1 - 1 k n-1-i .
Let us consider the ratio W v /C v . For any i = 0, . . . , n, we define

R i (v) = f v w v + u∈A f u w u f v + u∈A f u , A ∈ U i (v) .
We have R 0 (v) = {w v }. We introduce the set R(v) defined by

R(v) = n-1 i=0 R i (v).
Thus with probability 1,

W v /C v ∈ R(v).
Let x ∈ R(v). We have

Pr{W v /C v = x} = n-1 i=0 A∈Ui(v) Pr{W v /C v = x | B(v, i, A)} Pr{B(v, i, A)} = n-1 i=0 1 k i 1 - 1 k n-1-i A∈Ui(v) Pr{W v /C v = x | B(v, i, A)} = n-1 i=0 1 k i 1 - 1 k n-1-i A∈Ui(v)
1 {x=(fvwv+ u∈A fuwu)/(fv+ u∈A fu)} .

Thus

E{W v /C v } = n-1 i=0 1 k i 1 - 1 k n-1-i A∈Ui(v) x∈R(v) x1 {x=(fvwv+ u∈A fuwu)/(fv+ u∈A fu)} = n-1 i=0 1 k i 1 - 1 k n-1-i A∈Ui(v) f v w v + u∈A f u w u f v + u∈A f u x∈R(v) 1 {x=(fvwv+ u∈A fuwu)/(fv+ u∈A fu)} = n-1 i=0 1 k i 1 - 1 k n-1-i A∈Ui(v) f v w v + u∈A f u w u f v + u∈A f u .
Let us assume that all the f u are equal, that is for each u, we have f u = m/n. Simulations tend to show that the worst cases scenario of input streams are exhibited when all the items show the same number of occurrences in the input stream. We get

Pr{W v /C v = x} = n-1 i=0 1 k i 1 - 1 k n-1-i A∈Ui(v)
1 {x=(wv+ u∈A wu)/(i+1)} .

We define S = n i=1 w i . We then have Theorem 4.3:

E{W v /C v } = S -w v n -1 - k(S -nw v ) n(n -1) 1 -1 - 1 k n .
It important to note that this result does not depend on m.

Let us now consider a numerical application. We take k = 55, n = 4096 and the distinct values of w u equal to 1, 2, 3, . . . , 64, each item being present 64 times in the input stream, we get for v = 1, . . . , 64, E{W v /C v } ∈ [32.08, 32.92]. Note also from above that we have

1 ≤ W v /C v ≤ 64.
From the Markov inequality we have, for every x > 0,

Pr{W v /C v ≥ x} ≤ E{W v /C v } x .
By taking x = 64a, with a ∈ [0.6, 1), we obtain

Pr{W v /C v ≥ 64a} ≤ E{W v /C v } 64a ≤ 33 64a .
If r denotes the number of rows of the system, we have by the independence of the h functions, Pr{ min t=1,...,r

(W t,v /C t,v ) ≥ 64a} = (Pr{W v /C v ≥ 64a}) r ≤ 33 64a r .
By taking for instance a = 3/4 and r = 10, we get Pr{ min t=1,...,r

(W t,v /C t,v ) ≥ 48} ≤ 11 16
10 ≤ 0.024.

V. EXPERIMENTAL EVALUATION

In this section we evaluate the performance obtained by using POSG to perform shuffle grouping. We will first describe the general setting used to run the tests and will then discuss the results obtained through simulations (Section V-B) and with a prototype of POSG targeting Apache Storm (Section V-C).

A. Setup

Datasets -In our tests we consider both synthetic and real datasets. For synthetic datasets we generate streams of integer values (items) representing the values of the tuple attribute driving the execution time when processed on an operator instance. We consider streams of m = 32, 768 tuples, each containing a value chosen among n = 4, 096 distinct items. Synthetic streams have been generated using the Uniform distribution and Zipfian distributions with different values of α ∈ {0.5, 1.0, 1.5, 2.0, 2.5, 3.0}, denoted respectively as Zipf-0.5, Zipf-1.0, Zipf-1.5, Zipf-2.0, Zipf-2.5, and Zipf-3.0. We define w n as the number of distinct execution time values that the tuples can have. These w n values are selected at constant distance in the interval [w min , w max ]. We have also run tests generating the execution time values in the interval [w min , w max ] with geometric steps without noticing unpredictable differences with respect to the results reported in this section. The algorithm parameters are the operator window size N , the tolerance parameter µ, and the parameters of the matrices F and W: ε and δ.

Unless otherwise specified, the frequency distribution is Zipf-1.0 and the stream parameters are set to w n = 64, w min = 1 ms and w max = 64 ms, this means that the execution times are picked in the set {1, 2, • • • , 64}. The algorithm parameters are set to N = 1024, µ = 0.05, ε = 0.05 (i.e., c = 54 columns) and δ = 0.1 (i.e., r = 4 rows). If not stated otherwise, the operator instances are uniform (i.e., a tuple has the same execution time on any instance) and there are k = 5 instances. Let W be the average execution time of the stream tuples, then the stream maximum theoretical input throughput sustainable by the setup is equal to k/W . When fed with an input throughput smaller than k/W the system will be over-provisioned (i.e., possible underutilization of computing resources). Conversely, an input throughput larger than k/W will result in an undersized system. We refer to the ratio between the maximum theoretical input throughput and the actual input throughput as the percentage of over-provisioning that, unless otherwise stated, was set to 100%.

In order to generate 100 different streams, we randomize the association between the w n execution time values and the n distinct items: for each of the w n execution time values we pick uniformly at random n/w n different values in [n] that will be associated to that execution time value. This means that the 100 different streams we use in our tests do not share the same association between execution time and item as well as the association between frequency and execution time (thus each stream has also a different average execution time W ). We have also Maximum Execution Time Value -Figure 6 shows the average completion time L as a function of the maximum execution time w max . As expected, the average completion time increases with w max . On the other hand, and quite unexpectedly, the gain between the maximum, mean and minimum average completion times of POSG with respect to Round-Robin only slightly improves for growing values of w max , with an average speed up S L of 1.19.

Number of Execution Time

Values -Figure 7 shows the average completion time L as a function of the number of execution time values w n . We can notice that for growing values of w n both the average completion time values and variance decrease, with only slight changes for w n ≥ 16. Recall that w n is the number of completion time values in the interval [w min , w max ] that we assign to the n distinct attribute values. For instance, with w n = 2, all the tuples have a completion time equal to either 1 or 64 ms. Then, assigning either of the two values to the most frequent item strongly affects the average completion time. Increasing w n reduces the impact that each single execution time has on the average completion time, leading to more stable results. As in the previous plot, the gain between the maximum, mean and maximum average completion times of POSG and Round-Robin (in average 19%) is mostly unaffected by the value of w n .

Number of operator instances k -Figure 8 shows the speed up S L as a function of the number of parallel operator instances k. With k = 1 the speed up is equal to 1, which is the maximum theoretically achievable speed up with this configuration; this means that our algorithm is efficient as it does not introduce large delays in the average completion time. Increasing the number of instances slightly increases the speed up. However, this growth rapidly converges: there is a 4% increasing from 2 to 3 operators, while from 9 to 10 operators the growth is of only 1%. This result intuitively stems from the fact that moving from k = 2 to k = 3 we are increasing the available instances by 50%, while moving from k = 9 to k = 10 only adds 1/9 more instances.

Precision parameter ε -Figure 9 shows the speed up S L as a function of the precision parameter ε value that controls the number of columns in the F and W matrices. With smaller values of ε POSG is more precise but also uses more memory, i.e., for ε = 1.0 there is a single entry per row, while for ε = 0.001 there are 2781 entries per row. As expected, decreasing ε improves POSG performance: in average a 10 time decrease in ε (thus a 10 time increase in memory) results in a 30% increase in the speed up. Large values of ε do not provide good performance; however, starting with ε ≤ 0.09 the minimum average completion time speed up is always larger than 1. Time Series -Figure 10 shows the completion time as the stream unfolds (the x axis is the number of tuples read from the stream) for both POSG and Round-Robin. Each point on the plot is the maximum, mean and minimum completion time over the previous 2000 tuples. The plot for Round-Robin has been artificially shifted by 1000 tuples to improve readability. In this test the stream is of size m = 150, 000 split into two periods: the tuple execution times for operator instances 1, 2, 3, 4 and 5 are multiplied by 1.05, 1.025, 1.0, 0.975 and 0.95 respectively for the first 75, 000 tuples, and for the remaining 75, 000 tuples by 0.90, 0.95, 1.0, 1.05 and 1.10 respectively. This setup mimics an abrupt change in the load characteristic of target operator instances (possibly due to exogenous factors).

In the leftmost part of the plot, we can see POSG and Round-Robin provide the same exact results up to m = 10, 690, where POSG starts to diverge by decreasing the completion time as well as the completion time variance. This behaviour is the result of POSG moving into the RUN state at m = 10, 690. After this point it starts to schedule using the F and W matrices and the Greedy Online Scheduler algorithm, improving its performance with respect to Round-Robin.

At m = 75, 000 we inject the load characteristic change described above. Immediately POSG performance degrades as the content of F and W matrices is outdated. At m = 84, 912 the scheduler receives the updated F and W matrices and recovers. This demonstrates POSG's ability to adapt at runtime with respect to changes in the load distributions.

C. Prototype

To evaluate the impact of POSG on real applications we implemented it as a custom grouping function within the Apache Storm [START_REF]Apache Storm[END_REF] framework. We have deployed our cluster on Microsoft Azure cloud service, using a Standard Tier A4 VM (4 cores and 7 GB of RAM) for each worker node, each with a single available slot.

The test topology is made of a source (spout) and two operators (bolts) S and O. The source generates (reads) the synthetic (real) input stream. Bolt S uses either POSG or the Apache Storm standard shuffle grouping implementation (ASSG in the following) to route the tuples toward the k instances (tasks) of bolt O. Time Series -Figure 11 provides results for the prototype with the same settings of the test whose results were reported in Figure 10. We can notice the same general behaviour both in the simulator and in the prototype. In the right part of the plot POSG diverges from ASSG at m = 20, 978 and decreases the completion time as well as the completion time variance. On the right part of the plot, after m = 75, 000 POSG performance degrade due to the change in the load distributions. Finally, at m = 82, 311 the scheduler receives the updated F and W matrices and starts to recover. Notice also that during the execution with ASSG, 1, 600 tuples timed out (and where not recovered as the topology was configured disabling Storm fault tolerance mechanisms). This clearly shows how the shuffle grouping scheduling policy can have a large impact on the system performances.

Simple Application with Real Dataset -In this test we pretended to run a simple application on a real dataset: for each tweet of the twitter dataset mentioned in Section V-A we want to gather some statistics and decorate the outgoing tuples with some additional information. However the statistics and additional informations differ depending on the class the entities mentioned in each tweet belong. We assumed that this leads to a long execution time for media (e.g. possibly caused by an acces to an external DB to gather historical data), an average execution time for politicians and a fast execution time for others (e.g. possibly because these tweets are not decorated). We modeled execution times with 25ms, 5 ms and 1ms of busy waiting respectively. Figure 12 shows the average completion time L for both POSG and ASSG as a function of the number of instances k. For all values of k (except k = 1), POSG provides lower average completion times than Round-Robin, with a mean speed up S L of 1.37. Increasing the value of k decreases the gap between POSG and Round-Robin. However with k = 10 POSG still improves the average completion time by 16%. For k = 2 and k = 7, we can notice an unanticipated behaviour of Round-Robin: adding one more instance increases the completion times (in particular, with k = 2 there where 221 timeouted tuples). On the other hand, POSG average completion time always decreases with growing values of k. Notice also that, to provide this improvement, POSG exchanged only 916 additional messages (with respect to a stream of size m = 500, 000).

VI. RELATED WORK

Load balancing in distributed computing is a well known problem that has been extensively studied since the 80s [START_REF] Cardellini | The state of the art in locally distributed web-server systems[END_REF], [START_REF] Zhou | Performance Studies of Dynamic Load Balancing in Distributed Systems[END_REF]. Distributed stream processing systems have been designed, since the beginning, by taking into account the fact that load balancing is a critical issue to attain the best performance.

Hirzel et al. [START_REF] Hirzel | A catalog of stream processing optimizations[END_REF] recently provided an extensive overview of possible optimization strategies for stream processing systems, including load balancing. They identify two ways to perform load balancing in stream processing systems: either when placing the operators on the available machines or when assigning load to operator instances. In this latter case, the load balancing mechanism can be either pull based, i.e., it is the consumers responsibility to acquire the load from the producers, or push based, i.e., the converse.

In the the last year there has been new interest on improving load balancing with key grouping [START_REF] Gedik | Partitioning functions for stateful data parallelism in stream processing[END_REF], [START_REF] Rivetti | Efficient key grouping for near-optimal load balancing in stream processing systems[END_REF]. However, key grouping imposes some strict limitations for assigning tuples to operator instances; as such, solutions available for key grouping would underperform if applied with shuffle grouping.

Considering shuffle grouping, Arapaci et al. [START_REF] Arpaci-Dusseau | Cluster i/o with river: Making the fast case common[END_REF] as well as Amini et al. [START_REF] Amini | Adaptive control of extreme-scale stream processing systems[END_REF], among other contributions, provide solutions to maximise the system efficiency when the operator instances execution times are non uniform, either because the hardware is etherogeneous or because each instance carries out different computations. On the other hand, at the best of our knowledge, there is no prior work directly addressing load balancing with shuffle grouping on non uniform operator instances considering that the tuples execution time depend on the tuple themselves.

VII. CONCLUSIONS

In this paper we introduced Proactive Online Shuffle Grouping, a novel approach to shuffle grouping aimed at reducing the overall tuple completion time by scheduling tuples on operator instances on the basis of their estimated execution time. POSG makes use of sketch data structures to keep track of tuple execution time on operator instances in a compact and scalable way. This information is then fed to a greedy scheduling algorithm to assign incoming load.

The analysis of POSG backs up the results of the experimental evaluation, proving that the Greedy Online Scheduler algorithm is a (2 -1/k)-approximation of the optimal one as well as providing bounds and insights on the accuracy of the estimation of the execution time w t . Furthermore, we extensively tested POSG performance both through simulations and with a prototype implementation integrated within the Apache Storm framework. The results showed how POSG provides important speedups in tuple completion time when the workload is characterized by skewed distributions. Further research will be needed to explore how much the load model affect performance.

For example it would be interesting to include other metrics in the load model, e.g. network latencies, to check how much these may improve the overal performance.

FO 2 ,Fig. 1 . 3 :

 213 Fig. 1. Proactive Online Shuffle Grouping design with r = 2 (δ = 0.25), c = 4 (ε = 0.70) and k = 2.Listing III.1: Operator instance op: update F op and W op .

Fig. 2 .

 2 Fig. 2. Operator instance finite state machine.

Fig. 3 .

 3 Fig. 3. Scheduler finite state machine.

7 :

 7 h1 . . . hr of the operator instances 5: end init 6: function SUBMIT(tuple : t) return arg min op∈[k] { C[op]} 8: end function 9: function UPDATE C(tuple : t, operator : op) 10:i ← arg min i∈[r] {Fop[i, hi(t)]} 11: C[op] ← C[op] + (Wop[i, hi(t)]/Fop[i, hi(t)]) 12: end function Theorem 3.2 (Space Complexity of POSG):The space complexity of POSG for the operator instances is O 1 ε log 1 δ (log m + log n) , while the space complexity for the scheduler is O k 1 ε log 1 δ (log m + log n) . Proof: Each operator instance stores two matrices of size log 1

1 ε 1 ε log 1 δ 1 ε log 1 δ

 11111 Theorem 3.3 (Communication complexity of POSG): The communication complexity of POSG is of O m N log 1 δ (log m + log n) + k log m bit and O k m N messages. Proof: After executing N tuples, an operator instance may send the F op , W op matrices to the scheduler. This generates a communication cost of O m kN messages and O m kN (log m + log n) bits. When the scheduler receives these matrices, the synchronization mechanism kicks in and triggers a round trip communication (half of which is piggy backed by the tuples) with each instance. The communication cost of the synchronization mechanism O m N messages and O m N log m bits. Since there are k instances, the overall communication complexity is O k m N messages and O m N (log m + log n) + k log m bits Note that the communication cost is negligible since the window size N should be chosen such that N k (e.g., in our tests we have N = 1024 and k ≤ 10).

Fig. 5 .

 5 Fig. 5. Speed up S L as a function of the percentage of over-provisioning

Fig. 6 .

 6 Fig. 6. Average per tuple completion time L as a function of the maximum execution time value wmax.

Fig. 7 .Fig. 8 .

 78 Fig. 7. Average per tuple completion time L as a function of the number of execution time values wn.

Fig. 9 .

 9 Fig. 9. Speed up S L as a function of the precision parameter ε (i.e., number of columns c = e/ε).

Fig. 10 .Fig. 11 .

 1011 Fig. 10. Simulator per tuple completion time time-series.

Fig. 12 .

 12 Fig. 12. Prototype average per tuple completion time L as a function of the number of operators k.

This definition of frequency is compliant with the data streaming literature.

This work was partially funded by the French ANR project SocioPlug (ANR-13-INFR-0003), and by the DeSceNt project granted by the Labex CominLabs excellence laboratory (ANR-10-LABX-07-01).

build these associations using other distributions, namely geometric and binomial, without noticing unpredictable differences with respect to the results reported in this section. We retrieved a dataset containing a stream of preprocessed tweets related to Italian politicians crawled during the 2014 European elections. Among other information, the tweets are enriched with a field mention containing the entities mentioned in the tweet. These entities can be easily classified into politicians, media and others. We consider the first 500, 000 tweets, mentioning roughly n = 35, 000 distinct entities and where the most frequent entity ("Beppe Grillo") has an empirical probability of occurrence equal to 0.065.

Evaluation Metrics -The evaluation metrics we provide are (i) the average per tuple completion time: L alg (simply average completion time in the following), where alg is the algorithm used for scheduling, and (ii) the average per tuple completion time speed up (simply speed up in the following) achieved by POSG with respect to Round-Robin: S L . Recall that l alg (i) is the completion time of the i-th tuple of the stream when using the scheduling algorithm alg. Then we can define the average completion time L alg and speed up S L as follows:

l POSG (i) Whenever applicable we provide the maximum, mean and minimum figures over the 100 executions.

B. Simulation Results

Frequency Probability Distribution -Figure 4 shows the average completion time L for POSG, Round-Robin and Full Knowledge with different frequency probability distributions. The Full Knowledge algorithm represents an ideal execution of the Online Greedy Scheduling algorithm when fed with the exact execution time for each tuple. Increasing the skewness of the distribution reduces the number of distinct tuples that, with high probability, will be fed for scheduling, thus simplifying the scheduling process. This is why all algorithms perform better with highly skewed distributions. On the other hand, uniform or lightly skewed (i.e., Zipf-0.5) distributions seem to be worst cases, in particular for POSG and Round-Robin. With all distributions the Full Knowledge algorithm outperforms POSG which, in turn, always provide better performance than Round-Robin. However, for uniform or lightly skewed distributions (i.e., Zipf-0.5), the gain introduced by POSG is limited (in average 6%). Starting with Zipf-1.0 the gain is much more sizeable (25%) and with Zipf-1.5 we have that the maximum average completion time of POSG is smaller than the minimum average completion time of Round-Robin. Finally, with Zipf-2.5 POSG matches the performance of Full Knowledge. This behavior for POSG stems from the ability of its sketch data structures (F and W matrices, see Section III) to capture more useful information for skewed input distributions.

Input Throughput -Figure 5 shows the speed up S L as a function of the percentage of over-provisioning. When the system is strongly undersized (95% to 98%), queuing delays increase sharply, reducing the advantages offered by POSG. Conversely, when the system is oversized (107 to 115%), queuing delays tend to 0, which in turns also reduces the improvement brought by our algorithm. However, in a correctly sized system (i.e., from 100% to 109%), our algorithm introduces a noticeable speed up, in average at least 1.15 with a peak of 1.26 at 102%. Finally, even when the system is largely oversized (115%), we still provide an average speed up of 1.07.


Jean-Pierre Bigot

Abdellah Touil

Patrick Bonnet

Jean-Marc Lacôme

Rain-out investigation: initial droplet size measurement

Keywords: rain-out, loss of containment, flashing liquid jets, thermal fragmentation, droplet size distribution

Droplet size distribution inside water flashing jets and corresponding rain-out fraction were measured. Mass distribution showed that a few droplets are "large" (> 150 µm) and count for more than 85 % of the liquid mass in the jet because of their large individual mass. This could be due to incomplete thermal fragmentation. It could explain the rain-out falling near the orifice or pipe exit.

Introduction

The rain-out problem is part of the loss of containment scenario about liquefied gases reservoirs: how much of the released liquid phase will fall on the ground? This fraction will not directly participate to the toxic or flammable cloud.

Experimental data about this topic are seldom [START_REF] Johnson | RELEASE A model with data to predict aerosol Rain-out in accidental Releases[END_REF][START_REF] Hocquet | Loss of containment : experimental aerosol rain-out assessment[END_REF]. Models are either not sufficiently validated because of this lack of experimental data or even not in good agreement with the existing ones [START_REF] Wheatley | Discharge of liquid ammonia to moist atmospheres, survey of experimental data and model for estimating initial conditions for dispersion calculations[END_REF][START_REF] Ianello | Aerosol research program: improved source term definition for modelling the ambient impact of accidental release of hazardous liquids[END_REF][START_REF] Epstein | A model of the dilution of a forced twophase chemical plume in a horizontal wind[END_REF][START_REF] Papadourakis | Upper and lower bounds of droplet evaporation in two-phase jets[END_REF], UDM, Johnson 1999).

A former work [START_REF] Hocquet | Loss of containment : experimental aerosol rain-out assessment[END_REF] showed that rain-out forms a few meters downstream the accidental breach under the horizontal aerosol. However literature predicts small droplets (d d < 150 µm) will be formed through the thermal fragmentation mechanism [START_REF] Brown | Sprays formed by flashing liquid jets[END_REF][START_REF] Bushnell | Atomisation of superheated Water jets at low ambient pressures[END_REF][START_REF] Gooderum | Measurement of Mean Drop Sizes for Sprays from Superheated Water jets[END_REF][START_REF] Lienhard | The break-up of superheated liquid jets[END_REF][START_REF] Oza | An Experimental and Analytical Study of Flash-boiling Fuel Injection[END_REF][START_REF] Oza | On the Mechanism of Flashing Injection of Initially Sub-cooled Fuels[END_REF][START_REF] Razzaghi | Droplet size estimation of two phase flashing jets[END_REF][START_REF] Reitz | A Photographic Study of Flash-Boiling Atomisation[END_REF][START_REF] Park | An experimental investigation of the flash atomisation mechanism[END_REF][START_REF] Ramsdale | Droplet formation and rain-out from two phase flashing jets[END_REF][START_REF] Witlox | Unified Dispersion Model[END_REF][START_REF] Witlox | Flashing liquid jets and two-phase dispersion[END_REF]. Such small droplets would normally be * corresponding author. Tel. +33 (0)4 77 42 00 29; fax : +33 (0)4 77 42 49 96 94; bigot@emse.fr entrained by the aerosol. The aim of this work is to understand how this non homogeneous behaviour can occur.

Experimental

We used two different experimental set-ups in order to measure drop sizes, temperature and rain-out fraction for known initial conditions T res and P res . Both set-ups used the same test sections. The tested fluid was water.

Test sections: orifices and pipes

We used both stainless steel orifices and pipes (fig. 1). The 2 mm inner diameter orifice (D2) and pipe (D2L100) where mainly used. But some experiments were conducted with D5 and D8 orifices, D5L250 and D8L400 pipes.

Rain-out, mass flow rate and jet temperature measurements

The experimental set-up for these measurements is mainly composed of an upstream tank, the test section, jet temperature probes and rain-out capture basins (fig. 2. & 3.). The upstream tank consists of a jacketed stainless steel column (4.5 m high, .257 m inner diameter, 0.233 m 3 volume). It allows for a few minutes steady-state experiments up to P res = 1.1 MPa and T res = 170°C. A ± 0.1 °C temperature uniformity for the tested fluid inside the tank was obtained. Pressure at the test section level is maintained constant within ± 2 kPa.

Jet temperature is measured both on its axis and just above the capture basins through 20 equally spaced (1 m) Pt temperature probes ( ± 0.75K measurement uncertainty). Mass flow rate through the test section is measured by comparing the change of the pressure head above the test section during and experiment to the change when a known amount of liquid is added ( ± 1% accuracy). 15 capture basins ( 1m * 2 m or 1 m * 1 m) allow rain-out water capture (fig. 2. & 3.). Water captured in each capture basin is weighted. Ratio of mass in one capture basin to the total mass released from the upstream tank is a measure of the rain-out density function ( ± 2 % accuracy).

Droplet velocity and size measurements

The experimental set-up for these measurements is mainly composed of another upstream tank, the same test sections and a PDA system. The test section is connected to the upstream tank through a 3 m long, 0.013 m inner diameter, heat insulated flexible hose (fig. 4.).

This upstream tank (~1 m 3 ) allows the same initial thermodynamic conditions (1.1 MPa, 170°C) to be obtained as the other one. Internal electrical heating is provided, as well as N 2 inlet. Pressure and temperature change just upstream the test section during an experiment are typically of ± 10 kPa and ± 0.5 °C.

A DANTEC Dual-PDA1 allows local measurements (typically 1 mm 3 volume) of both two components of individual droplet velocity and diameter.

The measurement volume can be moved in the three directions (2.7 m -.6 m -1.5 m).

Thus different measurement locations can be tested during the same experiment. We generally covered the vertical plane which includes the jet axis as described in fig. 5.

Droplet size measurement accuracy

The PDA stops when 3000 measures are collected or after 5 seconds if less than 3000 measures were collected. At 25 and 50 mm downstream of the D2L100 pipe, the apparatus only detects a few droplets (~ 200) during the 5 s which are allowed. The difficulty is probably linked to the optical density of the jet in this region: laser beams generally cannot cross the jet without being refracted. At intermediate distances (generally comprising 200 mm downstream the orifice or pipe) 3000 droplets are measured. The number of measures decreases again farther than 400 or 600 mm. The main reason is probably the decrease in the droplets density. Sometimes droplets impact the lenses of the apparatus… We thus decided to use a distance of 200 mm downstream of the orifice or pipe on the jet axis as a reference location. Another justification for this location is that the change in the size distribution is very important in the first few centimetres and clearly slower after 200 to 400 mm.

We estimate the Sauter mean diameter (d 32 ) from the sample collected by the PDA.

This sample has to be large in order that the estimated value is close to the actual one. Most of the measured droplets are "small droplets" (d d < 150 µ m). Figure 6. shows that estimated mean value for the small droplets is correct within a few micrometers when the sample has 3000 droplets. The "larger drops" play a predominant role when estimating d 32 for the whole population because of their large V/S ratio. But these drops are relatively few among the whole population (~ 40 à 50 % at T res ~ 105°C, 3 to 10% at T res ~ 165°C). When the apparatus stops because he counted 3000 droplets, the sample of larger ones is still limited (100 and 300 droplets for the two tests depicted in figure 6). It is the reason why the convergence is slower. We can estimate the d 32 uncertainty as ± ± ± ± 100 µm for a sample of 3000 droplets.

Results and discussion

Tested reservoir conditions are all sub-cooled. They are summarized in figures 7.

(pipes) and 8. (orifices). Mass flow rates from experiments with pipes at temperatures lightly higher than the boiling one (up to 120 °C) also show a complete agreement ( ± 5%) with purely liquid flow assumption (constant linear friction coefficient: λ c = 0.016). Figure 10. represents mass velocities measured at higher temperatures (T > 120°C). The agreement with the Bernoullilike approach (or Lackmé approach: G =

Flow type

)) ( ( 2

res sat res l T P P - ρ
) is quite good. It is better than using a model for initially saturated fluid [START_REF] Fauske | Flashing flows or: some practical guidelines for emergency releases[END_REF] for example) even at low subcooling (10 to 20 kPa here). This means that the flow is liquid in the main part of the pipe but there is some form of vaporisation just upstream its end. [START_REF] Hocquet | Loss of containment : experimental aerosol rain-out assessment[END_REF]. This suggests a large influence of pressure gradient (P res -P sat (T res ))/L on twophase flow structure inside the pipe. The 2-phase flow region is much shorter for the shorter pipe, so that almost no internal fragmentation can occur (when ∆P/L > 1 MPa/m instead of ∆P/L < .1 MPa/m). This leads to the small pipe (D2L100) behaving like an orifice. Our pipe measurements are representative of such highly sub-cooled flows only. All measurements we made were very similar downsrtream orifices and pipes. We thus will subsequently present orifice data only. Two important features of these data are that there is no significant influence of the reservoir pressure on rain-out fraction and that data after a pipe are not so different from ones obtained after an orifice. The agreement is also good with water data from the CCPS experiments [START_REF] Johnson | RELEASE A model with data to predict aerosol Rain-out in accidental Releases[END_REF].

Rain-out measurements

Droplet diameter measurements; Number and mass distribution

Figures 13. shows that all distributions have a peak in the [0, 150 µm] range, whatever T res et P res ,. Similar distributions are again obtained downstream a pipe. The shape of this peak is similar to what was reported in the literature [START_REF] Brown | Sprays formed by flashing liquid jets[END_REF][START_REF] Bushnell | Atomisation of superheated Water jets at low ambient pressures[END_REF][START_REF] Gooderum | Measurement of Mean Drop Sizes for Sprays from Superheated Water jets[END_REF][START_REF] Lienhard | The break-up of superheated liquid jets[END_REF][START_REF] Oza | An Experimental and Analytical Study of Flash-boiling Fuel Injection[END_REF][START_REF] Oza | On the Mechanism of Flashing Injection of Initially Sub-cooled Fuels[END_REF][START_REF] Razzaghi | Droplet size estimation of two phase flashing jets[END_REF][START_REF] Reitz | A Photographic Study of Flash-Boiling Atomisation[END_REF][START_REF] Park | An experimental investigation of the flash atomisation mechanism[END_REF][START_REF] Ramsdale | Droplet formation and rain-out from two phase flashing jets[END_REF][START_REF] Witlox | Unified Dispersion Model[END_REF][START_REF] Witlox | Flashing liquid jets and two-phase dispersion[END_REF], qualitatively near a log-normal or a Rosin-Rammler distribution.

Our measurements show that another population seems to exist. At low temperature (T res = 80°C, 104° C and 124°C), one observes a quasi-uniform distribution (almost the same number of droplets in each size class) up to 600 m µ (PLDA limit for the used configuration). At ). The sensitivity of this fraction to pressure seems to be low, probably lower than the measurement uncertainty.

It seems to us that such a large fraction of large drops was never mentioned in the literature up to now especially in case of thermal fragmentation. This could be due to experimental considerations : in order to obtain best measurements where most of the droplets are present, it is often better to limit the measurement range. With our apparatus configuration the signal was very weaker in the [0, 600 µm] (mask C) range than in the [0, 200 µm] (mask A) range. Another reason is that most of the literature is devoted to very small orifices, generally less than 500 m µ ( Diesel injection for example). However preliminary experiments using shadow visualization of droplets under the aerosol confirm that they have diameters larger than 150 µm.

Change with P res et T res

The distribution which appeared as quasi-uniform in number at the lower temperatures (T res < 125°C) appears increasing with diameter when represented in mass because mass increases as diameter to the cube. The mass representation shows that the same distribution still exists at the higher temperatures for the larger droplets (we define them from the histograms as : d d > 325 m µ ). Further observing the mass histograms we are tempted to distinguish a third population: a peak between 150 m µ and 325 m µ is almost absent at the lower temperatures (T res < 105°C), but grows when temperature increases. When temperature is higher than T res = 136°C its mass is more than that of « small droplets ». These comments apply to both D2 orifice and D2L100 pipe. This would mean that the thermal fragmentation would only concern a fraction of the liquid, with the "larger drops" staying in a meta-stable state ! Fig 16 . shows that the mass fraction for these droplets is very high, even if it decreases when temperature increases: 90 % at T res ~ 105°C, still ~ 60 % at T res ~ 165°C. Perhaps the number of nuclei is not large enough for thermal fragmentation to be complete. This type of fragmentation would be more intensive at higher temperatures because more nuclei would be activated This means that there exists at least one external mechanism which breaks the droplets mainly in the first few centimetres, and continues during 200 to 400 mm.

Axial change of mean size

The Sauter mean diameter decreases when the number of « small droplets » increases (from pipe exit to ~ 200-400 mm downstream; fig. 18.). It can then increase when their relative weight decreases (farther than 200 to 400 mm). Their relative weight decreases slower when temperature is lower, which could explain that the trend for d 32 to increase comes farther. Note that the observed diameter increase is generally smaller than the uncertainties but the trend is observed for all the experiments.

Conclusion

Two pilot scale experimental set-up allowed us to characterize flashing water jets. We measured mass flow rate, initial droplet size and velocity distributions, plus spatial distribution of rain-out. Test sections were both orifices and pipes, mainly 2 mm inner diameter.

Mass flow rate measurements and spatial distribution of rain-out showed that almost no internal fragmentation in the pipe occurs when reservoir sub-cooling is large enough (∆P / L > 1 MPa / m). In such conditions both pipes and orifices exhibit the same jet behaviour (rain-out fraction and rain-out spatial distribution).

Reservoir pressure (P res ) has not a significant influence on total rain-out fraction as compared to reservoir temperature T res .

All droplet size distributions exhibit a peak in the [0 ÷ 150 µm] range as described in the literature. But some larger droplets are also detected. These last ones count fore more than 85 % of the liquid mass even when they count for less than 10 % of the droplets number!

The "small droplets" (d d < 150 µm) form mainly in the first few centimetres with the maximum number reached between 200 and 400 mm downstream the orifice or pipe. Mass histograms show that another peak forms between 150 and 325 µm. These two peaks mean that an external secondary fragmentation mechanism exists.

The former observations lead to the following scheme:

-A part of the liquid phase undergoes very rapidly (mostly less than 10 mm) a mechanical fragmentation and probably a thermal fragmentation. This leads to a population of "small droplets" (d d < 150 µm). This mechanism is stronger at the higher temperatures.

-When temperature increases another fraction of the droplets is concerned with thermal fragmentation. This leads to the "medium droplets" (150 µm < d d < 325 µm).

-An important residual quantity of droplets do not undergo secondary fragmentation.

They constitute the "large droplets" population ( 325 µm < d d < 600 µm).

The large mass fraction of large droplets (d d > 150 µm) could be an explanation for our former observation that rain-out is heterogeneous in nature [START_REF] Hocquet | Loss of containment : experimental aerosol rain-out assessment[END_REF] : it is composed of droplets which fall under the almost horizontal two-phase jet. Our next aim will be to prove that such size distributions can be a basis in order to improve rain-out prediction models. Future work will also have to take into account near saturated reservoir conditions and fluids other than water.

Figure 9 .(

 9 Figure 9. shows mass velocity (W/A) data for orifices (A is the nominal section of the

Figure 11 .

 11 Figure 11. shows typical spatial distributions for the rain-out downstream both orifice

Figure 12 .

 12 Figure 12. shows the global rain-out we measured versus reservoir temperature T res .

  higher temperatures (167°C) a few droplets with diameter larger than 150 µm still exist, most of them between 200 et 300 m µ . Droplet mass is more important than droplet number as far as rain-out phenomenon is concerned. This led us to represent size distribution using the mass probability density function. Figures 14. shows a completely different trend. The "small droplets" (dd < 150 µm), which represent the larger number ( τ > 90 % for T res ≥ ≥ ≥ ≥ 136°C and still τ > 45 % for T res ≥ ≥ ≥ ≥ 80°C) only represent 0,5 to 17 % of the mass! And the large ones count for the complement, i.e. 83 to 99,5 % of the mass! The mass fraction for the « large droplets » in our temperature range is always the larger one. It represents almost all the mass when T res < 125°C. It then slowly decreases to approximately 85% at T res ~ 165°C (fig. 15.

  Histogram at 80°C (fig.13.) can only be understood as resulting from pure mechanical fragmentation. It consists of a quasi-uniform distribution between 0 and 600 µm, plus a peak of very fine droplets (around d d = 15 µm). The histograms at 104°C or 124°C are qualitatively similar. The only difference is that the peak moved to lightly larger diameters. Fragmentation at such T res probably still corresponds to essentially mechanical fragmentation.Thermal fragmentation is thus not the only one which can form very fine droplets (d d < 50 µm). Thus we cannot decide if the peak of "small droplets" (d d < 150 µm, more than 90 % of the droplets at high temperature) has a thermal or mechanical origin. There is perhaps superimposition of a mechanical peak and a thermal one at lightly larger diameters. The growth of the thermal peak could eventually explain that the global peak moves towards slightly larger diameters when temperature increases.The "medium" peak (150 µm < d d <325 µm) is almost absent at the lower temperatures (fig.15.) (T res < 125°C). It is visible at T res ~ 165°C. This leads us to associate it with thermal fragmentation. The "larger drops" (d d > 325 µm) would be a remaining part of the primary fragmentation which would not have undergone secondary one.

Fig. 16 .

 16 Fig. 16. represents the probability density function data at different locations

Fig. 1 .

 1 Fig. 1.: scheme of the test sections (orifices and pipes)

Fig. 4 .

 4 Fig. 4.: scheme of the set-up for droplet size measurement

Figure 9 .

 9 Figure 9. : Mass velocity versus relative reservoir pressure (orifices)

Fig. 12 .Figure 9 Fig

 129 Fig. 12. : rain-out fraction versus reservoir temperature.

2 300 mW argon lasers in two orthogonal planes, 60X Fiber Flow series, 57X80 and 58N81 detectors, BSA P70 electronic module, mask C.