Youssef S G Nashed
email: nashed@ce.unipr.it

Roberto Ugolotti

Pablo Mesejo
email: pmesejo@ce.unipr.it

Stefano Cagnoni
email: cagnoni@ce.unipr.it

libCudaOptimize: an Open Source Library of GPU-based Metaheuristics

Keywords: I.2.5 [Artificial Intelligence]: Programming Languages and Software; D.2.13 [Reusable Software]: Reusable libraries Algorithms, Design Open Source Library, GPGPU, CUDA, Particle Swarm Optimization, Differential Evolution, Scatter Search, Solis and Wets local search

Evolutionary Computation techniques and other metaheuristics have been increasingly used in the last years for solving many real-world tasks that can be formulated as optimization problems. Among their numerous strengths, a major one is their natural predisposition to parallelization.

In this paper, we introduce libCudaOptimize, an open source library which implements some metaheuristics for continuous optimization: presently Particle Swarm Optimization, Differential Evolution, Scatter Search, and So-lis&Wets local search. This library allows users either to apply these metaheuristics directly to their own fitness function or to extend it by implementing their own parallel optimization techniques. The library is written in CUDA-C to make extensive use of parallelization, as allowed by Graphics Processing Units.

After describing the library, we consider two practical case studies: the optimization of a fitness function for the automatic localization of anatomical brain structures in histological images, and the parallel implementation of Simulated Annealing as a new module, which extends the library while keeping code compatibility with it, so that the new method can be readily available for future use within the library as an alternative optimization technique.

INTRODUCTION

In the last decades, several metaheuristics have been developed (among others, Particle Swarm Optimization (PSO) [START_REF] Kennedy | Particle Swarm Optimization[END_REF], Differential Evolution (DE) [START_REF] Storn | Differential Evolution-A Simple and Efficient Adaptive Scheme for Global Optimization over Continuous Spaces[END_REF], Ant Colony Optimization (ACO) [START_REF] Dorigo | Ant Colony Optimization[END_REF], Scatter Search (SS) [START_REF] Glover | Scatter search[END_REF], . . . ) and applied to a large variety of problems and fields. To facilitate the use of Evolutionary Computation (EC) methods in optimization problems, several software environments or libraries have been developed, like HeuristicLab [START_REF] Wagner | Heuristic Optimization Software Systems -Modeling of Heuristic Optimization Algorithms in the HeuristicLab Software Environment[END_REF], Matlab Optimization Toolbox [START_REF]Matlab Optimization Toolbox User's Guide[END_REF], CILib [START_REF] Pampara | CIlib: A collaborative framework for computational intelligence algorithms -part I[END_REF], jMetal [START_REF] Durillo | jMetal: A Java Framework for Developing Multi-Objective Optimization Metaheuristics[END_REF] or JCLEC [START_REF] Ventura | JCLEC: A Java Framework for Evolutionary Computing[END_REF]. Recently, these metaheuristics have also been developed on Graphics Processing Units (GPU) [START_REF] Krömer | A comparison of many-threaded differential evolution and genetic algorithms on cuda[END_REF][START_REF] Mussi | Evaluation of parallel particle swarm optimization algorithms within the CUDA TM architecture[END_REF], fully exploiting their intrinsic parallelism and obtaining significant speedups (up to 30 times) compared to single thread CPU implementations. However, no open-source software has been released to easily take advantage of this aspect.

The main idea behind our work is to offer a user the chance to apply metaheuristics as simply and fast as possible to his own problem of interest, exploiting the parallelization opportunities offered by modern GPUs as much as possible. To the best of our knowledge, there are no software tools in which the entire optimization process, from exploration operators to function evaluation, is completely developed on the GPU, and allows one to develop both local and global optimization methods. Only in the last years, some packages, like ParadisEO [START_REF] Melab | Towards paradiseo-mo-gpu: A framework for gpu-based local search metaheuristics[END_REF], have started to use GPUs to speed up their algorithms, however, as it stands, this is limited to parallel data access during fitness evaluation, while the method itself is still executed sequentially. Also, the GPU implementations so far in ParadisEO are strictly limited to local optimization/search methods.

We present libCudaOptimize, a GPU-based open source library that allows users to run their methods in parallel to optimize a fitness function, introduce a new optimization algorithm, or easily modify/extend existing ones. In the first case, the only thing one needs to do is to write the new fitness function in C++ or CUDA-C, while in the second and third cases, one can take advantage of the framework offered by the library to avoid the need to go deep into basic implementation issues, especially regarding parallel code.

libCudaOptimize is expected to be used by users who have, at least, a basic knowledge of C++. Although no explicit understanding of CUDA-C or even of metaheuristics is required it is very useful anyway, nonetheless, one can use this library just by writing a C++ fitness function and launching one of the optimization techniques already implemented (to date PSO, DE, SS and Solis&Wets local search (SW) [START_REF] Solis | Minimization by Random Search Techniques[END_REF]). This allows one to:

• implement commonly successful techniques with limited efforts;

• easily compare the results obtained by running different techniques on different functions;

• analyze the effects of changing values of the parameters which regulate the behavior of the optimization techniques on user-defined problems;

• run high-dimensional optimization experiments on consumer level hardware, thanks to the efficient CUDA-C parallel implementation.

The remainder of this work is organized as follows: in section 2 libCudaOptimize is described, then the operations needed to start working with the library are presented in section 3, followed by two case studies in section 4 and by conclusions in section 5.

THE PACKAGE

Implemented Methods

In the present version, the library implements three different global optimization methods (PSO, DE and SS) and one local search technique (SW), which was added to demonstrate that the possibilities of this tool may go beyond the implementation of population-based metaheuristics.

Particle Swarm Optimization

Particle Swarm Optimization [START_REF] Kennedy | Particle Swarm Optimization[END_REF] is a bio-inspired optimization algorithm based on the simulation of the social behavior of bird flocks. In the last fifteen years PSO has been applied to a very large variety of problems [START_REF] Poli | Analysis of the publications on the applications of Particle Swarm Optimisation[END_REF] and numerous variants of the algorithm have been presented [START_REF] Banks | A review of Particle Swarm Optimization. Part I: background and development[END_REF].

During the execution of PSO a set of particles moves within the function domain searching for the optimum of the function (best fitness value). The motion of each particle is driven by the best positions visited so far by the particle itself and by the entire swarm (gbest PSO) or by some pre-defined neighborhood of the particle (lbest PSO). Consequently, each particle relies both on "individual" and on "swarm" intelligence, and its motion can be described by the following two simple equations which regulate the position and the velocity updates:

Pn(t) = Pn(t -1) + vn(t) vn(t) = w • vn(t -1) + c1 • rand() • (BPn -Pn(t -1)) + c2 • rand() • (BLPn -Pn(t -1))
where Pn(t) and vn(t) are the position and velocity of the n th particle in iteration t; c1, c2 and w (inertia factor) are positive constants, rand() returns random values uniformly distributed in [0, 1], BPn is the best-fitness position visited so far by the particle and BLPn is the best-fitness position visited so far by any particle of a neighborhood of the particle (which may be as large as the current swarm: in this case, this position would be the global best).

In particular, the PSO version implemented in this library is the same described in [START_REF] Mussi | Evaluation of parallel particle swarm optimization algorithms within the CUDA TM architecture[END_REF]: an lbest PSO relying on a ring topology with two neighbors and constant inertia factor.

Differential Evolution

Differential Evolution [START_REF] Storn | Differential Evolution-A Simple and Efficient Adaptive Scheme for Global Optimization over Continuous Spaces[END_REF] recently gained credit as one of the most successful evolutionary algorithms. DE perturbs the current population members with the scaled differences of other individuals. Every element of the population acts as a parent vector and, for each of them, a donor vector is created. In the basic version of DE, the donor vector for the i th parent (Xi) is generated by combining three random and distinct elements Xr1, Xr2 and Xr3. The donor vector Vi is calculated as:

Vi = Xr1 + F • (Xr2 -Xr3)
where F (scale factor) is a parameter that strongly influences DE's performances and typically lies in the interval [0.4, 1]. After mutation, every parent-donor pair generates an offspring (called the trial vector) by means of a crossover operation. Cr is called crossover rate and appears as one of the control parameters of DE, like F . This trial vector is then evaluated and, if its fitness is better than the parent's, it will eventually replace it.

The library offers the choice between the two most commonly used kinds of crossover (binomial, also called uniform, and exponential). With respect to mutation schemes [START_REF] Das | Differential Evolution: A Survey of the State-of-the-Art[END_REF], DE/rand/1 (explained above), DE/target-to-best/1, and DE/best/1 are available.

Scatter Search

Scatter Search [START_REF] Glover | Scatter search[END_REF] is a population-based algorithm in which a systematic combination between solutions (instead of a randomized one, as usually happens in EC) taken from a smaller pool of evolved solutions named the reference set R (usually around ten times lower than typical EC population sizes). R is drawn from a randomly initialized population, and it is composed of:

• a Best Set comprising the |B1| individuals of the initial population having the best fitness function;

• a Diverse Set with the |B2| farthest individuals from the Best Set (with

|R| = |B1| + |B2|).
New offspring is generated by combining elements of the two sets. After this step, a local search is performed to improve the offspring and the two sets are eventually updated by including the best and the most diverse elements, respectively. If, after an iteration, no elements are replaced in R, a new population is created and the individuals that are the most distant from the Best Set are used to replace the Diverse Set. These operations are repeated until a stopping criterion is met.

The current implementation of SS in this library uses the BLX-α crossover as combination method, Solis&Wets as an improvement technique, while, for the update, two strategies are allowed: the former preserves only the best solutions, the latter preserves both the best ones and the most diverse ones with respect to R.

Solis&Wets local search

Solis&Wets local search method [START_REF] Solis | Minimization by Random Search Techniques[END_REF] is a randomized hillclimber with adaptive step size. Each step starts at a point x. A perturbation p is randomly chosen from a Gaussian distribution with standard deviation ρ. If either x + p or xp has a better fitness than x, a move to the best point is performed and a success is recorded, otherwise the position does not change and a failure is recorded. After N + consecutive successes ρ is increased, for getting faster to the local optima, while after N -failures in a row, ρ is consequently decreased.

In the library, SW starts with a different ρ for every solution, that equals half the distance to the nearest neighbor. The user can set the parameters N + and N -.

General-Purpose Computation on Graphics Hardware

General-purpose programming on GPU (GPGPU) is the way of using a graphic card, which typically handles computations only for computer graphics and gaming, to execute applications traditionally managed by the Central Processing Unit (CPU).

CUDA TM (Compute Unified Distributed Architecture) is a GPGPU environment, that includes a parallel computing architecture and programming model, developed by nVIDIA TM . This programming model requires the problem under consideration be partitioned into sub-problems, that are solved independently in parallel by blocks of threads. In turn, each sub-problem is also partitioned into finer pieces, that can be solved cooperatively in parallel by all threads within the same block. Blocks are organized into a one-dimensional, two-dimensional, or three-dimensional grid of thread blocks, as illustrated in Figure 1.

In particular, nVIDIA TM CUDA-C [11] is an extension of the C language for the development of GPUs routines (called kernels), which, when called, are executed N times in parallel by N different CUDA TM threads. Kernels are run on the device (GPU) while the rest of the code runs on the host (CPU). It is also important to notice that in CUDA TM , host and devices have separate memory spaces and, in order to execute a kernel, the programmer needs to explicitly allocate memory on the device and, if needed, transfer data from and back to the host. This is the main bottleneck which is encountered: to optimize code for speed the programmer should reduce as much as possible the amount of these transfers.

Regarding the programming model, and in order to understand the implementation details and the case studies under consideration, there are four variables particularly useful for users who want to deeply dig into the library:

• gridDim, that stores the dimensions of the grid as specified during kernel invocation;

• blockIdx, that refers to the block index within the grid;

• blockDim, that contains the dimensions of the block, and

• threadIdx, that stores the thread index within the block.

Figure 1: Grid of Thread Blocks [11].

Each of these variables is a dim3, a data type of CUDA-C that represents a three-dimensional vector, whose elements can be accessed as the elements x, y and z.

Implementation details

libCudaOptimize is entirely written in C++ and CUDA-C and relies on two classes: IOptimizer and SolutionSet (see Figure 2). The former is an abstract class that includes all methods used for evolving a set of solutions (or population/swarm, where every particular solution is an individual/particle, depending on the used terminology), for setting evolution parameters and a reference to the set (it can evolve more than one set in parallel), represented by an instance of the class SolutionSet. Every different metaheuristic is implemented as a sub-class of IOptimizer. All these classes (see some examples at the bottom of Figure 2) have methods that allow a user to set the parameters of the metaheuristic. Moreover, most of the relevant parameters can be passed to the optimizer at the moment of its instantiation.

The class SolutionSet represents one or more sets of solutions and can be accessed in the user-defined fitness function, where it is used to access the elements of the population and to update their fitnesses after evaluation. There are methods that allow users to access the solutions, and their corresponding fitnesses, both on the device and the host. In this way, the user can employ these information both on C++ and CUDA-C function easily.

RUNNING THE LIBRARY

In this section we will briefly describe the operations needed to install the library and to start working with it.

How to install it

libCudaOptimize has been tested on Windows 7 and Ubuntu Linux, using graphics cards with compute capability (CC) ranging from 1.3 to 2.1. GPUs with lower CC impose some limitations to the use of the library due to hardware limits: e.g. using a GPU with CC 2.0 or higher, the limit on the dimensions of a solution is 1024, while with a GPU with CC 1.3 this limit is lowered to 512.

! " # #$ % # " # #$! & ' ( ) ## ) * # +$ ,- * # +$ , * + ,. /
In order to install libCudaOptimize, the following steps should be followed:

1. Download from the project webpage on Sourceforge (http://sourceforge.net/p/libcudaoptimize/) the packages: libCudaOptimize, the proper library, and testLibCu-daOptimize, a small program which tests if the installation has been correctly performed. Both source code and binary installer are provided.

2. Build the library from source code: for doing so, the easiest way is to use CMake 1 . To build testLibCud-aOptimize one has to link against the previously installed libCudaOptimize static library and the CUDA utilities library.

How to use the library

Basically, there are two ways to use this library. The first and most direct one is just to apply the included heuristics to optimize a user-defined fitness function. All one needs to do, in this simplest case, is to write a function in C++ or, to fully exploit the parallelization potentiality of the package, in CUDA-C. Then, one must select the heuristic, set its parameters, run it, and retrieve the solution(s) found. An example of this usage is shown in section 4.1.

The second purpose of the library is to allow the user to design and implement an optimization technique, taking advantage of the structure of the algorithms implemented in libCudaOptimize. Since several EC methods share a similar structure, one can extend the superclass IOptimizer or one of its children in order to create a new optimizer. To do so, a mandatory step is to implement the four protected functions of IOptimizer shown in Figure 2: 1 www.cmake.org

• initSolutions initializes the candidate solutions within the search space, e.g. random initialization;

• step defines how the optimizer generates new potential solutions from the current population;

• fitnessEvaluation calls the user's fitness function;

• update is called after fitness evaluation and should update the population according to the results obtained: replace current individuals, update personal best locations, check constraints, . . . An example of this usage can be found in section 4.2.

It is important to notice that the user does not have to handle memory allocations and releases nor grid and kernels configuration, since these operations are taken care for by the library core.

CASE STUDIES

Optimization of a fitness function

The first case study examined is a method for automatic localization of the hippocampus in mouse brain histological images [START_REF] Ugolotti | Automatic Hippocampus Localization in Histological Images using PSO-Based Deformable Models[END_REF]. The localization uses two parametric models that are moved and deformed towards the image, in order to maximize a fitness function which is proportional to their overlap with two major structures of the hippocampus (see Figure 3) using PSO to generate candidate solutions. In the first implementation of this method, in which PSO was coded in Matlab and the fitness function in C++, these two parts were localized sequentially one after the other, but now this process can be run independently in parallel by two swarms.

The fitness function takes as inputs: (i) the image that contains the hippocampus, (ii) some vectors that represent the limits of the deformations, and (iii) another vector that represents the coordinates of the model. The values of this vector (usually having between 14 and 16 elements) are optimized by the methods provided by libCudaOptimize. Thanks to the library features, we can evolve two swarms in parallel, each of which will be responsible of one model. The first step that was performed was the conversion of the fitness function from C++ to CUDA-C, which is quite straightforward and will not be explained here. It is worth noting that one needs to supply a C++ function pointer to the optimizer in order to call the CUDA-C kernel (fitness function). This C++ fitness function receives as input a pointer to the SolutionSet object, from which it can access the coordinates and the fitnesses of the swarm(s), which will be passed to CUDA-C kernel, along with any other data that is needed: void localizeHippocampus(SolutionSet* s) { localizeHippocampus_kernel<<<...,...>>> (s->getDevicePositions(), s->getDeviceFitnesses(), ...); }

In the main function, all one has to do is choose the optimization technique, e.g. PSO, and create an instance of the correct extension of the class IOptimizer, setting, if necessary, some parameters (during the instantiation, or later using the functions listed in Figure 2). In this case, a pointer to the fitness function and the number of swarms (2), of solutions (64), and the dimension of the problem [START_REF] Storn | Differential Evolution-A Simple and Efficient Adaptive Scheme for Global Optimization over Continuous Spaces[END_REF].

One can also set other parameters for the optimization, like the termination criteria (that can be total execution time, number of fitness evaluations, number of generations/iterations, or required fitness value). After all the parameters have been set, one has just to start the optimization process and retrieve the results: void main() { PSO_Optimizer p(&localizeHippocampus, 16, 2, 64); p.optimize(); float* myResults = p.getBestSolution(); } As one can see, a C++ (or CUDA-C) fitness function can be bound to libCudaOptimize by adding less than 10 lines of code.

Figure 4 and5 show the execution time we obtained using libCudaOptimize changing the number of PSO generations (from 100 to 10000) and the number of particles of the swarms (from 16 to 512), respectively. Three different implementations have been compared:

1. both fitness function and PSO implemented in C++; 2. fitness function implemented in C++ using our parallel PSO;

3. both fitness function and PSO implemented in CUDA-C;

Tests were run on a 64-bit Intel(R) Core TM i7 CPU running at 2.67GHz using CUDA v. 4.1 on a nVIDIA TM GeForce GTS450 graphics card with 1GB of DDR memory and compute capability 2.1. For the first test, whose results are plotted in Figure 4, the optimization was run 10 times for each configuration, for a total of 1000 × 3 runs. As for the second, depicted in Figure 5, a total of 1250 × 3 experiments was performed. The plots clearly show the GPU version outperforming the two other implementations. In Figure 5, the CPU version appears to have better performance than the CPU+GPU one with smaller swarms (less than 60 particles). This is due to the overhead introduced by the data transfer between host and device memory. However, when dealing with more complex configurations this overhead is overshadowed by the advantages of the parallel PSO execution.

Implementation of a new optimization method

As a second case study, we extended the library by implementing a metaheuristic which was not included in the basic set of optimizers provided by the library: Simulated Annealing (SA) [START_REF] Kirkpatrick | Optimization by Simulated Annealing[END_REF]. This technique tries to mimic the physical annealing process, where a material is heated and slowly cooled into a uniform structure. SA may even perform bad moves (i.e. changes which leads to worse fitness) accordingly to a probability distribution dependent on the temperature of the system: a move is selected at random and, then, as the temperature decreases, the probability of accepting a bad move decreases (when temperature is zero, no bad moves are accepted, i.e. it behaves like hill climbing). The main building blocks of the algorithm are three: the candidate solution generation method (sometimes referred to as the neighborhood function), the acceptance probability function mentioned earlier, and the cooling schedule.

To make SA conform with the notions of a populationbased technique, we adapted the method for considering more than one candidate solution at the same time. For the sake of simplicity, the neighborhood function simply generates random solutions from the current solution positions with a standard deviation of 0.1. The acceptance probability function is defined as follows:

P (x, x′, T ) = exp( f (x) -f (x′) T )
where P is the probability function, x the current solution, x′ the candidate solution, T the temperature of the system, and f (x) the fitness of a solution. As for the cooling schedule, the temperature value is reduced by a factor of 0.98 after every iteration of the algorithm. Moving on to implementation details, code snippets are provided to explain the process. Firstly, we need to extend the IOptimizer implementing the required methods: The SA_Optimizer class has only two more member variables: a SolutionSet instance representing the neighbors of the current state in an iteration, and a float variable as the temperature of the system. The method initSolutions() is implemented similarly to the initialization functions of DE_Optimizer and PSO_Optimizer classes, in which the positions of the population of solutions are set randomly over the search space. The only functions that had to be implemented from scratch were step and update, by specifying a neighborhood generating procedure and the acceptance probability function respectively. Each of these functions calls a CUDA-C kernel to execute the appropriate method in parallel, specifying the thread block configuration required. Every thread block performs the calculations for a solution in a set, while every thread in the block represents a dimension of the problem under optimization. Code snippets for the host and device functions for the step function, used for creating new candidate solutions that will be evaluated by h_cudaSA_step is a host function whose only purpose is to call the kernel sa_step, which can not be called directly from a C++ class. This kernel is executed simultaneously by as many threads as the problem dimension. In practice, the number of threads for a given kernel should be a power of half the GPU warp size to achieve the coalesced access to global memory recommended by nVIDIA TM [START_REF]nVIDIA CUDA C programming -Best practices guide[END_REF]. The code above shows the most common and basic implementation of a neighborhood function, used by many derivative-free optimization methods. These algorithms usually test candidate positions sampled from a fixed or adaptive neighborhood around the current position. In our case, we generate candidate solutions by uniform sampling from a hypercube with side length of 0.2 and centered at the current solution. The CuRand library supplied with the CUDA SDK is used to generate random numbers, where, for every dimension of each solution a random number generator (curandState) is stored in a global memory array (devStates). First, the random number generator is loaded into a thread local variable for efficiency. Then, a new position y is computed for every dimension by adding a uniform random number in the range [-0.1, 0.1] to the current position. Finally, the neigh-bor position along with the random number generator are stored back in global memory at the end of the kernel.

#
It is worth noting the way the global memory is being accessed in the device code. blockIdx.x represents the set index, blockIdx.y represents the solution index within the set, and threadIdx.x is the dimension index in the search space. This SA code should be used as a guideline for the advanced user to extend the library with new optimization techniques.

CONCLUSIONS

We introduced libCudaOptimize, a free, fast and portable open source library for continuous optimization based on CUDA. Two practical examples of using and extending the library have been shown: the optimization of a fitness function which takes advantage of this framework, and the implementation of a new optimization technique that could be useful for solving more problems.

Regarding future work, several aspects can be improved or extended. Our next efforts will mainly be concerned about: the realization of some visualization and statistical tools in order to help behavioral analysis of EC techniques; more support for multiple solution sets, like allowing different sets to have independent termination criteria; the possibility to evolve solutions of data types other than floats; and the parallel implementation of other well-known optimization methods like Genetic Algorithms, Evolution Strategies or Evolutionary Programming as well as further expansions of the methods already present.

Figure 2 :

 2 Figure 2: UML diagram. For every class, the most important methods are shown.

Figure 3 :

 3 Figure 3: Example of a hippocampus and the result of the localization phase.

Figure 4 :Figure 5 :

 45 Figure 4: Execution time versus PSO generations, swarm size fixed at 64 particles.

Acknowledgments

Youssef S. G. Nashed and Pablo Mesejo are funded by the European Commission (Marie Curie ITN MIBISOC, FP7 PEOPLE-ITN-2008, GA n. 238819). Roberto Ugolotti is funded by Compagnia di San Paolo (Programma Neuroscienze) and Fondazione Cariparma.


Amin Mesmoudi
email: amin.mesmoudi@liris.cnrs.fr

Mohand-Saïd Hacid
email: mshacid@liris.cnrs.fr

Farouk Toumani
email: ftoumani@isima.fr

Blaise Pascal

Benchmarking SQL On MapReduce systems using large astronomy databases

. In order to evaluate the performances of existing SQL On MapReduce data management systems, we conducted extensive experiments by using data and queries from the area of cosmology. The goal of this work is to report on the ability of such systems to support large scale declarative queries. We mainly investigated the impact of data partitioning, indexing and compression on query execution performances.

Introduction and Context

In the age of bigdata, a massive set of digital information of unprecedented volumes are being collected and/or produced in several application domains, making it increasingly difficult to manage and query large data repositories. While traditional database management systems (DBMS) have gain their reputation thanks to their support of SQL, a declarative, simple and efficient language, they turn out to be less effective when it comes to manage very large data repositories [START_REF] Stonebraker | The end of an architectural era:(it's time for a complete rewrite)[END_REF]. As a consequence, there has been recently an increasing trend towards using ad hoc tools instead of DBMS in several data management applications. Being aware of this fact, database research and development community and DBMS vendors [START_REF] Gartner | Big data challenges[END_REF] have been conducting intensive efforts towards the development of advanced tools intended to be embodied in the advanced generations of DBMS products in order to support big data management. Several application domains, like social networks [START_REF] Smith | Reinventing Social Media[END_REF], astronomy [START_REF] Shaw | Lsst data challenge handbook[END_REF] and Web [START_REF] Dean | Mapreduce: Simplified data processing on large clusters[END_REF] express a clear need for new approaches for data management and analysis. For instance, Google provided the basis of MapReduce [START_REF] Dean | Mapreduce: Simplified data processing on large clusters[END_REF], a new computing paradigm designed to enable efficient processing of huge amounts of data on a multitude of machines in a cluster. MapReduce provides a simple programming framework that enables harnessing the power of very large data centers, while hiding low level programming details related to parallelization, fault tolerance, and load balancing. When using such a framework, users have only to specify two functions, namely Map and Reduce, and the system (e.g., Hadoop-http://hadoop.apache.org/) schedules 1 a parallel execution of these functions over the available nodes of a given cluster. The MapReduce approach has been proven particularly efficient to implement one pass algorithms (i.e., algorithms that require only a single scan of data) [START_REF] Lämmel | Google's mapreduce programming model-revisited[END_REF].

As part of the Petasky project (http://com.isima.fr/Petasky), the work presented in this paper addresses several challenges related to performance evaluation of MapReduce based systems in the context of a specific application related to analysis and management of scientific data in the field of cosmology. Petasky uses as application context, i.e., data, queries and system requirements, LSST (Large Synoptic Survey Telescope-http://lsst.in2p3.fr/projet-LSST.html), a project which targets the construction of a telescope of a new generation with a light-gathering power among the largest in the world. Data management and analysis is recognized as one of the most challenging aspects of the LSST [START_REF] Shaw | Lsst data challenge handbook[END_REF], as more than 30 TB (TeraBytes) of complex data (3.2 Gigapixel images, uncertain data, multi-scale data) must be processed and stored each night to produce the largest non-proprietary data set in the world. The LSST data management group (http://www.slac.stanford.edu/) recommends the use of an open Source system based on a shared nothing architecture. Two main factors motivated such a recommendation: (i) integration of personalized optimization and adhoc features is facilitated, and (ii) support of declarative queries. Taking into account these recommendations, our main objective was to evaluate the capabilities of some systems to manage LSST data, the expected data volume at the end of the acquisition process is 60 PB (PetaBytes), using as infrastructure a cluster of hundreds of commodity servers. We focused our study on MapReduce systems for the following reasons:

the programming model underlying such approaches enables easy development of scalable parallel applications, even for programmers without experience with distributed systems (this because the system hides the details of parallelization, fault-tolerance, locality optimization, and load balancing), -a large variety of problems can be expressed as MapReduce computations, -a MapReduce framework is able to scale to large clusters with thousands of nodes.

Previous works (e.g. [START_REF] Pavlo | A comparison of approaches to large-scale data analysis[END_REF]) on benchmarking have stressed the technological superiority of parallel DBMS compared to Hadoop, which is mainly due to a lack of 1 By distributing and coordinating the execution of jobs. sophisticated optimization mechanisms (e.g., compression, buffering, indexing) in Hadoop. Recently, new systems (e.g., HadoopDB [START_REF] Abouzeid | Hadoopdb: an architectural hybrid of mapreduce and dbms technologies for analytical workloads[END_REF][START_REF] Bajda-Pawlikowski | Efficient processing of data warehousing queries in a split execution environment[END_REF], Hive [START_REF] Thusoo | Hive-a petabyte scale data warehouse using hadoop[END_REF]) have been proposed to strenghten MapReduce technologies with a declarative query language, in order to facilitate the expression of repeated complex tasks, as well as with sophisticated optimisation techniques. These systems are based on the idea of generating, from an SQL(-Like) query, a set of MapReduce jobs whose execution and resources management will be supported by a MapReduce framework. In this paper, we specifically report on the capability of two such systems, namely Hive and HadoopDB, to accommodate LSST data management requirements.

To achieve our goal, we designed and set up a benchmark to evaluate the performance of SQL On MapReduce data management systems. We used data sets and queries from the area of cosmology, provided by the LSST project. We extensively investigated two systems as use cases: Hive, an HDFS based system and HadoopDB, a Hadoop/PostgreSQL based system. In contrast to existing benchmarks (e.g., TeraSort [START_REF]Sort Benchmark[END_REF], Pavlo et al. [START_REF] Pavlo | A comparison of approaches to large-scale data analysis[END_REF], Intel HiBench [START_REF] Huang | Hibench: A representative and comprehensive hadoop benchmark suite[END_REF]), our benchmark considers real structured data and real queries which makes evaluation of the systems relevant. We also defined additional queries (e.g., distance join,...) which can be very useful to evaluate other kinds of complex computations over structured data. In this context, we conducted extensive experiments to evaluate the two SQL On MapReduce systems to manage (e.g., storage, loading) LSST data on one hand and their capabilities (i.e., indexing, compression, buffering, and partitioning) to optimize some categories of queries on the other hand. In contrast to existing experiments we do not consider only execution time but also loading time. We investigated different configurations obtained by varying the values of the parameters we consider:

-Infrastructure: we used different sizes for the clusters, one with 25 nodes and another with 50 nodes. -Data: we use different datasets with sizes of 250 GB, 500 GB, 1 TB and 2 TB. -Partitioning: we used two partitioning schemas and we compared execution time for both schemas. -Indexing: we run our queries on both indexed and non-indexed data. For Hive, we used different attributes as indexes. -Selectivity: we considered queries with different selectivities (i.e., the number of records that satisfy the query).

Additional information and technical details about this benchmark are available at: http://com.isima.fr/Petasky/hive-vs-hadoopdb. The rest of this paper is organized as follows: In Section 2, we briefly review basic notions related to MapReduce based systems and we highlight the need for a new benchmark for SQL on MapReduce solutions. In Section 3, we present the main features of the datasets used in our experiments. In Section 4, we describe the experiments and we discuss the outcomes in Section 5. We conclude in Section 6.

Background and Related Work

In this section, we review basic notions underlying a MapReduce framework and its extensions to support SQL queries and we discuss existing MapReduce benchmarks.

We start by introducing a running example that will be used through the paper to illustrate the investigated issues.

A running example

We use an example from the field of astronomical observations [START_REF] Shaw | Lsst data challenge handbook[END_REF]. We consider a simple relational database schema which includes two tables. The first table, called Source, is used to store detected astronomical objects extracted from acquired images. This table includes attributes that represent the ID of the observation (SourceID), the ID of the detected object (ObjectID), the world coordinates: Right Ascension (RA) and Declination (Decl ) and science exposure ID (ExposureID). As query, we use a slightly adapted query2 from the LSST workload. This query, given below, returns appearance frequency of detected astronomical objects in the database:

SELECT ObjectID, count(SourceID) as freq (q) FROM Source GROUP BY objectID We shall illustrate now a MapReduce process using our running example. Figure 1 shows the sequence of data processing steps. First, we assume that data is partitioned over nodes of the cluster. An approach to use MapReduce framework to process our sample query is to define Map and Reduce functions as follows:

-Map: for each record assigned to a Mapper, one generates a (key, value) where key is the value of the attribute ObjectID and the value is 1. -Reduce: for each received input < Key, list < V alue >>, one applies the addition on list < value >, i.e., list < f req >.

The system starts by reading each record in the database and generates a set of There are several implementations of the MapReduce framework. The most popular one is Hadoop. Indeed, Hadoop uses HDFS [START_REF] Shvachko | The hadoop distributed file system[END_REF] to store and partition data on nodes of a cluster. Data is partitioned in chunks (by default each chunk has a size of 64 MB). Each Map task is assigned to a chunk. Hadoop also offers default functions that can be used as combiners and partitionners. These functions could easily be replaced by other functions defined by the user. Other implementations of MapReduce, e.g., Themis and SailFish, exist. Themis [START_REF] Rasmussen | Themis: an i/o-efficient mapreduce[END_REF] is a MapReduce implementation designed for small clusters where the probability of node failure is much lower. It achieves high efficiency by eliminating task-level fault tolerance. Compared to Hadoop, Sailfish [START_REF] Rao | Sailfish: A framework for large scale data processing[END_REF] changes the transport layer between Mappers and Reducers. The authors in [START_REF] Rao | Sailfish: A framework for large scale data processing[END_REF] tried to improve the disk subsystem performance for large scale MapReduce computations by aggregating intermediate data.

A new version of Hadoop, called YARN [START_REF] Vavilapalli | Apache hadoop yarn: Yet another resource negotiator[END_REF], enhances the power of a Hadoop framework by improving cluster utilization and providing support for workloads other than MapReduce. Tez [START_REF] Tez | [END_REF] is another in-progress project which allows to speed up data processing across both small-scale, low-latency and large-scale, high-throughput workloads. This is achieved by eliminating unnecessary tasks, synchronization barriers, and reads from and write to HDFS. Other implementations of MapReduce on specialized hardware are proposed. For example, Mars [START_REF] He | Mars: a mapreduce framework on graphics processors[END_REF] is a MapReduce framework on graphics processors.

Relational operations on top of MapReduce

A lot of recent research work has been devoted to the implementation and optimization of relational operations using the MapReduce computation model. For example, Select-P roject queries have been initially implemented using a single Map function. However, a drawback of such a solution comes from a need to achieve a full scan of all the data for each query, and this is a very costly task. To overcome such a limitation, main optimization techniques consist in the use of indexing and data compression. Instead of transferring all key/values pairs of chunks to Mappers, a filtering phase is applied to identify only relevant data to be transferred. With respect to data compression, a major proposal lies in the notion of RCFile [START_REF] He | Rcfile: A fast and space-efficient data placement structure in mapreduce-based warehouse systems[END_REF]. In this case, less data is transferred from disk to RAM. Besides Select-P roject, implementations of other relational operators have been proposed in the literature. Implementation of GROU P BY (see figure 1) queries is discussed in section 2.2 while Sorting (ORDERBY ) will be discussed in section 4.6.4.

The join operation is very expensive if one considers its implementation using the MapReduce model [START_REF] Afrati | Optimizing joins in a map-reduce environment[END_REF]. Indeed, in the case of non constrained inputs, many records that do not contribute to the final results will be transferred to Reducers. Some sophisticated implementations have been proposed to overcome the limitations of the naive approach. For example, Afrati et al. propose to control the number of Reducers to take advantage of the available resources for simple equijoin [START_REF] Afrati | Optimizing joins in a map-reduce environment[END_REF] and Multi-way equi-join [START_REF] Afrati | Optimizing multiway joins in a map-reduce environment[END_REF].

Another relevant (non classical) operator for the LSST queries is similarity join. Indeed, this operation allows to group together records that have a distance (computed from some attributes) less or equal than a given threshold. Several approaches (e.g., [START_REF] Silva | Exploiting mapreduce-based similarity joins[END_REF][START_REF] Okcan | Processing theta-joins using mapreduce[END_REF][START_REF] Metwally | V-smart-join: A scalable mapreduce framework for all-pair similarity joins of multisets and vectors[END_REF][START_REF] Lu | Efficient processing of k nearest neighbor joins using mapreduce[END_REF][START_REF] Kim | Parallel top-k similarity join algorithms using mapreduce[END_REF]) have been proposed to compute some kinds of similarity joins using the MapReduce model. For example, if one wants to group together observations with a distance less than [START_REF] He | Mars: a mapreduce framework on graphics processors[END_REF] The attributes S.Ra and S.DECL record the right ascension and declination, in degrees, related to a given observation S.

SQL on MapReduce

Several frameworks have been proposed to extend MapReduce models with SQL capabilities. We distinguish between two types of systems. In the first type, the implementation of the MapReduce framework does not change. For instance, Hive proposes to transform SQL queries into a set of MapReduce Jobs. HadoopDB follows the same principle but it uses an existing RDBMS to store the data in each worker, instead of the HDFS which is used by Hive. This allows to exploit indexing, buffering and compression capabilities offered by DBMSs.

The second type of systems requires a substantial change in the MapReduce framework. Essentially, this type of tools is based on the idea of loading all the data into the memory. For example, Shark [START_REF] Xin | Shark: Sql and rich analytics at scale[END_REF] is based on the Spark [START_REF] Spark | [END_REF] framework. Impala [2] is a Hive-compatible SQL engine with its own MPP (Massively Parallel Processing) execution engine. Stinger [START_REF] Stinger | [END_REF] is another in-progress project based on Tez. Presto [3] is an open source distributed SQL developed by Facebook for interactive queries. Other tools are proposed such as BigSQL [1] which is based on Hadoop on one hand and PostgresQL on the other hand.

MapReduce benchmark

Several benchmarks for MapReduce frameworks exist. Historically, the first one is TeraSort [START_REF]Sort Benchmark[END_REF] which is devoted to test MapReduce implementations using a large scale sort application. Hadoop offers a set of tools to benchmark several implementations of MapReduce frameworks. For example, TestDFSIO [START_REF]TestDFSIO[END_REF] is a tool devoted to test I/O capabilities of DFS systems used in MapReduce frameworks. Pavlo et al. [START_REF] Pavlo | A comparison of approaches to large-scale data analysis[END_REF] proposed a customized benchmark tailored for search engines where a comparison between Hadoop and two parallel DBMS (Vertica and DBMS-X) was proposed. The conclusion of that work highlights the inability of existing implementations (the version of 0.19 of Hadoop has been used ) of the MapReduce model to effectively process data. Three tasks have been tested: selection, aggregation and join.

The maximum number of attributes of the tables used in the experiments is nine [START_REF] Abouzeid | Hadoopdb: an architectural hybrid of mapreduce and dbms technologies for analytical workloads[END_REF] and the considered join query involves two tables. Authors stressed the technological superiority of parallel DBMS compared to Hadoop, which is mainly due to a lack of sophisticated optimization mechanisms (e.g., compression, buffering, indexing) in Hadoop.

In [START_REF] Abouzeid | Hadoopdb: an architectural hybrid of mapreduce and dbms technologies for analytical workloads[END_REF] the same benchmark was used. Authors compared Parallel DBMS, HadoopDB and Hadoop (as an implementation of MapReduce framework). As conclusion, authors noticed that a hybrid system (i.e., HadoopDB) MapReduce and DBMS could achieve performances of parallel DBMSs. Authors in [START_REF] Bajda-Pawlikowski | Efficient processing of data warehousing queries in a split execution environment[END_REF] extended experiments in [START_REF] Abouzeid | Hadoopdb: an architectural hybrid of mapreduce and dbms technologies for analytical workloads[END_REF] by comparing Hadapt (commercial version of HadoopDB), Hive and parallel DBMSs. We believe that this comparison is not complete. Indeed, authors tried to highlight the advantages of Hadapt compared to Hive and parallel DBMS without considering all cases.

Since then, several extensions have been proposed to optimize the performance of MapReduce programs. We review in the sequel the impact of existing techniques on performances. We also explain in details when one of the two tools (Hive, HadoopDB) outperforms the other.

Intel Hibench [START_REF] Huang | Hibench: A representative and comprehensive hadoop benchmark suite[END_REF] suite contains nine (9) workloads, including micro benchmarks, HDFS benchmarks, web search benchmarks, machine learning benchmarks, and data analytics benchmarks. Intel Hibench includes a dataset which is used in Pavlo et al. [START_REF] Pavlo | A comparison of approaches to large-scale data analysis[END_REF] for analytic tasks. In this work, the authors were interested in measuring resources consumption (e.g., RAM, DISK, CPU) for each execution step within a MapReduce program. BigDataBench [START_REF] Zhu | Bigop: Generating comprehensive big data workloads as a benchmarking framework[END_REF] is a benchmark suite that proposes various datasets and workloads for various applications from internet services domain. The datasets cover Wikipedia Entries, Amazon Movie Reviews, Google Web Graph, Facebook Social Network, E-commerce Transaction Data, ProfSearch Person Resumes, CALDA Data and TPC-DS Web Data. All the considered datasets, but E-commerce transaction data which uses a relational schema, include unstructured or graph data and focus on basic operations (e.g., grep, wordcount, sort) or analytical processing (k-mean, pagerank, ...). We view our benchmark as complementary to existing benchmarks, including Intel Hibench and BigDataBench, for several reasons. First, our benchmark focuses on an application domain, data management in the field of cosmology, which is not covered by these benchmarks. Datasets and query workloads in this field display some specific features related to the structure of the data, data volumes, complexity of queries (e.g., using specific UDFs) which are not provided by Intel Hibench and Big-DataBench benchmarks. Moreover, both Intel Hibench and BigDataBench target general MapReduce related applications while, in our work, we especially target SQL on MapReduce systems. We are interested in the implementation and the optimization of SQL operators using MapReduce model. Our datasets and workload allow us to conduct deep analysis on various aspects specifically related to this kind of systems. Indeed, we allow analysis of the behaviour of these systems with respect to selectivity, partitioning, infrastructure and (data volume) scalability. To the best of our knowledge the two benchmarks (i.e., Hibench, BigDataBench) do not allow to investigate in-depth query evaluation and optimization techniques with scalable structured data.

Our benchmark rests on a real case study. We consider structured data from astronomy where a set of SQL queries are already defined. Our experiments target SQL On MapReduce systems. Our objective is to report on the capabilities of those systems to manage (e.g., storage, loading) LSST data on one hand and their capabilities (i.e., indexing, compression, buffering, and partitioning) to optimize some categories of queries on the other hand. In contrast to existing experiments we do not consider only execution time but also loading time. We consider different configurations for the different parameters: Infrastructure, Data, Partitioning, Indexing and Selectivity. Also, our benchmark will serve as a baseline to assess query evaluation and optimization techniques that will be proposed in the context of the Petasky project.

Data characteristics

We focus on the issue of data management in the field of cosmology. Data are stored using relational tables where information about exposures, objects and observations can be found. The database is organized in the following four categories (represented as tables):

1. Exposure: describes each exposure, including the start date/time of exposure, the used filter, the position and orientation of the field of view on the sky, and other environmental information. 2. Object: describes attributes of each astrophysical object, including the world coordinates. 3. Source: describes each detected source on each image, including its location (x, y) on the detector, world coordinates (RA, Decl), brightness, size, and shape. 4. Moving Object: describes attributes of moving (solar system) objects, including orbital elements, brightness, albedo and rotation period. 5. Forced Source: describes measurements computed using source records.

Table 2 shows, for some tables, the estimated size of expected data at the end of the observation (which will start in 2020 for a duration of 10 years), the number of attributes as well as the expected final number of tuples. It is worth noting that: (i) the size of the data varies from Gigabytes (e.g., size of the table Moving Object) to a few Petabytes (e.g. size of the tables Source and ForcedSource), (ii) the number of records ranges from few millions (e.g., for the table Moving Object) to trillions (e.g., for the Source table), and (iii) the number of attributes varies from 7, for ForcedSource table, to 450 for the Object table.

In order to evaluate the capacity of existing systems to manage LSST data, the data management group of the LSST project has provided some datasets and a tool to generate other datasets which are compliant with the characteristics of real data. Table 3a shows the size of the four datasets PT1. Hive proposes HiveQL, a SQL-like language, to specify analysis tasks. Data is stored using HDFS. From an SQL-like query, Hive enables generating a set of MapReduce tasks and, in addition, it schedules the execution of the generated tasks. In our experiments, we used Hive 0.11.

HadoopDB

HadoopDB is based on Hive and uses HiveQL. Existing DBMS are used to store data in the nodes of the cluster. An execution plan, generated by Hive, is processed to push more complex tasks, as for example SQL queries, to nodes in the Map phase.

HadoopDB uses an XML file 7 to store access information to data. In particular, information about chunks and sub-chunks, related to each table, is stored in this file. HadoopDB processes queries as follows. First, HadoopDB parses an input query by resorting to Hive's mechanisms. Second, the catalog is parsed and the query is reformulated with respect to chunks and sub chunks information in the catalog. Then, a set of MapReduce jobs are generated together with a predefined execution order. In the Map phase of HadoopDB job, a query is sent to the DBMS and responses are formatted in a <key, value> format. After the execution of the Map and Reduce phases is completed, HadoopDB checks whether there is another job to execute. If any, partial results are stored in HDFS. If there is no additional job to be executed, responses are transferred to the node running a Hive client. In our work, HadoopDB is coupled with postgresQL in each node of the cluster.

Recently, we conducted experimental studies by using centralized systems [START_REF] Mesmoudi | A comparison of systems to large-scale data access[END_REF][START_REF] Mesmoudi | A test framework for large scale declarative queries: preliminary results[END_REF]: Mysql, PostgreSQL and DBMS-X (a commercialized DBMS). We found that PostgreSQL outperforms Mysql and DBMS-X with respect to LSST data. Also, in [START_REF] Abouzeid | Hadoopdb: an architectural hybrid of mapreduce and dbms technologies for analytical workloads[END_REF], the authors noticed that PostgreSQL outperforms Mysql for analytic queries. HadoopDB also offers a partitioning tool. This tool is used to minimize the intermediate results of jobs generated by Hive. The use of existing DBMS as data storage and access layer allows to exploit existing technologies such as indexing, buffering and compression.

Data loading and indexing

In this section, we analyze the data loading and indexing techniques of the two selected systems.

Hive

Data loading. The tables are loaded in the HDFS directly and stored as text files. Figure 2 shows the time needed to load data in the HDFS for the datasets used in our experiments. One can observe that the time needed for 25 nodes and 50 nodes is the same. The time needed to load datasets for a given cluster (e.g., 50 machines) is linear compared to the size of the dataset. If we have 2X of data 8 we need in average 2X of time9 to load data. Indeed, input operations (i.e., reading data from local hard disks) induce the most important part of the loading time.

Data indexing. Hive supports only simple indexes with a single attribute. To deal with our query workload, we created 5 indexes: SourceID, ObjectID, RA, DECL and ScienceCCDExposureId. The time overhead due to creation of indexes is shown on figure 3, whereas the space overhead (i.e., size of the indexes) is shown on figure 4. For each attribute, a set of <key, set of HDFS addresses> is stored in the HDFS. This structure is used to filter relevant records for a given query. Regarding the speed up, we observe a gain of 15% in the indexing time if we use 2X machines. 12

HadoopDB

Data loading. HadoopDB enables a user to customize the data partitioning by selecting a partitioning attribute that could be useful to speed up the processing of some queries. To achieve this task, HadoopDB proceeds in five steps: (i) data is loaded to the HDFS, (ii) a global hashing (partitioning) with respect to the number of nodes in the cluster is performed, (iii) each partition is loaded in a target node, (iv) a local hashing (partitioning) of chunks that can fit in memory is performed, In the same vein, when we move from a cluster of 25 nodes to a cluster of 50 nodes, time overhead (i.e., loading time) drops by 25%. We also evaluated the cost of loading 2 TB of data, after the global hash and during data loading from the HDFS. As our experiments are performed over a cloud platform (virtual environment), we observed a level of failures. Indeed, reading and writing simultaneously up to 40 GB (in parallel on all nodes) leads to frequent failures in the storage platform (ceph 13 ). Compared to Hive, where the time needed is equal to data loading in the HDFS, we observed that HadoopDB needs up to 300% of the time needed by Hive for 25 nodes and up to 200% for 50 nodes.

Data indexing. Indexes in HadoopDB are managed locally by the DBMS. For each chunk (stored in a separate database) a set of indexes is created. Unlike Hive, which creates a global index for the data, HadoopDB leaves this task to the DBMS. An index can be used only to speed up query evaluation on local chunks. A comparison between time needed for index creation and for total data loading is shown on figure 5. The task of creating indexes takes benefit of parallelization and hence makes a better usage of the available resources. Moreover, we can take advantage of hardware resources to accelerate the creation of indexes (50 vs 25 nodes). HadoopDB requires only 6% of the total data loading time and 28% of the data volume.

Queries

In our experiments, we considered a query workload made of different types of queries: SELECTION, GROUP BY, ORDER BY and JOIN. We executed the queries over the three datasets, respectively, of sizes 250 GB, 500 GB and 1 TB. We used two clusters of different sizes: a cluster with 25 nodes and another cluster with 50 nodes. For each task, we evaluated14 queries by considering two configurations: non-indexed data and indexed data.

number of records Table 4 shows the selectivity of queries while table 5 shows the selectivity of the predicates used by the queries.

SELECTION Queries

Table 6 shows the set of SELECTION queries used in our experiments. The query Q 1 retrieves information about a given Source (detection of sky objects) while query Q 2 selects Sources (detection) and position (given by RA, DECL) of a given object. This query enables identifying moving objects. The query Q 3 retrieves Objects and their detection (source) for a given sky area (bounded by RA and DECL). Q 4 returns Detections (SourceID) and their positions from a given Sci-enceCDDExposure. The four queries target the Source Table .   Queries over non indexed data. From a SQL query, Hive and HadoopDB generate a MapReduce Job. Hive processes data SELECTION and PROJECTION within the Map task, whereas HadoopDB incorporates, in the Map task, database queries by using JDBC. A same query is sent to all Mappers. The Reduce function is not needed for the two systems because no additional processing is required after the Map task.

Figure 6 compares the execution time of the two systems (HadoopDB and Hive) using three (3) datasets with 250 GB, 500 GB and 1 TB and a configuration of 50 nodes. We first executed the queries on no indexed data.

Regarding the dataset of 250 GB, Hive outperforms HadoopDB. This is due to a latency of HadoopDB when it accesses to local DBMS using JDBC. Hive does not suffer from the same overhead since it directly accesses the data in the HDFS. However, HadoopDB outperforms Hive for datasets of 500 GB and 1 TB. Indeed, the use of the buffer and data compression (for 1 GB raw data, only 540 MB of disk space is used to store this data) makes HadoopDB more efficient.

Regarding the scalability issue, we observe that if we double the data volume, the execution time increases in average by 250%. For example, the maximal time to query the dataset of 250 GB is 4 mins, whereas, the execution times observed for the same query over the datasets of 500 GB and 1 TB are respectively, 10 mins and 30 mins.

Queries over indexed data. Hive proceeds by filtering relevant parts of the index with respect to the condition expressed in a query. Only one index can be used to evaluate a query. The filtered part (keys, addresses) is then transferred to all the nodes of the cluster and only relevant HDFS blocks are processed. For example, query Q 3 contains two filter conditions, one on the attribute RA and the other on the attribute DECL. We evaluated query Q 3 twice. In the first evaluation, we used the index on RA. Due to its size, which exceeds 1 GB (size of heap memory) Fig. 7: SELECTION tasks with Index for all the datasets, the workers choose to not use this index to evaluate query Q 3 . In the second evaluation, we used an index on DECL. For the dataset of 1 TB, the size of the filtered part exceeds the size of the Heap memory.

In HadoopDB, indexes are automatically used by the local DBMS optimizer in each chunk.

Figure 7 shows the time needed to evaluate the queries using the indexes. One can observe that there is an improvement up to 70X for Hive, whereas, the improvement is only of 9X for HadoopDB.

Regarding data scalability, we need only 1.3X of time if we have 2X of data. For example, the maximal time to query the dataset of 250 GB is 2 mins, whereas, the datasets of 500 GB and 1 TB need 3 mins and 3.5 mins respectively.

Hive outperforms HadoopDB except for the query Q 3 . In most cases, in particular for selective queries, a global index (proposed in Hive) leads to better performances than a local index (proposed in HadoopDB). However, the limit of this strategy is that when the size of the filtered part becomes larger, the nodes are no longer able to manage the indexes.

Aggregate Queries

Table 7 shows the two considered aggregate queries. Query Q 5 computes the number of observations (sources) of a given bounded area of the sky. Only a few tuples (< 10) are expected as a result for this query. The considered area is extended in the query Q 6 , where the expected number of tuples to be returned by the query ranges from 9 to 58 millions of tuples.

Both Hive and HadoopDB use only one MapReduce Job to implement each of these queries. In the Map step, Hive performs data read, PROJECTION and filtering and as well as a partial GROUP BY. In the Reduce step, a global GROUP BY is performed. The number of Map tasks depends on the number of blocks of data in the HDFS. For example, for the 250 GB dataset, 887 Map tasks have been executed, whereas the number of Reduce tasks is determined by Hive using two parameters: the size of Map Input data and the maximum size of data handled by a Reducer (Hive.exec.reducers.bytes.per.reducer). For the dataset with 250 GB, we For HadoopDB, within the Map step, the same query is sent to the local DBMS where PROJECTION, SELECTION and partial GROUP BY are performed. In the Reduce step, the global GROUP BY is performed. The number of Map tasks depends on the number of chunks (and subchunks) created in the data loading phase. For example, for the 250 GB dataset we have 250 Map tasks, whereas the number of Reduce tasks is set to one (1) by HadoopDB. For Hive, and because of its size, the index ObjectID cannot be used for the query Q 6 . For example, for the 250 GB dataset, the size of the ObjectID index is equal to 5 GB. For Q 5 , only RA can be used and only for the datasets with the sizes 250 GB and 500 GB. Indeed, the sizes of the filtered part of indexes of DECL and ObjectID exceed the heap memory dedicated to the Mapper (> 1 GB).

A comparison of the execution time of queries Q 5 and Q 6 is shown on figure 8. We can observe that, for selective queries (e.g., query Q 5 ), HadoopDB outperforms Hive for both indexed and non indexed data. Indeed, due to the few tuples (< 10) transferred from Mappers to Reducers, the time needed to handle the Map tasks represents the most important part of the total time needed to process the whole query. The ability of HadoopDB, unlike Hive, to process compressed data (e.g., for the dataset with 1 TB, 554 GB of compressed data for HadoopDB and 944 GB of non-compressed data for Hive) in the Map phase makes HadoopDB more efficient. The difference in performances of HadoopDB compared to Hive can reach up to 15X, with a maximum of 35 mins of processing time for Hive and 12.5 mins for HadoopDB.

Due to the size of data transferred from Mappers to Reducers (e.g., 9 GB transferred in the case of the dataset of 1 TB), the time needed to handle the Reduce tasks impacts the global time needed to perform the whole query. Indeed, for nonselective queries (e.g., query Q 6 ), Hive parallelizes the processing of GROUP BY in the Reduce phase by using all the nodes of the cluster as Reducers, whereas, for HadoopDB only one node is used as a Reducer. In the case of non-indexed data, Hive is significantly better than HadoopDB (with and without indexes). Indexes are useless in this case because all the records must be accessed (full scan). The difference in performances of Hive compared to HadoopDB can reach up to 2X, with a maximum of 36 mins for Hive and 1 hour 10 mins for HadoopDB.

Regarding data scalability, we need in average only 2X of time if we have 2X of data. For example, the maximum time to query the dataset of 250 GB is 20 mins, whereas, the datasets with 500 GB and 1 TB need 40 mins and 1 hour 10 mins respectively.

In the case of non selective queries, the use of only one Reducer represents a bottleneck for HadoopDB. To optimize this kind of queries, HadoopDB proposes to modify the partitioning attribute. Indeed, our results were obtained while the Source table is partitioned using the SourceID attribute. When we change the partitioning attribute to ObjectID, we observe an improvement over the previous results, which can reach up to 5X. The results are shown in figure 9. The modification of the partitioning attribute allows: (i) to minimize the number of <key, value> transferred from Mappers to the Reducer and therefore the size of data to be processed by the Reducer, and (ii) to avoid the global GROUP BY in the Reducer side which was already done by the Mappers.

JOIN Queries

Table 8 shows the considered JOIN queries. Queries Q 7 and Q 5 express a JOIN between the tables Source and Object. The two queries retrieve information about objects and their observations (sources). In query Q 7 , the sky area is constrained The sky area of query Q 7 is included in the sky area of query Q 8 . Query Q 9 expresses a JOIN between three tables: Source, Science Ccd Exposure Metadata and RefSrcMatch. This query retrieves the Sources, from a given sky area, and their related metadata obtained from Science Ccd Exposure Metadata and RefS-rcMatch.

Query SQL syntax Q 7 SELECT *
HadoopDB does not support queries with Joins involving more than two tables, therefore the query Q 9 is not considered.

The number of MapReduce Jobs generated by Hive depends on the number of Joins in the query. For each JOIN, Hive creates a Job. Therefore, for queries Q 7 and Q 8 , Hive generates only one (1) job while it generates two (2) jobs for query Q 9 . In the case of two Joins or more, the result of each JOIN is used as an input for the next JOIN. Hive stores the partial results in the HDFS.

In the Map step, Hive performs data read, PROJECTION, filtering and tuples hashing (to group together tuples with same values of the JOIN key), while in the Query SQL syntax 10 Reduce step, a join is performed. The number of Map tasks depends on the number of blocks of data in the HDFS (e.g., for the 250 GB dataset and queries Q 7 and Q 8 , we have 938 Map tasks) whereas the number of Reduce tasks is set by Hive (e.g., for the 250 GB dataset and queries Q 7 and Q 8 , we have 252 Reduce tasks)

For HadoopDB, within the Map phase, the same Query is sent to the local DBMS where PROJECTION and SELECTION are performed. In the Reduce step, the JOIN is performed. The number of Map tasks depends on the number of chunks (and subchunks) created in the data loading phase (e.g., for the 250 GB dataset we have 250 Map tasks) whereas the number of Reduce tasks is set to one (1) by HadoopDB.

Figure 10 shows the execution time of the considered join queries. Hive outperforms HadoopDB for both queries Q 7 and Q 8 . Indeed, for the query Q 7 the difference of performance can reach 1.7X, whereas for the query Q 8 , the difference can exceed 12X.

As noticed for the GROUP BY queries, when the size of the result returned by the Mappers is large, the one-Reduce strategy adopted by HadoopDB becomes problematic and represents a bottleneck.

ORDER BY Queries

Table 9 shows our queries for the ORDER BY task. These queries retrieve objects and their detections (sources) ordered by ObjectID. Query Q 10 , considers a sky area bounded by RA and DECL, whereas query Q 11 considers a larger sky area bounded only by RA. Both HadoopDB and Hive generate only one MapReduce Job from these queries. In the Map step, Hive performs data READ, SELECTION, PROJECTION operations and a partial ORDER BY, whereas HadoopDB sends the same query to all local DBMSs. The number of Map tasks depends on the number of data blocs in the HDFS (e.g., for the dataset with 250 GB, we have 887 Map tasks). With HadoopDB, the number of tasks depends on the number of chunks created in the data loading stage. In the Reduce step, a global ORDER BY is performed by both systems using only one Reducer.

Due to their sizes, Hive can use indexes only for the datasets with 250 GB and 500 GB. For the 1 TB dataset, the size of the filtered part exceeds the size of heap memory used by the Mapper.

Figure 11 shows the evolution of execution time for HadoopDB and Hive over non indexed and indexed datasets. We observe that HadoopDB is significantly better than Hive with and without indexes.

With the index, a gain of 7X is reported for Hive and a gain of 11X is reported for HadoopDB. Furthermore, HadoopDB outperforms Hive with a factor that can exceed 2X. In this section, we first summarize the query evaluation mechanisms for SQL On MapReduce systems and the impact of different proposed optimization techniques on query evaluation. Then, we analyze the limits of these techniques with respect to LSST needs. We also make some proposals that could contribute to the design of a system able of meeting the LSST requirements in terms of data management.

In the LSST project, data is collected and exploited progressively. Indeed, an amount of 15 TB is expected to be collected each night. This data must be integrated with the already available one. In the sequel, we call such a data integration process: an integration iteration. Roughly speaking, to perform an integration iteration there is a need to process (index, compress, partition) the new data and integrate it with already available data.

We first start by analyzing the loading time. With respect to LSST data, both approaches proposed by Hive and HadoopDB are compatible with new data integration. However HadoopDB, is not efficient with respect to all kinds of queries, because when data is partitioned using a given attribute (e.g., objectID), in the next integration iteration, there is no guarantee that all the records with the same partitioning key are in the same local database. A partitioning is efficient with respect to new data but not for all the data stored in the database (old and new data). Therefore, it is questionable whether it is really worth to have a customized data partitioning strategy when data is collected and integrated progressively. The only method that enables to guarantee the optimality of query processing, is a redistribution (i.e., repartitioning) of all the available data each night. This solution is clearly not feasible due to induced time overhead: data loading time increases exponentially with respect to data volume. Indeed, more than 200-300% of the time is spent in partitioning data for a final result which is not necessarily very relevant.

Regarding indexing strategies, and with respect to data integration iteration, the strategy of HadoopDB remains very consistent. If we assume that partitioning does not trigger any particular problem, the data indexing strategy becomes even more interesting. Indeed, indexes are created at chunk level. When an index is created, the existing indexes are not impacted and hence there is no need to recalculate them. In contrast, the strategy of creating indexes in Hive is not compatible with integration constraints. Indeed, to ensure that an index will be used, it is necessary to recalculate it. Moreover, Hive is not stable in using indexes. The indexing mechanism of Hive should be revisited with respect to these constraints and extended with additional functionalities such as: multiple index (with multiple attributes), automatic selection of indexes, integration of indexes in additional operators (currently only the selection operator supports the use of indexes).

The index size is exponential with respect to the number of distinct values. We have to study new compressed structures. Indeed, 10% of space overhead for storing one index is not a realistic solution in particular in the case of tables with hundreds of attributes and which require several indexes. The size of indexes in HadoopDB is however very acceptable (3.2% of the overall size of a raw chunk). Moreover, HadoopDB supports multiple indexes, but the usage of indexes is however still limited to selection queries. The benefit of indexing in Hive is to provide a global view about data location with however a price to pay in terms of sizes of indexes and maintenance since there is a need to recalculate an index at each integration iteration step. Indexes proposed in HadoopDB do not require a large size and are calculated quickly which is compatible with integration iteration steps. Nevertheless, HadoopDB indexes do not provide a global view of the data.

We believe that a hierarchical indexing, with several levels, may offer a good compromise. Such an index structure can take advantages from the global view of data offered by Hive and the ease of use offered by HadoopDB.

Before comparing the execution time of the considered systems, we will first introduce some cost metrics which will be used to analyze the various query evaluation and optimization strategies of Hive and HadoopDB.

First of all, as mentioned previously, an SQL query is mapped to a set of MapReduce jobs. In each MapReduce job, we have five steps: (i) Data is read and transferred to Mappers as key/value pairs. We denote by CostRead Data, the cost of this step. (ii) A Map function is applied on the data read and key/value pairs received by the Mapper. We denote this cost by Cost Map. (iii) A network synchronization by sending key/value pairs generated by the mappers to Reducers. We denote by Cost network, the cost of this step. (iv) The reduce function is applied. We denote this cost by Cost reduce. (v) Partial results are written in the HDFS, these results will be, if necessary, used as input for the next job. We denote this cost by Cost WriteHDFS. These five steps are not required for all the queries. Indeed, for JOIN queries, we need all the steps, while for SELECTION queries only steps (i) and (ii) are required.

The overall cost of a job is obtained by summing these five costs. We denote this cost by Cost Job. Finally, we define the cost related to query processing as the sum of all costs of all jobs associated with the query. In the rest of this paper, we discuss the impact of each cost with respect to query types.

In the case of selection queries, and without considering indexes, only one job is generated both in the case of Hive and HadoopDB. In this case, the cost of a query is the same as the cost of its associated job. In the implementation of such a job, the two systems read data and applies the filtering associated with predicates in the query. Consequently the cost of such a job is: CostRead Data + Cost Map.

The CostRead Data, in HadoopDB, is given by the time necessary to read data from DBMS using JDBC. For Hive, this cost is related to chunks reading from the HDFS. As data in the DBMS is compressed, the CostRead Data related to HadoopDB job is less than CostRead Data related to Hive in most cases. The only different case is when CostRead Data, for HadoopDB, is dominated by JDBC latencies.

Both tools propose to use indexes to minimize CostRead Data. Indeed, in HadoopDB, indexes are used within the DBMS which allow to only access relevant disk pages by the DBMS. For Hive, we use an additional data structure that allows to locate only relevant HDFS chunks. In this vein, a filtering phase is used before applying the Map function. The Cost CostRead Data is minimized using the index. We have seen that when Hive decides to use an index, the result is better because Hive has a global view of the data which enables to locate relevant data more efficiently compared to HadoopDB.

The drawback of Hive is when the filter part has an important size. Indeed, when reading data, Hive extracts relevant chunks and sub-chunks, from index, that satisfy predicates in the query. If the filtered part cannot fit into the memory of Mappers, Hive is not able to use indexes. To conclude, we need a more sophisticated optimizer that can decide when a usage of an index is relevant and which index to use. In addition, Hive indexing mechanism must be extended to enable complex indexes with multiple attributes.

Another important point, for LSST project, is the choice of attributes to index. As some queries are known, it will be very interesting to have an index recommendation system for a large number of attributes.

With respect to GROUP BY queries, both Hive and HaddopDB read data, apply the filter related to predicates in a query and then apply partial GROUP BY, in the Map side. They send then the data to the Reducer(s) where a global GROUP BY is applied. Only one job is used, therefore the cost of a query is equivalent to the cost of the associated job. The cost of such a job is then: CostRead Data + Cost Map + Cost network + Cost reduce. For selective queries, only a few (<10) tuples are transferred from Mappers to Reducers. Therefore, CostRead Data and Cost Map dominate the Job cost. By compressing data HadoopDB outperforms Hive.

With respect to non-selective queries, the data transferred from Mappers to Reducers impacts the performance of the generated jobs. As HadoopDB uses one Reducer to perform the global GROUP BY and Hive uses several Reducers, we have observed that even when an index is used in HadoopDB, Hive outperforms HadoopDB for non-selective queries. Indeed, Cost reduce is shared between several machines, whereas for HadoopDB, due to the use of a single Reducer, this cost dominates the query processing cost and constitutes a bottleneck.

HadoopDB proposes to modify the partitioning attribute to deal with GROUP BY queries. Indeed, by changing the partitioning attribute, we are sure that most records having the same GROUP BY key are in the same node and hence there is no need to perform a global GROUP BY for the concerned records. In this case, the number of key/values transferred from Mappers to Reducers is minimized. We have observed a very significant gain in this case. The cost of network transfer and reduce phases are minimized using this technique. The drawback of this strategy comes from the fact that it requires to identify a priori the adequate partitioning attribute.

Determining a good partitioning strategy for LSST data is very challenging knowing that in LSST, several attributes are used as grouping attributes. It will be interesting to study the problem of partitioning data starting from a set of predefined queries. Another issue is related to the identification of the adequate number of Reducers to use for each query. In current state of affairs, Hive leaves this choice to the user. Extending such a system in order to be able to automatically determine the optimal number of Reducers to use during query execution is indeed an interesting research issue.

Regarding JOIN queries, Hive generates, for each JOIN, a MapReduce job.

HadoopDB does not support joins of more than two tables and do not allow multiple Reducers to perform the reduce function. Indeed, for JOIN queries and to the best of our knowledge, there is no optimization technique implemented in the open source release of HadoopDB. Also indexes are not used to perform joins. The cost of JOIN queries is obtained by the sum of costs of the associated jobs. Intermediate results are stored in the HDFS. In this case, the order of joining the relations has an important impact on query performances. To the best of our knowledge, no relevant techniques for finding the best order of joining tables is proposed so far. We believe that the optimizer of Hive should be revisited to incorporate sophisticated scheduling technique based on statistics. Indeed, the order of evaluation of joins will significantly impact query performance. In the TEZ system, Beta-version of Hadoop, MapReduce tasks can be pipelined without going through the store in the HDFS.

Regarding ORDER BY queries, both approaches propose to handle these queries in one job. HadoopDB outperforms Hive because HadoopDB takes advantages of compression and indexing, handled by the DBMS and minimizes the cost of CostRead Data. The ORDER BY in the Map phase is not mandatory because it can be performed by the DBMS when it reads the data. We believe that the MapReduce model is not suitable for handling JOIN and ORDER BY queries. We believe that this model is efficient for queries that need only one pass on the data (e.g., Selection and GROUP BY).

Another issue is the optimization of UDF (User Defined Functions). Indeed, UDF are very frequently used in astrophysics applications. Optimization within the MapReduce framework is not an easy task. Within Hive, we implemented some UDFs and we successfully executed those queries in a declarative manner. Hive offers several kinds of UDF: UDFs that can be evaluated in the Map phase, in the Reduce phase and as a separated job. For HadoopDB, the integration of UDFs is achieved manually as a separate MapReduce Job. HadoopDB lacks mechanisms for evaluating queries declaratively.

With respect to query frequency, in LSST we expect a half million of queries per day. At any time, we expect that the system should handle 50 simple queries (that need a few seconds to be evaluated) and 20 complex query (that could require hours or days to be evaluated). The versions of Hadoop used in our experiments are not able to support execution of multi-jobs. Actually, another version of Hadoop, called Yarn, provides this ability. Therefore, it will be interesting to investigate global execution and optimization of sets of queries in a MapReduce framework.

Conclusion

We designed and set up a benchmark for structured data from astronomy where a set of SQL queries are already defined. Our experiments targeted two SQL On MapReduce systems, Hive, a HDFS based system and HadoopDB, a Hadoop/Postgres based system. Our objective is to report on the capabilities of those systems to manage (e.g., storage, loading) LSST data on one hand and their capabilities (i.e., indexing, compression, buffering, and partitioning) to optimize some kinds of queries on the other hand. In contrast to existing experiments we did not consider only execution time but also loading time. We considered different configurations for the different parameters related to material, data, partitioning, indexing and selectivity. We reported on the impact of data partitioning, indexing and compression on query evaluation performances. We also highlighted the need for new techniques for query optimization in emerging systems.

There are many research directions that could be pursued to improve query processing in the case of LSST data. For example, it could be interesting to experiment another distribution model. Due to the cost of network communication between Mappers and Reducers, the BSP model [START_REF] Valiant | A bridging model for parallel computation[END_REF] could be an interesting alternative. We also believe that the extension of the presented analysis to other categories of systems (e.g., Hybrid In-Memory/Column-oriented DBMS) would be clearly an interesting future work. Another important issue is to automatically choose the parameters of the distribution framework (i.e., Hadoop), the partitioning attribute and the index to be created and used, from a workload. We are currently investigating these issues.

Fig. 1 :

 1 Fig. 1: Illustration of sample query processing using MapReduce

Fig. 2 :

 2 Fig. 2: Data loading: Hive vs. HadoopDB

Fig. 3 :

 3 Fig. 3: Hive: Data indexing

Fig. 4 :Fig. 5 :

 45 Fig. 4: Hive: Index size

( 8 *

 8 10 -3 %) (5 * 10 -3 %) (4 * 10 -4 %) (4 * 10 -3 %)

Fig. 8 :Fig. 9 :

 89 Fig. 8: GROUP BY tasks: Hive vs. HadoopDB

Fig. 10 :

 10 Fig. 10: JOIN tasks

Fig. 11 :

 11 Fig. 11: ORDER BY tasks

Table 1 :

 1 Table1ashows some sample data of the relation Source while table1bshows an extract of the expected results when the previous query is executed over table 1a. Examples The map outputs are then processed by a reduce function which implements an application specific business logic. A reduce function takes as input a key-list pair < k ; list(v ) >, where k is an intermediate key and list(v ) is the list of all the intermediate values associated with the key k , and returns as final result zero or more keyvalue pairs < k , v >. Several instantiations of the map and reduce functions can operate simultaneously on different nodes of the cluster. A Map's intermediate results are stored in the local file system and sorted by the keys. After all the Map tasks are completed, the MapReduce engine notifies the Reduce tasks (which are also processes that are referred to as Reducers) to start their execution. The Reducers pull the output files from the Map tasks in parallel and merge-sort the files obtained from the Map tasks to combine the key/value pairs into a set of new key/value pairs < k ; list(v ) >.

	SourceID	ObjectID	RA	DECL	ExposureID
	1	1	10	-3	1
	5	2	15	-4	1	ObjectID	freq
	9	2	20	-5	5	1	2
	13	4	25	-7	1	2	3
	2	1	40	-7	2	3	2
	6	2	45	-8	2	4	2
	10	3	50	-9	2	...	...
	14	4	55	-3	3
	...	...	...	...	...

(a) The Source table (b) Results of the query of content of the table Source and the result returned by the query (q)

2.2 MapReduce

MapReduce is a new programming model that is intended to be used to facilitate the development of scalable parallel computations on large server clusters

[START_REF] Dean | Mapreduce: Simplified data processing on large clusters[END_REF]

. A MapReduce framework enables to perform computations through two simple programming primitives: Map and Reduce functions. A Map function takes as input a data set in form of a set of key/value pairs and, for every pair < k, v > of the input returns zero or more intermediate key-value pairs < k ; v >.

  Reducer recipient. By default, a hash function is used. In the Shuffle and Sort phases, a transfer of local key, list < value > to Reducers is done. Finally, each Reducer applies the Reduce function on received pairs < key, List < value >>.

< ObjectID, 1 > by applying the Map function. Combiners perform a grouping on same ObjectIDs on < ObjectID, freq > leading to < ObjectID, list < f req >>. Partitioners are then used to choose a

Table 2 :

 2 Expected data characteristics

	Table	size	records	attributes
	Object	109 TB 38 Billions	470
	Source	3.6 PB	5 Trillions	125
	Moving Object	5 GB	6 Million	100
	Forced Source	1.1 PB 32 Trillions	7
	CCD Exposure 0.6 TB	17 Billions	45

Table 3 :

 3 1, PT 1.2, Winter 13 and SDSS Datasets characteristics We set up two clusters of 25 and 50 nodes, respectively. According to hdparm 5 , the hard disks deliver 113 MB/sec for buffered reads. Using this distributed architecture, we deployed HadoopDB and Hive. Hadoop, namely 0.19.1 and 1.1.1., in our experiments. Due to the incompatibility of HadoopDB with recent versions of Hadoop, HadoopDB is coupled with Hadoop 0.19.1, whereas Hive is coupled with Hadoop 1.1.1. Both versions run on Java 1.7. We deployed the system by using the configuration mentioned in HadoopDB original paper. 6 Data in HDFS (Hadoop Distributed File System) is stored using 256 MB data blocks instead of the default 64 MB. Each MapReduce job executor runs with a maximum heap size of 1024 MB.

				Table	Attributes	Records	Size
	Dataset PT 1.1 PT 1.2 Winter 13 SDSS	Size 90 GB 220 GB 3 TB 3 TB	Source Object RefSrcMatch Science Ccd Exposure Metadata	107 227 10 6	162 x 10 6 118 GB 4 x 10 6 7 GB 189 x 10 6 1.7 GB 41 x 10 6 16 GB
	(a) Datasets			(b) PT 1.2 characteristics
	Datasets	Source:	Source:	Object:	Object:	SourceID	ObjectID	DECL	RA	ScienceCcd
		Record	Size (GB)	Record	Size (GB)						ExposureID
	D250GB 325 x 10 6	236	9 x 10 6	14	325 x 10 6	9 x 10 6	325 x 10 6 162 x	84 x 10 3
	D500GB 650 x 10 6	472	18 x 10 6	28	650 x 10 6 18 x 10 6 650 x 10 6 162 x	84 x 10 3
	D1TB	1.3 x 10 9	944	36 x 10 6	56	1.3 x 10 9 36 x 10 6 1.1 x 10 9 162 x	84 x 10 3
	D2TB	2.6 x 10 9	1888	72 x 10 6	112	2.6 x 10 9 72 x 10 6 2.3 x 10 9 162 x	84 x 10 3
					(c) Generated dataset

(Sloan Digital Sky Survey, http://www.sdss.org/). Only SDSS dataset is a real dataset collected within the SDSS project. We chose to use for our experiments the dataset PT1.2 whose characteristics are depicted at Table

3b

. This dataset contains 22 tables stored as "CSV" files with a total size of 220 GB. At this stage, we exploited only 4 tables: The Source table that has 107 attributes and contains 180 millions of tuples (for a total size of 139 GB), the Object table which has 229 attributes and contains 5 millions of tuples (for a total size of 7 GB), the Science Ccd Exposure Metadata table which has 6 attributes and contains 41 millions of tuples (for a total size of 16 GB), and the RefSrcMatch table that has 10 attributes and contains 189 millions of tuples (for a total size of 1.5 GB). To address the scalability issue, we generated 4 , from PT 1.2 dataset, four (4) new datasets with a total size of 250 GB, 500 GB, 1 TB and 2 TB respectively. New tables are generated only for the tables Source and Object. For the other tables, we kept the same records. Table

3c

shows the characteristics of the generated data. The four first columns show the number of records and the size of the tables Source and Object respectively. The other columns show the number of distinct values for five attributes of the table Source involved in our workload.

4.3 Hive

Hive is based on Hadoop open Source that implements a MapReduce

[START_REF] Dean | Mapreduce: Simplified data processing on large clusters[END_REF]

framework.

Table 4 :

 4 Selectivity

	Predicate	D250GB	D500GB	D1TB	D2TB
	SourceID =id	1	1	1	1
		(3.07 * 10 -7 %) (1.7 * 10 -7 %)	(7.69 * 10 -8 %) (3.84 * 10 -8 %)
	ObjectID =id	43	43	43	43
		(1.32 * 10 -2 %) (7.67 * 10 -6 %) (3.3 * 10 -6 %)	(1.65 * 10 -6 %)
	2 < DECL < 2.05	1.6 x 10 6	3.3 x 10 6	6.6 x 10 6	13.2 x 10 6
		(0.49%)	(0.28%)	(0.5%)	(0.5%)
	359.959 < ra < 359.96	14 x 10 3	28 x 10 3	57 x 10 3	127 x 10 3
		(4 * 10 -3 %)	(2.5 * 10 -3 %)	(4 * 10 -3 %)	(4 * 10 -3 %)
	359.959 < ra < 359.96	21	43	86	172
		(6.46 * 10 -6 %) (3.5 * 10 -6 %)	(6.6 * 10 -6 %)	(3.3 * 10 -6 %)
	& 2 < decl < 2.05
	ScienceCcdExposureId = id 3.6 x 10 3	7.3 x 10 3	14.6 x 10 3	29.2 x 10 3
		(1.1 * 10 -3 %)	(6.4 * 10 -3 %)	(1.1 * 10 -3 %)	(1.1 * 10 -3 %)

of queries: For each query and for each data volume, the first line stands for the number of results and the second line stands for the percentage of results w.r.t the total number of records contained in the database Fig. 6: SELECTION tasks without index

Table 5 :

 5 Selectivity of predicates: For each predicates and for each data volume, the first line stands for the number of records satisfying the predicate and the second line stands for the those records w.r.t the total number of records contained in the database SELECT Sourceid, Objectid FROM Source WHERE ra > 359.959 and ra < 359.96 and decl < 2.05 and decl > 2;

	Query SQL syntax
	Q 1	SELECT	*	FROM	Source	WHERE
		Sourceid=29785473054213321;
	Q 2	SELECT Sourceid, ra,decl FROM Source WHERE
		Objectid=402386896042823;
	Q 3
	Q 4	SELECT Sourceid, ra,decl FROM Source WHERE
		scienceccdexposureid=454490250461;

Table 6 :

 6 SELECTION queries

  SELECT Objectid,count(sourceid) FROM Source WHERE ra > 359.959 and ra < 359.96 and decl < 2.05 and decl > 2 GROUP BY Objectid;

	Query SQL syntax
	Q 5
	Q 6	SELECT Objectid,count(sourceid) FROM Source GROUP BY
		Objectid;

Table 7 :

 7 Aggregation queries have 236 Reduce tasks obtained from the size of the Source table (i.e., 236 GB) and the size of data handled by a Reducer (i.e., 1 GB).

  WHERE s.ra > 359.959 and s.ra < 359.96 and s.decl < 2.05 and s.decl > 2 and s.filterId = 2 and rsm.refObjectId is not NULL;

					FROM	Source	JOIN	Object	on
		(source.objectid=object.objectid)	WHERE	ra	>
		359.959 and ra < 359.96 and decl < 2.05 and decl > 2;
	Q 8	SELECT	*	FROM	Source	JOIN	Object	on
		(source.objectid=object.objectid)	WHERE	ra	>
		359.959 and ra < 359.96;
	Q 9	SELECT s.psfFlux, s.psfFluxSigma, sce.exposureType
		FROM		Source	s	JOIN		RefSrcMatch	rsm
		ON	(s.sourceId		=	rsm.sourceId)	JOIN
		Science Ccd Exposure Metadata		sce	ON
		(s.scienceCcdExposureId = sce.scienceCcdExposureId)

Table 8 :

 8 JOIN queries

  SELECT Objectid,sourceid FROM Source WHERE ra > 359.959 and ra < 359.96 and decl < 2.05 and decl > 2 ORDER BY Objectid; 11 SELECT Objectid,sourceid FROM Source WHERE ra > 359.959 and ra < 359.96 ORDER BY Objectid;

Table 9 :

 9 ORDER BY queries

The query number 019 available at https://dev.lsstcorp.org/trac/wiki/db/queries/019.

http://www.astronomycafe.net/qadir/q1890.html

generated data comply with the characteristics of real data, we used an algorithm (https: //dev.lsstcorp.org/trac/wiki/db/Qserv/Duplication) designed by LSST data management group.

A tool that gives the average disc speeds (http://en.wikipedia.org/wiki/Hdparm).

the same configuration is used in[START_REF] Pavlo | A comparison of approaches to large-scale data analysis[END_REF], the popular Hadoop benchmark paper.

Called Catalog.

In the rest of paper, we use the notation nX of data to denote the size n times.

In the rest of paper, we use the notation nX of time to denote the time n times.

the number of distinct values is the same for all the datasets.

The number of distinct values double if we double the size of the dataset.

In the rest of paper, we use the notation nX machines to denote the number of machines n times.

http://ceph.com/

Additional results can be found in the project WebSite http://com.isima.fr/Petasky/ hive-vs-hadoopdb

This work is partially supported by Centre National de la Recherche Scientifique-CNRS, under the project Petasky-Mastodons (http://com.isima.fr/Petasky) A. Mesmoudi and M-S Hacid


Eric Chauvet

Anne-Marie Jean-Louis

Production de litière de la ripisylve de la Garonne et apport au fleuve

and research institutions in France or abroad, or from public or private research centers.

To cite this version : Chauvet, Eric and Jean-Louis, Anne-Marie Production de litière de la ripisylve de la Garonne et apport au fleuve. (1988) Acta Oecologica, Oecologia Generalis, vol. 9 (n° 3). pp. 265-279. ISSN 0243-766X Open Archive TOULOUSE Archive Ouverte (OATAO)

OATAO is an open access repository that collects the work of Toulouse researchers and makes it freely available over the web where possible. This is an author-deposited version published in : http://oatao.univ-toulouse.fr/ Eprints ID : 14392 Any correspondance concerning this service should be sent to the repository administrator: staff-oatao@listes-diff.inp-toulouse.fr