Miguel Liroz-Gistau
email: miguel.liroz_gistau@inria.fr

Reza Akbarinia
email: reza.akbarinia@inria.fr

Divyakant Agrawal
email: agrawal@cs.ucsb.edu

Patrick Valduriez
email: patrick.valduriez@inria.fr

FP-Hadoop: Efficient Processing of Skewed MapReduce Jobs

Keywords: MapReduce, Data Skew, Parallel Data Processing

Nowadyas, we are witnessing the fast production of very large amount of data, particularly by the users of online systems on the Web. However, processing this big data is very challenging since both space and computational requirements are hard to satisfy. One solution for dealing with such requirements is to take advantage of parallel frameworks, such as MapReduce or Spark, that allow to make powerful computing and storage units on top of ordinary machines. Although these key-based frameworks have been praised for their high scalability and fault tolerance, they show poor performance in the case of data skew. There are important cases where a high percentage of processing in the reduce side ends up being done by only one node.

In this paper, we present FP-Hadoop, a Hadoop-based system that renders the reduce side of MapReduce more parallel by efficiently tackling the problem of reduce data skew. FP-Hadoop introduces a new phase, denoted intermediate reduce (IR), where blocks of intermediate values are processed by intermediate reduce workers in parallel. With this approach, even when all intermediate values are associated to the same key, the main part of the reducing work can be performed in parallel taking benefit of the computing power of all available workers.

We implemented a prototype of FP-Hadoop, and conducted extensive experiments over synthetic and real datasets. We achieved excellent performance gains compared to native Hadoop, e.g. more than 10 times in reduce time and 5 times in total execution time.

Introduction

In the past few years, advances in the Web have made it possible for the users of information systems to produce large amount of data. However, processing this big data is very challenging since both space and computational requirements are hard to satisfy. One solution for dealing with such requirements is to take advantage of parallel frameworks, such as MapReduce [START_REF] Dean | MapReduce: Simplified data processing on large clusters[END_REF] or its IO-efficient versions such as Spark [START_REF] Zaharia | Resilient distributed datasets: A fault-tolerant abstraction for in-memory cluster computing[END_REF], that allow to make powerful computing and storage units on top of ordinary machines.

The idea behind MapReduce is simple and elegant. Given an input file of key-value pairs, and two functions, map and reduce, each MapReduce job is executed in two main phases. In the first phase, called map, the input data is divided into a set of splits, and each split is processed by a map task in a given worker node. These tasks apply the map function on every key-value pair of their split and generate a set of intermediate pairs. In the second phase, called reduce, all the values of each intermediate key are grouped and assigned to a reduce task. Reduce tasks are also assigned to worker machines and apply the reduce function on the created groups to produce the final results.

Although MapReduce and Spark frameworks have been praised for their high scalability and fault tolerance, they show poor performance in the case of data skew. There are important cases where a high percentage of processing in the reduce side ends up being done by only one node. Let's illustrate this by an example.

Example 1. Top accessed pages in Wikipedia. Suppose we want to analyze the statistics1 that the free encyclopedia, Wikipedia, has published about the visits of its pages by users. In the statistics, for every hour, there is a file in which for each visited page, there is a line containing some information including, among others, its URL, language and the number of visits. Given a file, we want to return for each language, the top-k% accessed pages, e.g., top 1%.

To answer this query, we can write a simple program as in the following Algorithm 2 : map( id : K1, content : V1 ) foreach line lang, page id, num visits, ... in content do emit (lang, page info = num visits, page id ) end reduce( lang : K2, pages info : list(V2) )

Sort pages info by num visits foreach page info in top k% do emit (lang, page id ) end Algorithm 1: Map and reduce functions for Example 1

In this example, the load of reduce workers may be highly skewed. In particular, the worker that is responsible for reducing the English language will receive a lot of values. According to the statistics published by Wikipedia 3 , the percentage of English pages over total was more than 70% in 2002 and more than 25% in 2007. This means for example that if we use the pages published up to 2007, when the number of reduce workers is more than 4, then we have no way for balancing the load because one of the nodes would receive more than 1/4 of the data. The situation is even worse when the number of reduce tasks is high, e.g., 100, in which case after some time, all reduce workers but one would finish their assigned task, and the job has to wait for the responsible of English pages to finish. In this case, the execution time of the reduce phase is at least equal to the execution time of this task, no matter the size of the cluster.

There have been some proposals to deal with the problem of reduce side data skew. One of the main approaches is to try to uniformly distribute the intermediate values to the reduce tasks, e.g., by dynamically repartitioning the keys to the reduce workers [START_REF] Kwon | Skewtune: mitigating skew in mapreduce applications[END_REF]. However, this approach is not efficient in many cases, e.g., when there is only one single intermediate key, or when most of the values correspond to one of the keys.

One solution for decreasing the reduce side skew is to filter the intermediate data as much as possible in the map side, e.g., by using a combiner function. However, the input of the combiner function is restricted to the data of one map task, thus its filtering power is very limited for some applications. Let's illustrate this by using our problem of top-1%. Suppose we have 1TB of Wikipedia data, and 200 nodes for processing them. To be able to filter some intermediate data by the combiner function, we should have more than 1% of the total values of at least one key (language) in the map task. Thus, if we use the default splits of Hadoop (64 MB size), the combiner function can filter no data. The solution is to increase significantly the size of input splits, e.g. more than 10GB (1% of total). However, using big splits is not advised since it decreases significantly the MapReduce performance due to the following disadvantages: 1) more map-side skew : with big 3 http://en.wikipedia.org/wiki/Wikipedia:Size of Wikipedia splits, there may be some map tasks that take too much time (e.g. because of their slow CPU), and this would increase significantly the total MapReduce execution time; 2) less parallelism: big split size means small number of map tasks, so several nodes (or at least some of their computing slots) may have nothing to do in the map phase. In our example, with 10GB splits, there will be only 100 map tasks, thus half of the nodes are idle. This performance degradation is confirmed by our experimental results reported in Section 5.11).

In this paper, we propose FP-Hadoop, a Hadoop-based system that uses a novel approach for dealing with the data skew in reduce side. In FP-Hadoop, there is a new phase, called intermediate reduce (IR), whose objective is to make the reduce side of MapReduce more parallel. More specifically, the programmer replaces his reduce function by two functions: intermediate reduce (IR) and final reduce (FR) functions. Then, FP-Hadoop executes the job in three phases, each phase corresponding to one of the functions: map, intermediate reduce (IR) and final reduce (FR) phases. In the IR phase, even if all intermediate values belong to only one key (i.e., the extreme case of skew), the reducing work is done by using the computing power of all available workers. Briefly, the data reducing in the IR phase has the following distinguishing features:

• Parallel reducing of each key: The intermediate values of each key can be processed in parallel by using multiple intermediate reduce workers.

• Distributed intermediate block construction:

The input of each intermediate worker is a block composed of intermediate values distributed over multiple nodes of the system, and chosen using a scheduling strategy, e.g. locality-aware.

• Hierarchical execution: The processing of intermediate values in the IR phase can be done in several levels (iterations). This permits to perform hierarchical execution plans for jobs such as top-k% queries, in order to decrease the size of the intermediate data more and more.

• Non-overwhelming reducing: The size of the intermediate blocks is bounded by configurable maximum value that prevents the intermediate reducers to be overwhelmed by very large blocks of intermediate data.

We implemented a prototype of FP-Hadoop, and conducted extensive experiments over synthetic and real datasets. The results show excellent performance gains of FP-Hadoop compared to native Hadoop. For example, in a cluster of 20 nodes with 120GB of input data, FP-Hadoop outperformed Hadoop by a factor of about 10 in reduce time, and a factor of 5 in total execution time.

This paper is a major extension of [START_REF] Akbarinia | An efficient solution for processing skewed mapreduce jobs[END_REF] and [START_REF] Liroz-Gistau | Fp-hadoop: Efficient execution of parallel jobs over skewed data[END_REF], with at least 30% of new materials. In the current paper, we propose a fault-tolerance mechanism that assures the correctness of the results in the case of failures, and reduces the amount of data to be re-processed compared to the native Hadoop. Additionally, we describe the design in more details and provide a more extensive experimental evaluation.

The rest of this paper is organized as follows. In Section 2, we explain how Hadoop's MapReduce wroks and give the necessary details to present our approach. In Section 3 we present the principles of FP-Hadoop including its programming model. Then, in Section 4, we give more details about its design. In Section 5, we report the results of our experiments done to evaluate the performance of FP-Hadoop. In Section 6, we discuss related work, and Section 7 concludes.

MapReduce Background

In this section, we first briefly explain how MapReduce works in Hadoop. This will be useful to understand the technical details of FP-Hadoop. Then, we give an abstract view of the MapReduce execution. This abstract view is useful to better understand the main differences between the programming models of Hadoop and FP-Hadoop.

Job Execution in Hadoop

In Hadoop, for executing a MapReduce job, we need a master node for coordinating the job execution, and some worker nodes for executing the map and reduce tasks [START_REF] White | Hadoop -The Definitive Guide: Storage and Analysis at Internet Scale[END_REF]. The worker nodes can be configured with a predefined number of slots for map and reduce tasks, so that each slot is able to execute a single task at a given time.

When a MapReduce job is submitted to a node, it computes the input splits. The number of input splits can be personalized, but typically there is a one-to-one relationship between splits and file chunks in the filesystem, which by default have a size of 64MB. The location of these splits and some information about the job are submitted to the master that creates a job object with all the necessary information, including the map and reduce tasks to be executed. One map task is created per input split.

When a map task is assigned to a worker, it is executed in a Java Virtual Machine (JVM). The task reads the corresponding input split, applies the map function on each input element (e.g. line), and generates intermediate key-value pairs, which are firstly maintained in a buffer in main memory. When the content of the buffer reaches a threshold (by default 80% of its size), the buffered data is stored on the disk in a file called spill. Before writing the content of the buffer into the spill, the keys are divided into several partitions (as many as reduce tasks) using a partitioning function, and then the values of each key are sorted and written to its corresponding partition in the spill. An optional combiner function may be applied on the buffer data just before they are written into the spills. The objective of this function is to decrease the size of intermediate data that should be transferred to the reduce workers. Once a map task is completed, the generated spills are merged into a final output file and the master is notified.

In the reduce phase, each partition is assigned to one of the reduce tasks. Each reduce task retrieves the key-value pairs corresponding to its partition from all the map output files, and merges them using the merge-sort algorithm. The transfer of data from map workers to the reduce workers is called shuffling, and can be started when a map task finishes its work. However, the reduce function cannot be applied until all the map tasks have finished and their outputs merged and grouped. Each reduce task groups the values of the same key, applies the reduce function on the corresponding values, and generates the final output results. When, all reduce tasks of a job are completed successfully, the user is notified by the master.

An Abstract View

In an abstract view, the input of the map phase in MapReduce can be considered as a set of data splits, which should be processed by all workers. Thus, each map worker takes one split, processes it, and then takes another one until there are no splits in the set. Thus, we can hope for good parallelism in the map side, since no worker is idle until there is any split in the split set. But, in the reduce side, there is not such a parallelism, because the values of each key should be processed by one reduce worker. Thus, there may be situations where a high volume of the work is done by a single worker or a few number of workers, while the others are idle.

We have made FP-Hadoop more parallel and efficient than Hadoop since all reduce workers can contribute in reducing the map output even if the output belong to only one key.

FP-Hadoop Principles

In this section, we introduce the programming model of FP-Hadoop, its main phases, and the functions that are necessary for executing jobs. The design details of FP-Hadoop development are given in the next section.

Programming Model

In FP-Hadoop, the output of the map tasks is organized as a set of blocks (splits) which are consumed by the reduce workers. More specifically, the intermediate key-value pairs are dynamically grouped into splits, called Intermediate Result Splits (IR splits for short). The size of an IR split is bounded between two values, MinIRSize and MaxIRSize, configurable by the user. Formally, each IR split is a set of (k, V ) pairs such that k is an intermediate key and V is a subset of the values generated for k by the map tasks.

FP-Hadoop executes the jobs in three different phases (see Figure 1

k1 V ′ 1,1 k2 V ′ 2,1 V ′ 2,2 k3 V ′ 3,1
Final reduce workers R1 ...

Rr

Results

O1

...

Or

Map phase

Intermediate reduce phase Final reduce phase the sense that the map workers apply the map function on the input splits, and produce intermediate key-value pairs. The only difference is that in FP-Hadoop, the map output is managed as a set of IR fragments that are used for constructing IR splits. There are two different reduce functions: intermediate reduce (IR) and final reduce (FR) functions.

In the intermediate reduce phase, the IR function is executed in parallel by reduce workers on the IR splits, which are constructed using a scheduling strategy from the intermediate values distributed over the nodes. More specifically, in this phase, each ready reduce worker takes an IR split as input, applies the IR function on it, and produces a set of key-value pairs which may be used for constructing future IR splits. When a reduce worker finishes its input split, it takes another split and so on until there is no more IR splits. In general, programming the IR function is not very complicated; it can be done in a similar way as the combiner function of Hadoop. In Section 3.2, we give more details about the IR function, and how it can be programmed.

The intermediate reduce phase can be repeated in several iterations, to apply the IR function several times on the intermediate data and reduce incrementally the final splits consumed by the FR function (see Figure 2). The maximum number of iterations can be specified by the programmer, or be chosen adaptively, i.e., until the intermediate reduce tasks input/output size ratio is higher than a given threshold.

In the final reduce phase, the FR function is applied on the IR splits generated as the output of the intermediate reduce phase. The FR function is in charge of performing the final grouping and production of the results of the job. Like in Hadoop, the keys are assigned to the reduce tasks according to a partitioning function. Each reduce worker pulls all IR splits corresponding to its keys, merges them, applies the FR function on the values of each key, and generates the final job results. Since in FP-Hadoop the final reduce workers receive the values on which the intermediate workers have worked, the load of the final reduce workers in FP-Hadoop is usually much lower than that of the reduce workers in Hadoop.

In the next subsection, we give more details about the IR and FR functions, and explain how they can be programmed.

IR and FR Functions

To take advantage of the intermediate reduce phase, the programmer should replace his/her reduce function by intermediate and final reduce functions. Formally, the input and output of map (M), intermediate (IR) and final reduce (FR) functions is as follows:

M : (K 1 , V 1 ) → list(K 2 , V 2 ) IR : (K 2 , partial list(V 2 )) → (K 2 , partial list(V 2 )) FR : (K 2 , list(V 2 )) → list(K 3 , V 3 )
Notice that in IR function, any partial set of intermediate values can be received as input. However, in FR function, all values of an intermediate key are passed to the function.

Given a reduce function, to write the IR and FR functions, the programmer should separate the sections that can be processed in parallel and put them in IR function, and the rest in FR function. Formally, given a reduce function R, the programmer should find two functions IR and FR, such that for any intermediate key k and its list of values S, we have:

R(k, S) = FR(k, IR(k, S 1 ), ..., IR(k, S n ) )
for every partition S 1 ∪ ... ∪ S n = S

In Table 1, we enumerate some important functions, and show their intermediate and FR functions. There are many functions for which we can use the original reduce The following example shows how a Top-k query can be implemented in FP-Hadoop using the same function for IR and FR.

Map Phase M • • • M M • • • M M • • • M M • • • M Intermediate Phase Iteration 1 IR • • • IR • • • IR • • • IR • • • Intermediate Phase Iteration 2 IR • • • IR • • • Final Reduce Phase F R0 F R1
R = Top-k IR = FR = Top-k R = SkyLine IR = FR = SkyLine R = Union IR = FR = Union R = Sum IR = FR = Sum R = Max IR = FR = Max R = Min IR = FR = Min R = Avg IR = (Sum,
Example 2. Top-k. Consider a job that given a scoring function computes the top-k tuples of a big table. In this job, the map function computes the score of each read tuple, and emits (key, {tupleID, score}), where key is the identifier of the set of values on which we want to find the top-k values. Then, both IR and FR functions can be implemented as a function that, given a set of {tupleID, score} pairs, returns the k pairs that have the highest scores. Thus, in practice, the intermediate reduce workers generate partial top-k results (top-k tuples of their input IR splits), and the FR function produces the final results from intermediate partial results. If the value of k is big, e.g. 1% of the input as in our motivating example in Introduction, then we may need more iterations in the intermediate phase to reduce the intermediate data more and more. In this case, the execution of the job for overloaded key(s) can be hierarchical as in Figure 2. To plan such executions, it is sufficient to set the maximum number of iterations to the required level, and then FP-Hadoop organizes multiple iterations for overloaded keys when it is needed .

A class of functions that can usually take advantage of the intermediate reduce phase is that of aggregate functions. Examples of such functions are Sum, Min/Max (using the same function as IR and FR), Avg and Std (using different functions for IR and FR). Aggregate functions can be classified into three groups [START_REF] Gray | Data cube: A relational aggregation operator generalizing group-by, cross-tab, and sub totals[END_REF]:

• Definition 1 (Distributive). Let F be a reduce function, and S a set of values, and

S 1 •...•S n a partition- ing over S. F is distributive if there is a function G such that F (S) = G( F (S 1 ), ..., F (S n ) ).
• Definition 2 (Algebraic). Let F be a reduce function, and S a set of values, and S 1 • ... We precise that if it is difficult to find an efficient IR function for a job, it is sufficient to specify no IR function, then the final reduce phase starts just after the map phase, i.e., like in Hadoop.

Design Details

In this section, we describe our design choices in FP-Hadoop. We focus on the activities that do not exist in Hadoop or are very different in FP-Hadoop, particularly: 1) management of IR fragments used for making IR splits; 2) intermediate task scheduling; 3) intermediate and final reduce task creation; 4) management of multiple iterations in the intermediate reduce phase.

IR Fragments for Constructing IR Splits

In this subsection, we describe our approach for constructing the IR splits that are the working blocks of the reducers in the intermediate reduce phase.

In Hadoop, the output of the map tasks is kept in the form of temporary files (called spills). Each spill contains a set of partitions, such that each partition involves a set of keys and their values. These spills are merged at the end of the map task, and the data of each partition is sent to one of the reducers.

In FP-Hadoop, the spills are not merged. Each partition of a spill generates an IR fragment, and the IR fragments are used for making IR splits. When a spill is produced by a map task, the information about the spill's IR fragments, which we call IRF metadata, is sent to the master node by using the heartbeat message passing mechanism 4 . An IR fragment is uniquely identified by the task which produced it, its spill number and the partition. The data flow between FP-Haddop components is shown in Figure 3.

Notice that the choice of using spills as the output unit of the map phase provides the following advantages. The intermediate phase may start as soon as sufficient spills have been produced, even if no map task has yet finished, thus increasing the parallelism. Moreover, the fact that spills are bounded in size guarantees that IR split maximum size is respected. This would not be the case if the whole output of a map task is used, since its size is unbounded and could be by itself bigger than the IR split limit.

For keeping IRF metadata, the master of FP-Hadoop uses a specific data structure called IR fragment table (IRF Table ). Each partition has an entry in IRF Table that points to a list keeping the IR fragment metadata of the partition, e.g., size, spill and the ID of the map worker where the IR fragment has been produced. The master uses the information in IRF Table for constructing IR splits and assigning them to ready reduce workers. This is done mainly based on the scheduling strategies described in the next subsection.

Scheduling Strategies

The master node is responsible for scheduling the intermediate and final reduce tasks. For this, it uses a component called Reduce Scheduler that schedules the tasks by using a customizable scheduling strategy. Here we describe the scheduling strategies which are used in FP-Hadoop; we present the details of Reduce Scheduler in Section 4.4.

For scheduling an intermediate reduce task, the most important issue is to choose the IR fragments that belong to the IR split that should be processed by the task. The strategies, which are currently implemented in FP-Hadoop are:

• Greedy: In this strategy, IR fragments are chosen from within the biggest partition, i.e., the partition whose IR fragments account for the highest total size. In the IRF Table , for each partition, along with the list of IR fragments we store its total size, allowing to chose the partition by a simple vertical scan of IRF Table . After choosing the partition, we select IR fragments starting from the head of the list until reaching the MaxIRSize value, i.e., the upper bound size for an IR split. This strategy is the default strategy in FP-Hadoop.

• Locality-aware: In this strategy, the objective is to choose for a worker w the IR fragments that are on its local disk or close to it. Consequently, we chose the partition p for which the IR fragments produced in w account for the biggest size, provided that p's total size is as least MinIRSize. From the partition's fragments, we select those local to w until reaching MaxIRSize. If their total size is inferior to MinIR-Size we keep taking fragments until reaching MinIR-Size, first choosing from those produced in the same rack as w and then from the same data center.

Multiple Iterations

FP-Hadoop can be configured to execute the intermediate reduce phase in several iterations, in such a way that the output of each iteration is consumed by the next iteration. Notice that the output of IR tasks in phase n produces the IR fragments that are to be consumed in phase n + 1.

An example of Top-k query execution over Wikipedia data (described in Section 5) is shown in Figure 4, where each phase is depicted as a rectangle whose height represents its input size and its length the execution time in seconds. In this example, there are two iterations for the intermediate reduce phase (shown with gray rectangles). We can see how each iteration further reduces the intermediate data size so that when the final reduce phase (in blue) is executed, its input size is sufficiently small.

In FP-Hadoop, there is a parameter MaxNumIter that defines the maximum number of iterations. Notice that this parameter just establishes the maximum number, but in practice each partition may be processed in a different number of iterations, for instance depending on its size, input/output ratio or skew. FP-Hadoop provides the following strategies for configuring the number of iterations:

• Size-based: In this approach, an iteration is launched only if its input size is more that a given threshold, MinIterSize. By default, FP-Hadoop uses the same value as MinIRSize (i.e. the minimum size of IR splits). The actual number of iterations may vary among partitions, and the maximum number may not be reached, e.g. if the size of intermedi-ate data was originally small or has been sufficiently reduced during previous iterations. The extreme scenario corresponds to the case when the map output for a given partition is less than MinIterSize. In this case, the final reduce phase is directly launched.

• Input/Output-Ratio-based: In this approach, an iteration is launched when the ratio between the input and output size of the previous iteration is greater than a system parameter IORatio. The rationale is that if the input/output ratio is lower than some value, then further intermediate iterations will not help reducing the size of the partition in consideration, and hence, will not improve the job execution performance.

• Skew-based: In this approach, an intermediate iteration is launched for a given partition if its size is SkewRatio times higher than the average partition size. Thus it only executes more iterations for the overloaded partitions. For example by setting SkewRatio equal to 2, only the partitions whose size is at least twice the average size are consumed in a new iteration, and the others are sent to the final reduce phase.

Reduce Scheduler

To schedule the intermediate and final reduce tasks, the master node uses a component called Reduce Scheduler. The scheduling is done using a generic algorithm whose pseudo-code is shown in Algorithm 2. The algorithm scans IRF Table, and selects the best partition for constructing the IR split which should be consumed by the worker. Then, based on the size of the data in the partition and the maximum number of iterations, it decides whether to launch a final reduce task or an intermediate task. If it decides to launch a final reduce task, it uses all IR fragments of the partition. In the other case, i.e., an intermediate reduce task, it chooses the fragments of the IR split from the best partition based on the scheduling strategy.

The main functions that are used in the generic scheduling algorithm (Algorithm 2) are as follows:

• isBestPartition(). Based on the scheduling strategy, this function selects a partition from which the IR fragments of the task will be chosen. For example, with the Greedy strategy, it selects the partition that has the biggest data size among the partitions satisfying the scheduling constraints, i.e., their total size is at least MinIRSize, and the number of fragments of the partition are at least MinIRFs, which can be configured by the user.

• shouldRunFR(). This function decides if the final reduce task should be launched for the data of a partition. The decision depends on the number of iterations that should be executed in the intermediate Algorithm 2: Generic scheduling algorithm reduce phase. If the current iteration for a partition is the last iteration to be done, and the tasks of the iteration have finished successfully, the final reduce task can be scheduled.

• create FR Task(). This function creates a final reduce task. It simply uses all the IR fragments from the partition given by isBestPartition function.

• create IR Task(). This function creates an intermediate reduce task. It takes the partition given by isBestPartition function, and chooses a set of IR fragments as the IR split of the task. The fragments are chosen based on the scheduling strategy, for instance local fragments would be favored in the locality-aware strategy, whereas in the basic Greedy scheduling strategy, IR fragments are added to the task's IR split while they do not surpass MaxIRSize.

The task metadata are sent to the ready worker (see Figure 3).

To implement new scheduling strategies in Reduce Scheduler, usually it is sufficient to change the above functions. For instance, to implement locality-aware strategy instead of Greedy strategy, we changed isBestPartition function (that selects the partition to use) in such a way that it gives the priority to the partition that has the maximum IR fragments generated in the ready worker, and then changed generateIntermediateRT function to choose the IR fragments accordingly, e.g., favor local fragments.

Fault Tolerance

In Hadoop, the master marks a task as failed either if an error is reported (e.g., an Exception of the program or a sudden exit of the JVM) or if it has not received a progress update from the task for a given period [START_REF] Dean | MapReduce: Simplified data processing on large clusters[END_REF]. In both cases, the task is scheduled for re-execution, if possible in a different node. During the execution of a job, the output of completed map tasks should be kept available in the workers' disks until the job finishes. The reason is that any reduce task reads data from all map tasks' output. If a worker fails, intermediate data stored on its disk is not available any more. Consequently, in addition to running tasks, all successfully completed map tasks executed in the worker need to be rescheduled to make their output available again.

In FP-Hadoop, the behavior in the case of failure is slightly different, because we have to consider intermediate reduce tasks and we can also profit from the fact that they only read data from a subset of the map tasks. As in Hadoop, we distinguish two scenarios: task failure and worker failure.

Task Failure

For map and final reduce tasks, the behavior is exactly the same as in native Hadoop, that is, the failed task is scheduled for re-execution, if possible in a different node.

In the case of IR tasks, we need to re-inject their input IR fragments metadata into the IRF Table . Then, the fragments will be assigned to another task, not necessarily grouped in the same IR Split. Since new IRFs may be available, the reduce scheduler may choose to combine them in different ways.

The case of an IR task failure is illustrated in the example shown in Figure 5. The state of the IRFs list corresponding to a given partition p in the IRF Table is presented before and after each event. In time t = t 1 task ir i is scheduled with input fragments f 1 and f 2 . Then, in time t = t 2 the task fails and, consequently, its input fragments are injected back in the list. Notice that since t 1 some fragments have already been consumed (e.g. f 3 ) and new fragments inserted (e.g. f 6 ). Finally, when an idle worker asks for a new task in time t = t 3 , fragments are grouped differently and its input is composed of fragments f 6 and f 1 .

Worker Failure

If a worker node fails, if in the node there were running tasks, they need to be re-scheduled. Map tasks are simply re-executed and the input fragments of intermediate and final reduce tasks are re-injected into the IRF Table .  In FP-Hadoop, as opposed to native Hadoop, not all completed tasks' output is needed in the future, since their data may have already been consumed and reduced by intermediate reduce tasks in subsequent iterations of the execution tree. Indeed, FP-Hadoop uses a mechanism that only requires the re-execution of a minimal amount of tasks in the case of a worker failure.

Each IR fragment stores within its metadata the information about their provenance, that is, a reference to the fragments that were used on its generation. In the failure of a worker, just after input IR fragments of running tasks are reinserted to the IRF Table, each fragment stored in the failed node is replaced by the fragments that were used on its construction. If within those fragments, there are some fragments that are stored in the failed node, they are replaced by their predecessors as well. Fragments are replaced recursively by their predecessors until they are available or they have been generated in a map task. Only in that case, the map task is scheduled for re-execution. Re-injected fragments can be consumed by new IR tasks following the standard procedure.

Let's illustrate this with the example of Figure 6, in which the execution tree for a given partition p is shown. Reduce task r was running in the worker node when it failed. IR fragments stored in the failed node are shadowed. The input fragments f 12 and f 14 are re-inserted to the IRF Table as in a single task failure. Then, unavailable fragments are replaced recursively by their predecessors. In this way, f 14 is replaced by f 11 and f 12 and then f 12 by f 7 and f 8 . No map task needs to be re-executed and the re-injected fragments can be consumed again by new tasks.

FP-Hadoop recovery strategy only re-processes the data that is needed to finish the execution of the job, as opposed to native Hadoop, which needs to re-execute all map tasks and non-finished reduce tasks executed in the failed worker. Furthermore, as IR tasks have already reduced the size of the intermediate data, the amount of data to be treated is reduced even more, thus accelerating the recovery process.

Performance Evaluation

We implemented a prototype of FP-Hadoop by modifying Hadoop's components. In this section, we report on the results of our experiments for evaluating the performance of FP-Hadoop 5 . We first discuss the experimental 5 The executable code of FP-Hadoop as well as the tested jobs are accessible in the following page: http://gforge.inria.fr/plugins/mediawiki/wiki/fp-hadoop/ setup such as the datasets, queries and the experimental platform. Then, we discuss the results of our tests done to study the performance of FP-Hadoop in different situations, particularly by varying parameters such as the number of nodes in the cluster, the size of input data, etc.

Setup

We run the experiments in the Grid500 platform6 in a cluster with up to 50 nodes. The nodes are provided with Intel Quad-Core Xeon L5335 processors with 4 cores each, and 16GB of RAM. Nodes are connected through a switch providing a Gigagbit ethernet connection to each node.

We compare FP-Hadoop with standard Apache Hadoop and SkewTune [START_REF] Kwon | Skewtune: mitigating skew in mapreduce applications[END_REF] which is the closest related work to ours (see a brief description in Related Work Section).

In all our experiments, we use a combiner function (for Hadoop, FP-Hadoop and SkewTune) that is executed on the results of map tasks before sending them to the reduce tasks. This function is use to decrease the amount of data transferred from map to reduce workers, and so to decrease the load of reduce workers.

The results of the experiments are the average of three runs. We measure two metrics:

• Execution time. This is the time interval (in seconds) between the moment when the job is launched and the moment when it ends. This is our default metrics reported for most of the results.

• Reduce time. We use this metric to consider the time that is used only for shuffling and reducing. It is measured as the time interval between the moment when the last map task is finished and the end of the job.

With respect to Hadoop's configuration, the number of slots was set to the number of cores. All the experiments are executed with a number of reduce workers equal to the number of machines. We change io.sort.factor to 100, as advised in [START_REF] White | Hadoop -The Definitive Guide: Storage and Analysis at Internet Scale[END_REF], which actually favors Hadoop. For the rest of the parameters, we employ Hadoop's default values.

The default values for the parameters which we employ in our experiments are as follows. The default number of nodes which we use in our cluster is 20. Unless otherwise specified, the input data size in the experiments is 20 GB.

In FP-Hadoop, we use the default Greedy scheduling strategy as the high throughput of the network limits the impact of the locality-aware strategy. The default value for MinIRSize is set to 512 MB. The value of MaxIRSize is always twice as that of MinIRSize, and the maximum number of iterations is set to 1.

Queries and Datasets

We use the following combinations of MapReduce jobs and datasets to assess the performance of our prototype:

Top-k% (TK). This job, which is our default job in the experiments, corresponds to the query from the Wikipedia example described in the introduction of the paper. The input dataset is stored in the form of lines with the schema: Visits(language, article, num views, other data)

Our query consists on retrieving for each language the k% most visited articles. The default value of k is 1, i.e., by default the query returns 1% of the input data. We have used real-world and synthetic datasets. The realworld dataset (TK-RD) is obtained from the logs about Wikipedia page visits 7 consisting of a set of files each containing the statistics collected for a single hour. We also produced two synthetic datasets, in which we can control the number of keys and their skew. In the first synthetic dataset (TK-SK), which is the default dataset, the number of articles per language follows a Zipfian distribution function

f (l, S, N ) = 1/l s N n=1
(1/n s ) that returns the frequency of rank l, where S and N are the parameters that define the distribution, i.e., f (1, S, N ) returns the frequency of the most popular language. The default value for Zipf exponent parameter (S) is 1, and the parameter N is equal to the number of languages (10 by default). In the second synthetic dataset (TK-U), the articles are uniformly distributed in the keys.

We perform several tests varying the data size, among other parameters, up to 120GB. The query is implemented using a secondary sort [START_REF] White | Hadoop -The Definitive Guide: Storage and Analysis at Internet Scale[END_REF], where intermediate keys are sorted first by language and then by the article's number of visits, but only grouped by language.

Inverted Index (II). This job consists of generating an inverted index with the words of the English Wikipedia articles 8 , as in [START_REF] Kwon | Skewtune: mitigating skew in mapreduce applications[END_REF]. We use a RADIX partitioner to map letters of the alphabet to reduce tasks and produce a lexicographically ordered output. We execute the job with a dataset containing 20GB of Wikipedia articles.

PageRank (PR). This query applies the PageRank [8] algorithm to a graph in order to assign weights to the vertices. As in [START_REF] Kwon | Skewtune: mitigating skew in mapreduce applications[END_REF] we use the implementation provided by Cloud9 9 . And as dataset, we use the PLD graph from Web Data Commons 10 whose size is about 2.8GB.

Wordcount (WC). Finally, we use the wordcount job provided in Hadoop standard framework. We apply it to a dataset generated with the RandomWriter job provided in the Hadoop distribution. We test this job with a 100GB dataset.

7 http://dumps.wikimedia.org/other/pagecounts-raw/ 8 http://dumps.wikimedia.org/enwiki/latest/ 9 http://www.umiacs.umd.edu/ jimmylin/Cloud9/docs/index.html 10 http://webdatacommons.org/hyperlinkgraph/

Scalability

We investigate the effect of the input size on the performance of FP-Hadoop compared to Hadoop. Using TK-SK dataset, Figures 7b and7a show the reduce time and execution time respectively, by varying the input size up to 120 GB, MinIRSize set to 5 GB, and other parameters set as default values described in Section 5.1. Figures 7c and7d show the performance using TK-RD dataset with sizes up to 100 GB, while other parameters as default values described in Section 5.1. As expected, increasing the input size increases the execution time of both Hadoop and FP-Hadoop, because more data should be processed by map and reduce workers. However, the performance of FP-Hadoop is much better than Hadoop when we increase the size of input data. For example, in Figure 7a, the speed-up of FP-Hadoop vs Hadoop on execution time is around 1.4 for input size of 20GB, but this gain increases to around 5 when the input size is 120GB. For the latter data size, the reduce time of FP-Hadoop is more than 10 times lower than Hadoop.

The reason for this significant performance gain is that in the intermediate reduce phase of FP-Hadoop the reduce workers collaborate on processing the values of the keys containing a high number of values while in Hadoop a single task has to process all this data. This is illustrated in Figures 7e and7f, where we compare the execution time of the longest reduce tasks of Hadoop and FP-Hadoop for TK-RD and 20GB. We can see that the longest task in Hadoop is the responsible of the poor performance in the reduce phase, which explains the total execution time. In FP-Hadoop, the longest task is 4 times shorter, and this is a consequence of the improved parallelism.

Effect of Cluster Size

We study the effect of the number of nodes of the cluster on performance. Figure 8a shows the execution time by varying the number of nodes, and other parameters set as default values described in Section 5.1. Increasing the number of nodes decreases the execution time of both Hadoop and FP-Hadoop. However, FP-Hadoop benefits more from the increasing number of nodes. In Figure 8a, with 5 nodes, FP-Hadoop outperforms Hadoop by a factor of around 1.75. However, when the number of nodes is equal to 50, the improvement factor is around 4. This increase in the gain can be explained by the fact that when there are more nodes in the system, more nodes can collaborate on the values of hot keys in FP-Hadoop. In opposition, in Hadoop, although using higher number of nodes can decrease the execution time of the map phase, it cannot significantly decrease the reduce phase time, in particular if there are intermediate keys with high number of values.

Effect of the Number of Intermediate Keys

In our tests, we use the attribute language as the intermediate key for grouping the intermediate values. Here, we report the results of our experiments done to study the effect of this parameter on the performance of FP-Hadoop and Hadoop.

Figure 8b shows the execution time with varying the number of keys (languages) from 50 to 5, and other parameters set as default values. When the number of keys is lower than 20, Hadoop cannot take advantage of all available reduce nodes (there were 20 cluster nodes in this test). However, in FP-Hadoop, the intermediate reduce workers can contribute in processing the values of the keys that have high numbers of values. This is why there is a significant difference between the execution time of FP-Hadoop and Hadoop in these cases.

Even for the cases where the number of keys is higher than the number of nodes (i.e., 20), the execution time of FP-Hadoop is better than that of Hadoop, because in these cases, there are keys with high number of values, and Hadoop cannot well balance the load of reduce workers.

For Hadoop, when increasing the number of keys, the execution time decreases until some number of keys and then it becomes constant. We performed our tests with up to 200 keys, and observed that after 60 keys, there is no decrease in the execution time of Hadoop.

Effect of High Skew in Reduce Workers Load

Here, we study the effect of high data skew on performance by varying the zipf exponent (S) used for generating synthetic datasets with Zipfian distribution. The higher is S, the higher is the skew in the size of the data (articles) assigned to the keys (languages).

Figure 8c shows the execution time with varying zipf exponent S from 1 to 5, and other parameters set as default values. The figure shows that the data skew has a big negative impact on the performance of Hadoop, but its impact on FP-Hadoop is slight. The gain factor of FP-Hadoop increases by increasing S. The reason is that by increasing S, there will be intermediate keys with higher number of intermediate values, and these keys are the bottlenecks in Hadoop, because the values of each key are processed by only one reduce worker.

Effect of Data Skew on Different Queries

Figure 8d shows the total execution time of FP-Hadoop by using several queries and their corresponding data as described in Section 5.1. The results show that FP-Hadoop outperforms Hadoop, in all cases except for the word-count query (WC).

The extent of the gain depends on the amount of reduction that can be performed in the intermediate phase as a result of the partial aggregation. This explains while the best case is for TK-Sk, where each IR task can reduce the data back to k tuples and also while the reduction in PR is small, since only the mass contributions can be aggregated while the node structure has to be passed along untouched.

For the word-count query, the execution time of FP-Hadoop is a little bit (3%) higher than Hadoop, and this increase corresponds to the overhead of the intermediate reduce phase. Indeed, in this query, there is no skew in the reduce side, because the partitioner can balance well the reduce load among workers. FP-Hadoop spends some time to detect that there is no skew in the intermediate results, and then launches the final reduce phase. Thus, its execution time is slightly higher than Hadoop. Notice that even this small overhead can be avoided by disabling the intermediate reduce phase.

Effect of Intermediate Data Filtering

Using parameter k in the top-k% query, we can control the amount of data that may be filtered by the intermediate reduce tasks. In fact, in the top-k% query we return k% of the data that have the highest values. Thus, in the output of a map task or input data of an intermediate reduce task, if the amount of data is higher than k% of the total data, then we can keep the k% highest ranked data and filter the rest. Therefore, the lower is k, the higher is the capacity of data filtering by intermediate reduce tasks.

Figure 8e shows the performance of the two systems by varying k in top-k% query and other parameters set as default values. The figure shows that the lower is k, the better is the performance gain of FP-Hadoop. The reason is that with lower k values, the intermediate reduce tasks can filter more intermediate values. In particular, for values lower than 10%, FP-Hadoop can profit well from the filtering in intermediate reduce workers. For values higher than 10%, the data filtering by intermediate reduce workers decreases significantly, this is why the gain of FP-Hadoop reduces. However, even for k values higher than 10%, there is a significant difference between the execution time of FP-Hadoop and Hadoop.

Analysis of FP-Hadoop Parameters

We investigate the effect of the MinIRSize parameter on the performance of FP-Hadoop. For this, we use two configurations: 1) FP-Hadoop: the default configuration described in Section 5.1, with one iteration in the intermediate reduce phase; 2) FP-Hadoop-Iterations: in this configuration, we set the maximum number of iterations to be 10. As discussed in Section 4.3, the real number of iterations may be lower than this maximum value, e.g. because in an iteration the size of a partition may be lower than MinIRSize.

By varying MinIRSize, Figure 9a shows the execution time of FP-Hadoop and FP-Hadoop-Iterations by varying the MinIRSize parameter for processing an input of size 100GB, and other parameters set as default values. The results show that in FP-Hadoop with one iteration, when we set MinIRSize to small values (e.g., lower than 128MB), the execution time is not very good. This suggests to not choose very small values for MinIRSize, when using only one iteration. The reason is that in these cases, we cannot well take advantage of the one iteration in the intermediate reduce phase, since with very small IR Splits the amount of data that can be filtered by intermediate tasks may not by large. However, by using more iterations in FP-Hadoop-Iterations, even with small MinIRSize values, the response time is good, since it uses more iterations to take advantage of the intermediate phase. Figure 9b shows the number of iterations done by FP-Hadoop-Iterations, when using different MinIRSize values. We observe that as MinIRSize increases, less iterations are needed to complete the intermediate phase.

In our experiments, when using FP-Hadoop with one iteration, the best configuration for MinIRSize is in the cases where it is close to the ratio of the map output size over the number of reduce workers.

We also study the number of IR tasks launched by the Reduce Scheduler. Figure 9c depicts the number of scheduled IR tasks vs. the partition sizes for TK-SK with 100GB with MinIRSize of 500MB and 1GB. The relation is clear: the bigger the partition, the higher the number of scheduled IR tasks. This is an expected behavior of FP-Hadoop, which tries to fully take advantage of parallelism for the overloaded partitions.

Comparison with SkewTune

Here, we compare FP-Hadoop with SkewTune [START_REF] Kwon | Skewtune: mitigating skew in mapreduce applications[END_REF]. Figures 10a and 10b show the reduce time and execution time of both approaches, using the data and parameters described in Section 5.1. For these experiments, we downloaded the SkewTune prototype that is publicly accessible 11 . The data which we used are the default data and sizes described in Section 5.1 (e.g. 20GB of data for TK-SK). As the results show, FP-Hadoop can outperform SkewTune with significant factors. This is particularly noticeable for TK-SK, where SkewTune cannot divide the execution of the most popular key into several tasks.

Effect of Map Split Size

As discussed in Introduction of this paper, in order to use the combiner function to decrease the reduce skew 11 https://code.google.com/p/skewtune/ in the jobs such as Top-k%, we need to increase the size of the map splits significantly, giving more chance to the combiner to filter the intermediate data. In this section, we study the effect of increasing the map split size on performance of Hadoop and FP-Hadoop. Notice that in our previous experiments for all compared systems, we used the combiner function, and the default map split size was 64MB which was the default value in Hadoop. Figures 11a and11b show the execution times for top-k% and II queries, respectively. We varied the map split size from 64MB to 4GB, and other parameters set as default values described in Section 5.1. In order to do so, we reinjected the input data with the corresponding file system block size, as in Hadoop the default behavior creates a map split for each file block.

For top-k% query, we observe a slight improvement in performance at the beginning (until the size of 256MB) for both Hadoop and FP-Hadoop. But, the performance degrades significantly for higher split sizes. For the inverted index (II) query, increasing the map split size has also a negative impact. The reason is that although increasing the size of the splits in the map phase may increase the chance of the combiner function to filter more intermediate data, using big size splits increases significantly the execution time of this phase, particularly because less map workers may participate in the map phase. Thus, the conclusion is that the combiner function can not make Hadoop as efficient as FP-Hadoop in processing high skewed data (e.g. in top-k% query) , even by using big size map splits.

Discussion

Overall the performance results show the effectiveness of FP-Hadoop for dealing with the data skew in the reduce side. For example, the results show that for 120GB of input data, FP-Hadoop can outperform Hadoop with factors of 10 and 5 in reduce time and total execution time respectively. This gain increases when augmenting the data size. The results also show that increasing the number of nodes of the cluster can significantly increase the gain of FP-Hadoop compared to Hadoop.

The results also show that there are jobs for which the IR phase has no benefits. This occurs particularly Hadoop that keeps the results of map tasks in memory and transfers them to the reduce tasks via message passing, i.e., without passing via the local disks. M3R is very efficient, but can be used only for the applications in which intermediate key-values can fit in memory. SpongeFiles [START_REF] Elmeleegy | Spongefiles: Mitigating data skew in mapreduce using distributed memory[END_REF] is a system that uses the available memory of nodes in the cluster to construct a distributed-memory, for minimizing the disk spilling in MapReduce jobs, and thereby improving performance. The idea of using main memory for data storage has been also exploited in Spark [START_REF] Zaharia | Resilient distributed datasets: A fault-tolerant abstraction for in-memory cluster computing[END_REF], an alternative to MapReduce, that uses the concept of Resilient Distributed Datasets (RDDs) to transparently store data in memory and persist it to disc only when needed. The concept of intermediate reduce phase proposed in FP-Hadoop can be used as a complementary mechanism in the systems such as Haloop, M3R, SpongeFiles and Spark, to resolve the problem of data skew when reducing the intermediate data.

In MapReduce Online [START_REF] Condie | Mapreduce online[END_REF], instead of waiting for reduce tasks to pull the map outputs, the map tasks push their results periodically to the reduce tasks. This allows to increase the overlap between the map and shuffle phases, and consequently to reduce the total execution time. Similarly, we also benefit from an increased overlap as we do not need to wait for the end of a map task in order to start transferring its output. But, MapReduce Online does not resolve the problem data skew in overloaded keys.

There have been systems proposing new phases to MapReduce in order to deal with special problems. For example in Map-Reduce-Merge [START_REF] Yang | Mapreduce-merge: simplified relational data processing on large clusters[END_REF], in addition to the map and reduce phases, a third phase called merge is added to MapReduce in order to merge the reduce outputs of two different MapReduce jobs. The merge phase is particularly used for implementing multi-join operations. However, Map-Reduce-Merge and other solutions proposed for join query processing, e.g. [START_REF] Afrati | Fuzzy joins using mapreduce[END_REF][START_REF] Okcan | Processing theta-joins using mapreduce[END_REF], cannot be used for resolving the problem of data skew due to overloaded keys.

In general, none of the existing solutions in the literature can deal with data skew in the cases when most of the intermediate values correspond to a single key, or when the number of keys is less than the number of reduce workers. But, FP-Hadoop addresses this problem by enabling the reducers to work in the IR phase on dynamically generated blocks of intermediate values, which can belong to a single key.

Conclusion

In this paper, we presented FP-Hadoop, a system that brings more parallelism to the MapReduce job processing by allowing the reduce workers to collaborate on processing the intermediate values of a key. We added a new phase to the job processing, called intermediate reduce phase, in which the input of reduce workers is considered as a set of IR Splits (blocks). The reduce workers collaborate on processing IR splits until finishing them, thus no reduce worker becomes idle in this phase. In the final reduce phase, we just group the results of the intermediate reduce phase.

By enabling the collaboration of reduce workers on the values of each key, FP-Hadoop improves significantly the performance of jobs, in particular in the case of skew in the values assigned to the intermediate keys, and this is done without requiring any statistical information about the distribution of values.

We evaluated the performance of FP-Hadoop through experiments over synthetic and real datasets. The results show excellent gains compared to Hadoop. For example, over a cluster of 20 nodes with 120GB of input data, FP-Hadoop can outperform Hadoop by a factor of about 10 in reduce time, and a factor of 5 in total execution time. The results show that the higher is the number of nodes, the higher can be the gain of FP-Hadoop. They also show that the bigger is the size of the input data, the higher can be the improvement gain of FP-Hadoop.

Aknowledgements

Experiments presented in this paper were carried out using the Grid'5000 experimental testbed, being developed under the INRIA ALADDIN development action with support from CNRS, RENATER and several universities as well as other funding bodies (see https://www.grid5000.fr).

Figure

  Figure 1: FP-Hadoop job processing scheme. The input of the intermediate reduce phase is a set of intermediate splits (blocks) which are generated dynamically using the intermediate data distributed over map workers. No intermediate reduce worker is idle until there is any intermediate split. The intermediate reduce phase can be done in several iterations.

  Count), FR = (Sum, Count) THEN Sum/Count function both in the intermediate and final reduce phases, i.e., we have IR = FR = R. Examples of such functions are Top-k, SkyLine, Union and Sum.

2 3heartBeat: IRF j info 4 Figure 3 :

 243 Figure 3: Data flow between a map worker, the master and an intermediate reduce worker. The communicated messages are shown in their sent order.

Figure 4 :

 4 Figure 4: Input size of different phases in multiple iterations example

Figure 6 :

 6 Figure 6: Worker failure example.

Figure 9 :

 9 Figure 9: Analysis of FP-Hadoop parameters: (a) MinIRSize and iterations, (b) number of performed iterations (c) number of IR tasks.

  ): map, intermediate reduce, and final reduce. The map phase is almost the same as that of Hadoop in

	Input Splits	Map workers	Intermediate key-values	Intermediate reduce workers	Intermediate key-values
	D1	D2	M1	k1	k2	k3	R1
				V1,1	V2,1	V3,1
	D3	D4	...	V1,2	V2,2		...
	...		Mm		V2,3		Rr
	Dn-1 Dn			V2,4
					V2,5

  1: FP-Hadoop job processing scheme. The input of the intermediate reduce phase is a set of intermediate splits (blocks) which are generated dynamically using the intermediate data distributed over map workers. No intermediate reduce worker is idle until there is any intermediate split. The intermediate reduce phase can be done in several iterations.

  Let F be a reduce function, and S a set of values, and S 1 • ... • S n a partitioning over S. F is Holistic if there is not a mvalued function (with bounded m) that characterizes the computation. In other words, F is holistic if it is neither distributive nor algebraic.Intuitively, a function is distributive if it can be computed in a distributed manner. Examples of such functions are Count, Sum, Min, and Max. Distributive functions can particularly benefit from the usage of intermediate reduce tasks, as they usually reduce the data size significantly. Algebraic functions are the functions that can be computed by using a bounded number of distributive functions. Examples of algebraic functions are AVG and Std. These functions can also take advantage of intermediate reduce tasks. In holistic functions, there is no constant bound on the storage size needed to describe a sub-aggregate. Some holistic functions, such as SkyLine, may take great advantage of running an intermediate phase. For some other holistic functions, it may be difficult to find an efficient intermediate function.

• S n a partitioning over S. F is Algebraic if there is a m-valued function G and a function H such that F (S) = H( G(S 1 ) 1 , ..., G(S n ) 1 , ..., G(S 1 ) m , ..., G(S n ) m ).

• Definition 3 (Holistic).

  IR task info: ..., IRF i info, ...

							Master
				IR Fragments Table (IRF Table)
					IRF i info
				p0	map 1 spill 1	map 2 spill 1	...
					size	size
							Reduce
				...			Scheduler
				pr	map 1 spill 1	...
					size
				1
		heartBeat: IRF i info, ...
	Map Worker					Intermediate Reduce Worker
	Input	Map task map 1					IR task ir1
	Split	Buffer					IRF0
							IR fragments
		IRF i	spill 1
	map function	p0 p0	...	pr			...	IR Split	IR function	IRF j
			...
			spill m				...
		p0	...	pr
							IRFn

http://dumps.wikimedia.org/other/pagecounts-raw/

This program is just for illustration; actually, it is possible to write a more efficient code by leveraging on the sorting mechanisms of MapReduce.

This mechanism is used for communication between the master and workers

http://www.grid5000.fr

in the cases where there is no skew in the intermediate data or the skew can be significantly decreased in the map phase by using a combiner function, e.g. in the word count job which we tested. In these cases, it is sufficient to not declare an IR function, then FP-Hadoop executes the job without performing the IR phase.

Related Work

In the literature, there have been many efforts to improve MapReduce [START_REF] Lee | Parallel data processing with mapreduce: a survey[END_REF], particularly by supporting high level languages on top of it, e.g., Pig [START_REF] Olston | Pig latin: a not-so-foreign language for data processing[END_REF], optimizing I/O cost, e.g., by using column-oriented techniques for data storage [START_REF] Floratou | Column oriented storage techniques for mapreduce[END_REF][START_REF] Lin | Llama: leveraging columnar storage for scalable join processing in the mapreduce framework[END_REF][START_REF] Chang | Bigtable: A distributed storage system for structured data[END_REF], supporting loops [START_REF] Bu | The haloop approach to large-scale iterative data analysis[END_REF], adding index [START_REF] Dittrich | Hadoop++: Making a yellow elephant run like a cheetah (without it even noticing)[END_REF], caching intermediate data [START_REF] Elghandour | Restore: reusing results of mapreduce jobs in pig[END_REF], supporting special operations such as join and Skyline [START_REF] Afrati | Fuzzy joins using mapreduce[END_REF][START_REF] Huang | MELODY-JOIN: efficient earth mover's distance similarity joins using mapreduce[END_REF][START_REF] Yang | Mapreduce-merge: simplified relational data processing on large clusters[END_REF][START_REF] Jiang | Map-join-reduce: Toward scalable and efficient data analysis on large clusters[END_REF][START_REF] Silva | Exploiting mapreduce-based similarity joins[END_REF][START_REF] Okcan | Processing theta-joins using mapreduce[END_REF][START_REF] Deng | Massjoin: A mapreduce-based method for scalable string similarity joins[END_REF][START_REF] Fries | Phidj: Parallel similarity self-join for high-dimensional vector data with mapreduce[END_REF], balancing data skew [START_REF] Gufler | Load balancing in MapReduce based on scalable cardinality estimates[END_REF][START_REF] Ramakrishnan | Balancing reducer skew in mapreduce workloads using progressive sampling[END_REF], etc. Hereafter, we briefly present some of them that are the most related to our work.

The approach proposed in [START_REF] Ramakrishnan | Balancing reducer skew in mapreduce workloads using progressive sampling[END_REF] tries to balance data skew in reduce tasks by subdividing keys with large value sets. It requires some user interaction or user knowledge of statistics or sampling, in order to estimate in advance the values size of each key, and then subdivide the keys with large values. Gufler et al. [START_REF] Gufler | Load balancing in MapReduce based on scalable cardinality estimates[END_REF] propose an adaptive approach which collects statistics about intermediate key frequencies and assigns them to the reduce tasks dynam-ically at scheduling time. In a similar approach, Sailfish [START_REF] Rao | Sailfish: a framework for large scale data processing[END_REF] collects some information about the intermediate keys, and uses them for optimizing the number of reduce tasks and partitioning the keys to reducer workers. However, these approaches are not efficient when all or a big part of the intermediate values belong to only one key or a few number of keys (i.e., less than the number of reduce workers).

SkewTune [START_REF] Kwon | Skewtune: mitigating skew in mapreduce applications[END_REF] adopts an on-the-fly approach that detects straggling reduce tasks and dynamically repartitions their input keys among the reduce workers that have completed their work. This approach can be efficient in the cases where the slow progress of a reduce task is due to inappropriate initial partitioning of the key-values to reduce tasks. But, it does not allow the collaboration of reduce workers on the same key.

Haloop [START_REF] Bu | The haloop approach to large-scale iterative data analysis[END_REF] extends MapReduce to serve applications that need iterative programs. Although iterative programs in MapReduce can be done by executing a sequence of MapReduce jobs, they may suffer from big data transfer between reduce and map workers of successive iterations. Haloop offers a programming interface to express iterative programs and implements a task scheduling that enables data reuse across iterations. However, it does not allow hierarchical execution plans for reducing the intermediate values of one key, as in our intermediate reduce phase. Memory Map Reduce (M3R) [START_REF] Shinnar | M3r: Increased performance for in-memory hadoop jobs[END_REF] is an implementation of


Paul Feliot

Yves Le Guennec

Julien Bect

Emmanuel Vazquez

Bayesian optimization techniques

Keywords: Bayesian optimization, Multi-objective, Hidden constraints, Expected improvement, Sequential Monte Carlo

ou non, émanant des établissements d'enseignement et de recherche français ou étrangers, des laboratoires publics ou privés.

Introduction

The purpose of the environment control system (ECS) of a commercial aircraft is to provide a certain level of comfort to the passengers by regulating the temperature and pressure of the air injected into the cabin. The system is based on an inverse Brayton thermodynamic cycle. Hot and pressurized air is taken from the engines at the compressor stage through the bleed and ram air from the outside of the aircraft is used as coolant. For safety reasons, the hot air from the engines passes through a first heat exchanger where it is cooled down below the critical fuel ignition temperature. Then it is pressurized through a compressor and cooled again using a second heat exchanger. It then passes through a turbine where work is extracted to propel the compressor. The cooled and expanded air exiting the turbine is eventually mixed with hot air from the first heat exchanger outflow to reach the desired temperature and pressure before injection into the cabin.

The design of an optimal ECS is a complex problem in practice. It has been addressed in previous studies under different optimality conditions and modelling assumptions (see, e.g., [START_REF] Vargas | Thermodynamic optimization of finned crossflow heat exchangers for aircraft environmental control systems[END_REF][START_REF] Bejan | The need for exergy analysis and thermodynamic optimization in aircraft development. Exergy[END_REF][START_REF] Pérez-Grande | Optimization of a commercial aircraft environmental control system[END_REF]). In their article, Pérez-Grande and Leo [START_REF] Pérez-Grande | Optimization of a commercial aircraft environmental control system[END_REF] study an aircraft-on-cruse scenario and propose a one dimensional model of the two heat exchangers. The system is designed in order to achieve minimal mass and entropy generation, two objectives that are shown to be antagonistic and which both affect the overall performance of the aircraft.

In this article, we extend their work by considering also the sizing of the rotating machines and by considering an aircraft-on-ground scenario, which corresponds to the most critical situation for the ECS in terms of cold production, and is therefore dimensioning. The design optimization of the system is performed using the BMOO algorithm [START_REF] Feliot | A Bayesian approach to constrained single-and multiobjective optimization[END_REF], which implements a Bayesian approach to the multi-objective optimization problem in the presence of non-linear constraints. The problem consists in finding an approximation of the set Γ = {x ∈ X : c(x) ≤ 0 and ∄ x ′ ∈ X such that c(x ′ ) ≤ 0 and f (x ′ ) ≺ f (x)} where X ⊂ R d is the search domain, c = (c i ) 1≤i≤q is a vector of constraint functions (c i : X → R), c(x) ≤ 0 means that c i (x) ≤ 0 for all 1 ≤ i ≤ q, f = (f j ) 1≤j≤p is a vector of objective functions to be minimized (f j : X → R), and ≺ denotes the Pareto domination rule (see, e.g., [START_REF] Fonseca | Multiobjective optimization and multiple constraint handling with evolutionary algorithms. I. A unified formulation[END_REF]).

The article is organized as follows. First we detail in Section 2 the model that is used to estimate the performances and main characteristics of the system. A one-dimensional analysis is performed to establish the 2 Thermodynamic analysis of the ECS

Sizing scenario

The architecture of the ECS is represented on Figure 1. Bleed air from the engines arrives into the system at location 1. Ram air from the outside of the aircraft is levied at location 0r and is used as coolant. The hot air enters a first heat exchanger where it is cooled down below the fuel ignition temperature. A by-pass at location 2 then permits to regulate the system by controlling the air flowrate entering the air cycle machine (ACM). Cooled and expanded air exits the ACM at location 5 and is mixed with warm air from the by-pass to reach the desired pressure and temperature before injection into the cabin at location 6.

In practice, the system must be able to satisfy strict specifications under different environmental conditions and operating situations. In this work, we consider a scenario where the aircraft is on ground, full of passengers, equipments running, and with an outside temperature of 50 • C. In that situation, the ECS must be able to maintain the cabin temperature at T c = 24 • C. This scenario corresponds to the most demanding specification in terms of cold production, and is therefore dimensioning for the system. Formally, this means that the ECS must be able to dissipate enough heat to compensate for the thermal power P HT produced by the passengers, the crew, the equipments and the environment:

P HT = P out + P eq + N pax P pax + N crew P crew ,
where P out is the outside flow dissipation, P eq is the thermal power produced by the equipments, P pax and P crew are the thermal powers produced by a passenger and by a crewmember and N pax and N crew are the number of passengers and the number of crewmembers in the aircraft.

In this scenario, the by-pass is wide open so that all the air from location 2 goes to the ACM. Also, there is no relative velocity between the aircraft and the ambient air when it is grounded and therefore there is no natural coolant flowrate. In this work, we consider a system where the ram flowrate is created by an auxiliary fan placed at the ram air entrance and powered by the turbine of the ACM. The sizing of this auxiliary fan is not taken into account in this study but we will assume that the ram flowrate can be controlled.

Heat exchangers

We now detail the model that is used to emulate the system. For the heat exchangers HX1 and HX2, we use a model from [START_REF] Pérez-Grande | Optimization of a commercial aircraft environmental control system[END_REF]. The two heat exchangers are compact cross-flow heat exchangers with unmixed fluids. For this kind of heat exchangers, the energy exchanged per unit time between the ram and bleed air can be formulated as:

ܷ ସ ܹ ସ ܥ ସ ܥ ଷ ܷ ଷ ܹ ଷ ߚ ଷ ߙ ସ ߱ ߱
ṁc p (T t1 -T t2 ) = ṁr c p (T t3r -T t2r ), (1)
ṁc p (T t3 -T t4 ) = ṁr c p (T t2r -T t1r ), (2)
where ṁ and ṁr denote respectively the bleed and ram air flowrates, c p is the thermal capacity of the air and is assumed constant, and T ti represents the stagnation temperature at location i ∈ {1, 2, 3, 4, 5, 1r, 2r, 3r}. Besides, we can define the efficiencies ǫ 1 and ǫ 2 of the two heat exchangers HX1 and HX2 as the ratio between the energy effectively exchanged and the total energy exchangeable:

c p (T t1 -T t2 ) = ǫ 1 c p (T t1 -T t2r ), (3)
c p (T t3 -T t4 ) = ǫ 2 c p (T t3 -T t1r ). (4)
Note that Eq.(3-4) only hold when ṁ ≤ ṁr . Otherwise, T t1 -T t2r should be replace by T t3r -T t2r in Eq.( 3) and T t3 -T t1r should be replaced by T t2r -T t1r in Eq.( 4). The efficiencies ǫ 1 and ǫ 2 of the two heat exchangers depend on their geometry and we use the ǫ-Ntu model detailed in [START_REF] Pérez-Grande | Optimization of a commercial aircraft environmental control system[END_REF] to estimate them.

In this study, the pressure drops as the air passes through the heat exchangers are considered constant:

P t2 -P t1 = ∆P HX , (5)
P t4 -P t3 = ∆P HX , (6)
where P ti represents the stagnation pressure at location i and ∆P HX is a constant pressure loss. While this permits to simplify the model, it is inaccurate because the pressure losses do also depend on the geometries of the heat exchangers and are responsible for a non-negligible proportion of the entropy generated by the system. In particular, the frictions are expected to rise as the volume of the heat exchangers decreases, which further increases the necessary balance between entropy generation and mass. Again, the reader is referred to [START_REF] Pérez-Grande | Optimization of a commercial aircraft environmental control system[END_REF] for a discussion on a possible model of the pressure drops.

Compressor and turbine

We consider a centrifugal compressor and an axial flow turbine. It is assumed that the air enters the compressor axially and exits parallel to the blades (the slip factor is neglected), with an angle β 3 as illustrated on Figure 2. Similarly, it is assumed that the air enters the turbine with an angle α 4 corresponding to the stator blades angle and goes out axially. The rotational speed of the shaft linking the compressor with the turbine is denoted ω. The letters C, U and W on Figure 2 denote respectively the air absolute velocity vector, the blade tip speed vector and the relative velocity vector, such that C = U + W . In the following, the subscripts u, m, and x will stand respectively for the tangential, meridional and axial components of the velocity vectors. The power exchanged per unit time between the machines and the fluid and the change of momentum of the fluid are related by Euler's theorem as:

ẆC = ṁ(U 3 C 3u -U 2 C 2u ), ẆT = ṁ(U 5 C 5u -U 4 C 4u ),
where ẆC and ẆT denote respectively the power received by the fluid from the compressor and from the turbine. Note that with this convention ẆT < 0 and ẆC > 0. The turbine is converting part of the fluid energy into rotation speed, and the compressor is augmenting the energy of the fluid through its mechanical work, which increases the temperature of the fluid.

Under the assumption that the flow enters the compressor and exits the turbine axially, the tangential components of the air velocity vectors are neglected and C 2u = C 5u = 0. The Euler theorem simplifies to:

ẆC = ṁU 3 C 3u , ẆT = -ṁU 4 C 4u ,
The conservation of the mass at locations 3 and 4 gives the following additional equations, relating the air flowrate to its velocity and a control surface.

ṁ = 2πρr 3 b 3 W 3m , ṁ = 2πρr 4 b 4 C 4m ,
where ρ is the air density (which is assumed constant), r 3 and r 4 are respectively the compressor outlet blade radius and turbine inlet blade radius, and b 3 and b 4 are respectively the compressor and turbine tip blade heights (see Figure 3). Using the velocity triangles of Figure 2, the Euler theorem can then be rewritten using the rotating machines geometries and rotational speed:

ẆC = ṁ r 2 3 ω 2 - ṁ tan(β 3 ) 2πρb 3 ω , (7)
ẆT = - ṁ2 tan(α 4 ) 2πρb 4 ω. (8)
Besides, the work extracted from the turbine is used to propel the compressor and the auxiliary fan. Writing down the conservation of energy per unit time we get:

ẆC + ẆT + 1 η F ṁ3 r 2ρ 2 A 2 r = 0, (9)
where A r is a control surface at the ram air entrance and η F is the ratio between the kinetic energy per unit time produced by the auxiliary fan and the power furnished by the turbine to the fan.

The powers ẆC and ẆT can also be expressed as functions of the stagnation temperatures by considering the change in total enthalpy of the fluid passing through the rotating machines (the other contributions are neglected):

ẆC = η C ṁc p (T t3 -T t2 ), (10)
ẆT = 1 η T ṁc p (T t5 -T t4 ), (11)
where η C and η T are respectively the compressor and turbine isentropic efficiencies. Finally, the stagnation pressure ratios are given by the isentropic relations:

P t3 P t2 = 1 + η C T t3 -T t2 T t2 γ γ-1 , (12)
P t5 P t4 = 1 + 1 η T T t5 -T t4 T t4 γ γ-1 , ( 13
)
where γ is the air isentropic coefficient.

Mass and entropy generation rate of the system

The mass of the system can be approximated by considering estimates of the volumes of its components and representative densities. For the two heat exchangers, we consider rectangular volumes and a representative density ρ HX (see Table 3). where L x1 , L y1 , L z1 , L x2 , L y2 , L z2 are the heat exchangers dimensions (see Table 1), and M HX1 and M HX2 are the mass of the heat exchangers. For the compressor and turbine of the ACM, we consider separately the volumes of the blades and the volume of the machine body (see Figure 3):

M HX1 = ρ HX • L x1 L y1 L z1 , M HX2 = ρ HX • L x2 L y2 L z2 , ݎ ଷ ݄ ൈ ݎ ଷ ܾ ଷ ݎ ଶ ݎ ଶ௧ ݎ ସ ݄ ௧ ൈ ݎ ସ ܾ ସ ݎ ହ ݎ ହ௧
V C,blade = e c h c r 3 (r 3 -r 2p ) 2 - (r 3 -r 2t )(h c r 3 -b 3 ) 2 , V C,body = πr 2 3 h c (r 3 + r 2p ) 3 - πh c r 3 2p 3 ,
for the compressor, and

V T,blade = e t h t r 4 (r 4 -r 5p ) 2 - (r 4 -r 5t )(h t r 4 -b 4 ) 2 , V T,body = πr 2 4 h t (r 4 + r 5p ) 3 - πh t r 3 5p 3 ,
for the turbine, where e c and e t are the compressor and turbine blades thickness, and h c and h t are aspect ratios. The mass of the system is then given by the following.

M = M HX1 + M HX2 + ρ steel (Z C V C,blade + V C,body ) + ρ steel (Z T V T,blade + V T,body ), (14)
where Z C and Z T are respectively the number of blades of the compressor and of the turbine. The entropy generation rate of the ECS is the sum of the contributions along the bleed stream and along the ram stream, from entrance to exit:

Ṡ = ṁ c p log T 5 T a -R log P 5 P a + ṁr c p log T 3r T a -R log P 3r P a , ( 15
)
where R is the perfect gas constant, T a and P a are the ambient temperature and pressure, and T i and P i are respectively the static temperature and the static pressure at location i ∈ {5, 3r}. The equations giving the static properties for the bleed stream are gathered in Table 3 in the additional material. For the ram stream, it is assumed that the Mach number remains low through the heat exchangers. Thus, T 3r = T t3r , T 2r = T t2r , T 1r = T t1r . For the static pressures P 2r and P 3r , the law of perfect gas is used.

3 Optimization of the system

Formulation of the optimization problem

We consider an optimization problem using the 18 design variables given in Table 1. All other design parameters and physical properties are fixed (see Table 4 in appendix). Under the model developed in Section 2, the ECS is thus ruled by a system of 13 equations (Eq. ( 1)-( 13)) with 13 unknowns, which are the stagnation temperatures and pressures, the powers exchanged between the fluid and the compressor and turbine, and the rotational speed of the rotating ensemble: T t2 , T t3 , T t4 , T t5 , T t2r , T t3r , P t2 , P t3 , P t4 , P t5 , ẆC , ẆT and ω. Eq.( 14) and Eq.( 15) give respectively the mass and the entropy generation rate, which are the objectives of the optimization. Additionally, we formulate the following 15 inequality constraints (see Table 3): : P HT ≤ ṁc p (T c -T 5 ) .

c 1-2 : T min ≤ T 5 ≤ T max , c 3
The constraints c 1 to c 4 are standard specifications. The air injected into the cabin must lie between T min = 15 • C and T max = 25 • C and at a pressure close to the atmospheric pressure. Thus we take P min = 101.3 kPa and P max = 1.05P min . The constraints c 5 to c 8 are on the heat exchangers efficiencies. The design should be efficient enough but not too expensive to manufacture. The constraints c 9 to c 12 are on the air velocity. In the model, we have assumed that the air density remains constant throughout the bleed stream, which is inaccurate if the flow becomes supersonic. We take a 5% margin to account for the possible variations of uncertain parameters and avoid numerical instabilities. Similarly, it is required via constraints c 13 and c 14 that the compressor and turbine blade tip speeds be subsonic. Constraint c 15 stems from the sizing scenario considered in this study: The dissipated power must be greater than the power produced by the passengers, the crew, the equipments and the environment (see Section 2.1). Note that an equality constraint is not necessary because the constraint is expected to be active at the optima.

To ensure the feasibility of the system and avoid numerical issues, we enforce the following restrictions on the design variables (see Figure 3):

d 1 : ṁ ≤ ṁr , d 2 : b 3 ≤ h c r 3 , d 3 : b 4 ≤ h t r 4 , d 4-5 : r 2p + 0.02 ≤ r 2t ≤ r 3 , d 6-7 : r 5p + 0.02 ≤ r 5t ≤ r 4 , d 8 : ∆ ≥ 0, d 9 : tan(β3) b3 ≥ -tan(α4) b4 .
where ∆ in d 8 is the discriminant of Eq ( 7)-( 9), seen as a second order polynomial equation in ω. The conditions d 8 and d 9 are necessary to ensure that there exists a real solution ω > 0 to Eq [START_REF] Rasmussen | Gaussian processes for machine learning[END_REF]. When two such solutions are possible, we take the largest one. Note that this parametrization implies that the optimization needs to be performed on a non-hypercubic design domain.

Optimization algorithm

The optimization is performed using the BMOO algorithm [START_REF] Feliot | A Bayesian approach to constrained single-and multiobjective optimization[END_REF]. This algorithm implements a Bayesian approach to the multi-objective optimization problem in the presence of non-linear constraints. The objectives and constraints of the problem are modeled using Gaussian process emulators (see, e.g., [START_REF] Rasmussen | Gaussian processes for machine learning[END_REF]) and the algorithm performs a sequential optimization procedure where the next sample is chosen as the maximizer of an extended version of the expected improvement sampling criterion (see, e.g., [START_REF] Jones | Efficient global optimization of expensive black-box functions[END_REF]). In practice, this requires to solve an auxiliary optimization problem at each iteration. The BMOO algorithm uses sequential Monte Carlo techniques to conduct this auxiliary optimization (see, e.g., [START_REF] Del Moral | Sequential Monte Carlo samplers[END_REF]). A population of candidate designs distributed according to a density of interest in the design space is sampled at each iteration and the maximizer of the extended expected improvement is chosen out of this population.

We take advantage of this to handle non-hypercubic design domains by truncating the density of interest so as to propose only candidates that lie in the desired region. This is straightforward because sequential Monte Compressor outlet angle (rad) Carlo methods do not require that the normalizing constant of the target density be known. For initialization, a pseudo maximin design of experiments on the non-hypercubic design domain (see e.g. [START_REF] Auffray | Maximin design on non hypercube domains and kernel interpolation[END_REF]) can be achieved using rejection sampling. A large population of particles is sampled uniformly on the containing hypercube defined using the values of Table 1. The particles which do not respect the constraints d 1 to d 9 are then discarded and the population of surviving particles is pruned until the desired population size is reached. During the pruning step, particles that are too close to other particles are discarded, thus raising the maximin distance. Note that in practice, this requires that the volume of the design domain be not too small compared with the volume of the containing hypercube (the ratio of volumes was estimated close to 6% for this particular application).

β 3 [-π 3 , π 3 ] Turbine inlet angle (rad) α 4 [0, π 3
Because the computation of the objectives and constraints values for a given design requires to solve the nonlinear system formed by Eq.(1) to Eq.( 13), it may happen that no solution can be found, in which case it is not possible to provide values of the constraints and objectives for the design under study. Also, some designs can lead to supersonic solutions for which the values of temperatures and pressures predicted by the model can be inaccurate. When this happens, we prefer to consider such designs as simulation failures and not use the values returned by the model. In the optimization procedure, this is taken into account in order to prevent the optimizer to explore regions where simulation failures are likely, by multiplying both the sampling criterion of BMOO and the density in the sequential Monte Carlo procedure by a probability of observability. This technique has been proposed by Lee and co-authors [START_REF] Lee | Optimization subject to hidden constraints via statistical emulation[END_REF]. A statistical model is learned on the observed/non-observed data and provides a probability of satisfying the hidden constraints leading to simulation failures. In this work, a nearest-neighbours classifier using 5 neighbours and the L 2 distance is used to that purpose.

Optimization results

The algorithm is run with a limiting budget of N max = 500 calls to the simulation model, and an initial design of N init = 90 samples. The set of optimal trade-off solutions found by the algorithm is shown on Figure 4.

Among the initial design of experiments, 44 experiments led to simulation failures and 92 additional failures occurred during the optimization process. Further investigation revealed that most of the simulation failures occurred because the flow was supersonic in the compressor, which happens with high probability when the bleed flowrate is high and the compressor radii are low. Regarding the constraints satisfaction, no feasible observations were made in the initial sample and the algorithm found one after 25 iterations.

The design parameters associated to 7 trade-off solutions chosen along the Pareto front are given in Table 2. Several observations can be made on these results. First we note that the bleed flowrate remains constant along the front. This is because c 15 is active (see Section 3.1) and T 5 = T min for optimal designs, which forces the value of the bleed flowrate. The ram flowrate is less constrained and varies along the front. We note its strong influence on the entropy generation rate (see Eq.( 15)). The variation of the mass on the other hand mostly comes from the variations of L z1 and L z2 . As the heat exchangers height is raised, the entropy generation rate is lowered but the mass augments. The values of L x1 , L y1 , L x2 , and L y2 are set close to their maximal values, which

Conclusions

In this article, a one dimensional model of the environment control system of a commercial aircraft is proposed. The model permits to emulate the behaviour of the system when the geometries of its components vary, for a scenario where the aircraft is on ground, full of passengers, equipments running, and with an outside temperature of 50 • C. The system is optimized using the BMOO algorithm, which implements a Bayesian approach to the multi-objective optimization problem in the presence of non-linear constraints, and trade-off design solutions in terms of mass and entropy generation rate of the system are identified.

As a particularity, the optimization is performed on a non-hypercubic design domain and involves hidden constraints. This is a situation that is often encountered in engineering design optimization. The BMOO algorithm is successfully adapted to this new setup, which makes it possible to conduct a multi-objective optimization using a reasonable number of calls to the numerical simulation model.

The BMOO algorithm is primarily designed to address problems where the computational time associated to the model evaluation is high, which is not the case here. In this study, most of the computational time required to conduct the optimization was taken by the optimizer and more work is needed to make the implementation of the algorithm more efficient. Nevertheless, the algorithm achieves very satisfactory results and is a competitive algorithm to address multi-objective optimization problems with several constraints.

A Additional material

Figure 1 :

 1 Figure 1: Architecture of the environment control system of a commercial aircraft

Figure 2 :

 2 Figure 2: Compressor (left) and turbine (right) velocity triangles.

Figure 3 :

 3 Figure 3: Geometrical parametrization of the compressor (left) and of the turbine (right).

  ] Heat exchanger 1: x length (m) L x1 [0.025, 0.7] Heat exchanger 2: x length (m) L x2 [0.025, 0.7] Heat exchanger 1: y length (m) L y1 [0.025, 0.7] Heat exchanger 2: y length (m) L y2 [0.025, 0.7] Heat exchanger 1: z length (m) L z1 [0.025, 0.7] Heat exchanger 2: z length (m) L z2 [0.025, 0.7]

T 5 =

 5 T t5 -

Table 1 :

 1 Design variables description

	Description	Not.	Domain	Description	Not.	Domain
	Bleed flowrate (kg.s -1 )	ṁ	[2, 8]	Ram flowrate (kg.s -1 )	ṁr	[2, 8]
	Compressor outlet radius (m)	r 3	[0.1, 0.3]	Turbine inlet radius (m)	r 4	[0.1, 0.3]
	Compressor inlet foot radius (m)	r 2p	[0.03, 0.1] Turbine outlet foot radius (m)	r 5p	[0.03, 0.1]
	Compressor inlet tip radius (m)	r 2t	[0.04, 0.2] Turbine outlet tip radius (m)	r 5t	[0.04, 0.2]
	Compressor outlet blade height (m)	b 3	[0.01, 0.1] Turbine inlet blade height (m)	b 4	[0.01, 0.1]

Table 3 :

 3 Static properties equations

	Static temperatures	Static pressures	Fluid velocities

Table 4 :

 4 Parameters values used in the experiments of Section 3

	Description	Not.	Value	Description	Not.	Value
	Simulation parameters
	Ambient temperature (K)	T a	323	Ambient pressure (Pa)	P a	101.3e3
	Number of passengers	N pax	120	Number of crewmembers	N crew	5
	Thermal power passengers (W) Thermal power equipments (W) Bleed temperature (K)	P pax P eq T 1	70 4800 473	Thermal power crew (W) Outside flow dissipation (W) Bleed pressure (Pa)	P crew P out P 1	100 3000 260e3
	Pressure losses (Pa)	∆P HX	40e3	Valve opening	θ	0
	Ram stream cross surface (m 2 )	A r	0.20	Fan efficiency	η F	0.95
	Air specific heat (J.kg -1 .K -1 )	c p	1004	Air isentropic coefficient	γ	1.4
	Perfect gaz constant (J.kg -1 .K -1 )	R	287
	Heat exchangers
	Viscosity bleed (kg.m -1 .s -1 )	µ	2.28e-5 Viscosity ram (kg.m -1 .s -1 )	µ r	2.28e-5
	H.T. ratio bleed stream (m -1 )	β	2231	H.T. ratio ram stream (m -1 )	β r	1115
	Plate spacing bleed stream (m)	b	5.21e-3 Plate spacing ram stream (m)	b r	12.3e-3
	Prandtl number bleed stream	P r	0.7	Prandtl number ram stream	P r r	0.7
	Hydraulic diameter bleed (m)	Dh	1.54e-3 Hydraulic diameter ram (m)	Dh r	3.41e-3
	Convection length bleed (m)	λ	0.035	Convection length ram (m)	λ r	0.035
	Representative density (kg.m -3 )	ρ HX	1415	Fin thickness (m)	δ	0.102e-3
	Wall thickness (m)	t w	6e-4	Thermal conductivity (W.m -1 .K -1 )	k w	237
	Compressor			Turbine
	Adiabatic efficiency	η c	0.8	Adiabatic efficiency	η t	0.92
	Aspect ratio	h c	0.7	Aspect ratio	h t	0.5
	Blades thickness (m)	e c	0.01	Blades thickness (m)	e t	0.01
	Number of blades	Z c	21	Number of blades	Z t	21

Acknowledgements: This research work has been carried out within the Technological Research Institute SystemX, using public funds from the French Programme Investissements d'Avenir. We also thank Airbus Group Innovations for their contributions to this work.

permits to achieve efficiencies between 0.7 and 0.8. Note that the pressure losses are assumed constant in this study. Further work is required to better understand their impact on the entropy generation rate when the heat exchangers dimensions become small. Regarding the turbine and compressor dimensions, they are set as small as possible, which keeps the mass low and augments the fluid velocity, thus achieving good performances.