Multi-Objective Big Data Optimization with jMetal and Spark

Multi-Objective Big Data Optimization with jMetal and Spark Crist´obal Barba-Gonzal´ez, Jos´e Garc´ıa-Nieto, Antonio J. Nebro?, and Jos´e F. Aldana-Montes Dept. de Lenguajes y Ciencias de la Computaci´on, University of Malaga, ETSI Inform´atica, Campus de Teatinos, Malaga - 29071, Spain [email protected], [email protected], [email protected], [email protected] Abstract. Big Data Optimization is the term used to refer to optimization problems which have to manage very large amounts of data. In this paper, we focus on the parallelization of metaheuristics with the Apache Spark cluster computing system for solving multi-objective Big Data Optimization problems. Our purpose is to study the influence of accessing data stored in the Hadoop File System (HDFS) in each evaluation step of a metaheuristic and to provide a software tool to solve these kinds of problems. This tool combines the jMetal multi-objective optimization framework with Apache Spark. We have carried out experiments to measure the performance of the proposed parallel infrastructure in an environment based on virtual machines in a local cluster comprising up to 100 cores. We obtained interesting results for computational effort and propose guidelines to face multi-objective Big Data Optimization problems. Keywords: Multi-objective Optimization, Big Data, jMetal, Spark, Parallel Computing. 1 Introduction Over the past few years, Big Data technologies have attracted more and more attention, leading to an upsurge in research, industry and government applications. There are multiple opportunities and challenges in Big Data research. One area in particular where Big Data is promising is Global Optimization [26]. The issue is that Big Data optimization problems may need to access a massive amount of data to be solved, which introduces a new dimension of complexity apart from features such as non-linearity, uncertainty and conflicting objectives. Focusing on multi-objective optimization, metaheuristic search methods, such as evolutionary algorithms, have been widely applied to a great number of academic and industry optimization problems [6]. Depending on the problem, a metaheuristic may need to perform thousands or even millions of solution evaluations. In the case of complex system optimization, the computational effort, in ?Corresponding author an[email protected] terms of time consumption and resource requirements, to evaluate the quality of solutions can make it impracticable to apply current optimization strategies to such problems. This issue is even harder when dealing with Big Data environments, where a huge volume of data hast to be accurately and quickly managed. One strategy to cope with these difficulties is to apply parallelism [17]. In the last few years, a number of approaches consisting in adapting metaheuristic techniques to work in parallel on Hadoop, the de facto Big Data software platform, have been proposed. These proposals are related to data mining or data management applications, such as: feature selection [1], data partitioning [12], dimension reduction [23], pattern detection [5], graph inference [16], and task scheduling [22]. Most of these approaches are based on the MapReduce programming model [15]. However, MapReduce entails a series of drawbacks that make it unsuitable to be properly integrated with metaheuristics in particular and with global optimization techniques in general. Chief among them are: high latency queries, non-iterative programming model, and weak real-time processing. Therefore, there is a demand for new challenging approaches to integrate Big Data based technologies with global optimization algorithms in order to cope with all these issues. In this paper, our approach is to address the parallelization of metaheuristics in Hadoop-based systems with Apache Spark [25], which is defined as a fast and general engine for large-scale data processing. Our proposal is to use the jMetalSP framework1, which combines Spark with the jMetal multi-objective optimization framework [11]2. Concretely, we have included support in jMetalSP to parallelize metaheuristics with Spark in an almost transparent way, hence avoiding the intrinsic shortcomings of the usual MapReduce model when applied to global optimization. We aim to consider two scenarios: first, to use Spark as an engine to evaluate the solutions of a metaheuristic in parallel, and, second to study the influence of accessing a massive amount of data in each evaluation of a metaheuristic algorithm. Instead of focusing on a particular optimization problem, we have defined a generic scenario, in which a benchmark problem is modified to artificially increase its computing time and to read data from the Hadoop file system (HDFS). We have carried out a number of experiments to measure the performance of the proposed method in a parallel virtualization infrastructure of multiple machines in an in-house cluster. The main contributions of this paper are as follows: –We provide a software solution for parallelizing multi-objective metaheuristics included in the jMetal framework to take advantage of the high performance cluster computing facilities provided by Spark. This way, developers and practitioners are provided with an attractive tool for Big Data Optimization. 1In URL https://github.com/jMetal/jMetalSP 2In URL http://jmetal.github.io/jMetal/ 2 –We perform a thorough experimentation of our proposal from three different viewpoints. First, by measuring the performance of algorithms in terms of computational effort in a Hadoop parallel environment; second, by analyzing the influence of accessing data stored in HDFS in each evaluation of a metaheuristic; and third, by combining both approaches. To carry out this study, we define different data access/processing tasks to be done when evaluation a solution, so we can measure the performance of the algorithms according to computational effort and size of data. This allows us to compute the speedups that can be obtained and identify the system’s limits, to determine whether or not it is worth using more resources to solve the problem. The remainder of this article is organized as follows. The next section presents an overview of the related work in the literature. Section 3 details our Big Data optimization approach. In Section 4, the experimental framework and parameters settings are described. Section 5 details the experimental results and analyses. Finally, Section 6 outlines some concluding remarks and plans for future work. 2 Related Work Large amounts of data and high dimensionality characterizes many optimization problems in interdisciplinary domains such as, biomedical sciences, engineering, finance, and social sciences. This means that optimization problems handling such spatio-temporal restrictions often deal with tens of thousands of variables or features extracted from documents, images and other objects. To tackle such challenging problems, a series of proposals have appeared in the last decade, which combine metaheuristics with data mining or data management applications, and adapt them to perform in Hadoop environments. Concretely, in [1] a swarm intelligence approach is adapted to optimize the features that exist in large protein sequences using a two-tier hybrid model by applying both filter and wrapper methods. A similar approach is proposed in [12], where a particle swarm optimization algorithm (PSO) is used to discover clusters in data that are continuously captured from students’ learning interactions. In [2], intrusion detection is managed with a MapReduce strategy based on a PSO clustering algorithm. An interesting method has been reported in [23], in which, by using sensor data to generate a PCA (Principal Component Analysis) model to forecast photovoltaic energy, it is possible to reduce the dimensionality of data with the collaboration of other artificial intelligence techniques, such as: fuzzy interference, neural networks, and genetic algorithms. In the case of biomedical sciences, the reconstruction of gene regulatory networks is a complex optimization problem that is attracting particularly special from the research community, since it is considered to be a potential Big Data problem in the specialized literature [26,3]. Along the same lines, in the study carried out in [16], the authors proposed a parallel method consisting in a hybrid genetic algorithm with PSO by means of the MapReduce programming model. The resulting approach was tested in different cloud computing environments. 3 Most of these approaches are based on the MapReduce (MR) programming model [15], which yields a competitive performance in comparison with other parallel (and sequential) models for the specific problems tackled. In addition, other algorithmic adaptations to MapReduce operations can be found in metaheuristics, such as: PSO [18], Differential Evolution [9] and Ant Colony Optimization [24]. However, as stated, the MapReduce model entails a series of drawbacks that make it unsuitable to be integrated with metaheuristics in particular and with global optimization techniques in general. These are directly related to: high-latency queries, a non-iterative programming model, and weak real-time processing. For instance, the following issues which we aim to cover with our proposal, combining jMetal and Spark: –MapReduce uses coarse-grained tasks to do its work, which can be too heavyweight for iterative algorithms, like metaheuristic algorithms. In the proposals analyzed, developers use various MapReduce hacks or alternative tools to overcome these limitations, but this highlights the need for a better computation engine that supports these algorithms directly, while continuing to support more traditional batch processing of large datasets. Our software proposal follows an iterative programming model, which eases the adaptation of algorithms and the integration with software classes managing a multitude of optimization problems. –Another problem with current optimization algorithms using MapReduce is that they have no awareness of the total pipeline of Map plus Reduce steps, so they cannot cache intermediate data in memory for faster performance. Instead, they flush intermediate data to disk between each step. With Spark the managed data can be cached in memory explicitly, thus improving performance significantly. –Existing proposals in the literature were not evaluated on well-grounded Big Data environments. Most of them were tested to show their ability to solve a given optimization problem in a parallel infrastructure composed of up to ten machines, and therefore critical aspects such as data volume and variable computational effort remain open issues. In the present study, a thorough experimentation is carried out to measure the performance of algorithms in terms of scalability in a 100-core Hadoop-based cluster, and to analyze the influence of accessing a large amount of data in each evaluation of a multi-objective metaheuristic. 3 Big Data Optimization Approach To develop Big Data optimization applications it is necessary to have software tools capable of coping with the requirements of such applications. Our contribution in this sense is to propose jMetalSP [4][7], an open-source platform3 combining the jMetal optimization framework [11] with the Apache Spark cluster computing system [25]. Below we describe the adopted Big Data Optimization scheme in the context of jMetal, Spark, and jMetalSP. 3URL: https://github.com/jMetal/jMetalSP 4 Fig. 1. Class diagram with core classes and interfaces of jMetal 5. Algorithm 1 Template of a metaheuristic 1: A(0) ←GenerateInitialSolutions() 2: t←0 3: Evaluate(A(0)) 4: while not StoppingCriterion( ) do 5: S(t)←Generation(A(t)) 6: Evaluate(S(t)) 7: A(t+ 1) ←Update(A(t), S(t)) 8: t←t+ 1 9: end while jMetal is an algorithmic framework, which includes a number of optimization metaheuristics of the state of the art [11]. It mostly centers in multi-objective optimization, although it also provides single-objective algorithms. In the work presented here, we use jMetal 5 [19], which follows the architecture depicted in Fig. 1. The underlying idea is that an algorithm (metaheuristic) manipulates a number of solutions with some operators to solve an optimization problem. jMetal 5 provides algorithm templates mimicking the pseudo-code of a generic metaheuristic like that shown in Algorithm 1, where a set Aof some initial solutions is iteratively updated by generating a set Sof new solutions until a stopping condition is achieved. Another feature of jMetal is that it offers an interface (called SolutionListEvaluator) to encapsulate the evaluation of a list of solutions (i.e., a population in the context of evolutionary algorithms): public interface SolutionListEvaluator<S> { List<S> evaluate(List<S> solutionList, Problem<S> problem); void shutdown(); } 5 The encapsulated behavior of this interface is that the method evaluate of the problem (see Fig. 1) is applied to all the solutions in the list, yielding to a new list of evaluated solutions. Metaheuristics using this interface can be empowered with different evaluator implementations, so that the current way of evaluating solutions is transparent to the algorithms. For example, many algorithms incorporate a method similar to this one (which corresponds to step 6 of the template shown in Algorithm 1): protected List<S> evaluatePopulation(List<S> population) { population = evaluator.evaluate(population, problem); return population; } In this way, the actual evaluator is instantiated when configuring the settings of the metaheuristic, so no changes in the code are needed. jMetal 5 currently includes two implementations of SolutionListEvaluator: sequential and multithreaded. Our approach has been then to develop an evaluator based on Spark. Apache Spark [25] is based on the concept of Resilient Distributed Datasets (RDD), which are collections of elements that can be operated in parallel on the nodes of a cluster by using two types of operations: transformations (e.g., map, filter, union, etc.) and actions (e.g., reduce, collect, and count). The Spark based evaluator in jMetal creates an RDD with all the solutions to be evaluated, and a map transformation is used to evaluate each solution. The evaluated solutions are then collected and returned to the algorithm. It is worth noting that algorithms do not need to be modified to use Spark, although the problem to be solved must fulfill the requirements imposed by this platform, as the algorithms run map processes. For example, the evaluate method of the problem must not modify variables outside the scope of the RDD containing the list of solutions to be evaluated. Currently, five multi-objective metaheuristics in jMetal 5 use evaluators, so all of them can take advantage of the one based on Spark: NSGA-II [10], SPEA2 [27], SMPSO [20], GDE3 [14] and PESA2 [8]. This scheme can be also used in a number of single-objective algorithms: generational genetic algorithm (gGA), differential evolution (DE), and two PSO algorithms. jMetalSP is a new project for Big Data Optimization with multi-objective metaheuristics [4] based on jMetal and Spark. It is currently intended to solve dynamic multi-objective optimization problems in Hadoop environments by using the streaming data processing capabilities of Spark, while jMetal provides the optimization infrastructure for implementing the dynamic problems and the dynamic algorithms to solve them. We have extended jMetalSP with the Spark evaluator, so it can be used also to solve non-dynamic optimization problems. The attractive point of the adopted approach is that a number of single and multi-objective metaheuristics can be executed in parallel without requiring any modification. If we look closely at steps 3 and 6 in Algorithm 1 we can see that the set Scan be evaluated in parallel by the Spark-based evaluator, so the resulting parallel model is a heartbeat algorithm: a parallel step is alternated with a sequential one (for the rest of the phases in the main loop of the metaheuristic). 6 Fig. 2. Computational environment for Big Data Optimization used to test the performance of the proposed jMetal+Spark software solution, under different conditions of computational effort and Big Data management This obviously prevents linear speedups, and this is the price to pay for having a very simple mechanism that uses jMetal’s algorithms in a Big Data infrastructure. In this paper, our main interest is to measure the effective time reductions that can be achieved in different contexts. 4 Experimental Framework To evaluate the performance of the proposed approach, a series of experiments have been conducted from three points of view: (1) computational effort, in terms of which we measure the performance of the parallel model; (2) data management, which is focused on testing the ability to manage a large number of data files; and (3) a combination of (1) and (2). Therefore, for each, we follow a different problem configuration involving: time consuming delays, different data block sizes, and different cluster sizes. We are seeking the maximum limit that a multi-objective algorithm can manage when handling Big Data without a loss in performance. For this reason, with the aim of guiding us and to simplify this experimentation, we have centered on one optimization problem and one algorithm to solve it, although without emphasizing on the solution quality, as the behavior of the parallel algorithm is the same as its sequential counterpart. Specifically, we have selected the NSGA-II algorithm [10] and the multi-objective optimization problem ZDT1 [13]. From an algorithmic point of view, we have used a common parameter setting of NSGA-II to test our proposal in all the experiments. The variation operators are SBX crossover and Polynomial mutation, with crossover and mutation rates pc= 0.9 and pc= 1.0/L, respectively (Lbeing the number of decision variables of the problem), and both having a distribution index value of 20. The selection 7 strategy is binary tournament and the population size has been set to 100 individuals. Finally, the stopping condition is met when the total number of 25,000 candidate solutions have been evaluated. In others words, the NSGA-II performs 250 evolution steps or iterations of the population throughout the running time. All the experiments have been conducted in a virtualization environment running on a private high-performance cluster computing platform. This infrastructure is located at the Ada Byron Research Center at the University of M´alaga (Spain), and comprises a number of IBM hosting racks for storage, units of virtualization, server compounds and backup services. Our virtualization platform is hosted in this computational environment, whose main components are illustrated in Fig. 2. Concretely, this platform is made up of 10 virtual machines (VM1 to VM10), each one with 10 cores, 10 GB RAM and 250 GB virtual storage (adding up to 100 cores, 100 GBs of memory and 2.5 TB HD storage). These virtual machines are used as Slave nodes with the role of TaskTracker (Spark) and DataNode (HDFS) to perform fitness evaluations of algorithmic candidate solutions in parallel. The Master node, which runs the core algorithm (NSGA-II), is hosted in a different machine (VM0) with 8 Intel Core i7 processors at 3.40 GHz, 32 GB RAM and 3 TB storage space. All these nodes are configured with a Linux CentOS 6.6 64-bit distribution. The whole cluster is managed with Apache Ambari 1.6.1 and executes the Apache Hadoop version 2.4.0. This Hadoop distribution integrates HDFS and Apache Spark 1.6. The jMetalSP framework is then deployed on this infrastructure, providing optimization algorithms with Spark methods to evaluate candidate solutions in parallel, in addition to managing HDFS files. 5 Experiments This section describes the set of experiments and the analyses carried out to test our approach. We focus on the speedup and efficiency analysis in terms of computational effort and data management. 5.1 Speedup and Efficiency One of the most widely used indicators for measuring the performance of a parallel algorithm is the Speedup (SN). The standard formula of the speedup is represented in Equation 1 and calculates the ratio of T1over TN, where T1is the running time of the analyzed algorithm in 1 processor and TNis the running time of the parallelized algorithm on Nprocessing units (processors or cores). SN=T1 TN (1) EN=SN N×100 (2) 8 Table 1. Experimental results of NSGA-II (jMetalSP) executed on 1, 10, 20, 50, and 100 cores with different time delays in each problem evaluation Running Time (hours) Speedup Efficiency Delay T1T10 T20 T50 T100 S10 S20 S50 S100 E10 E20 E50 E100 10 s 76.50 10.90 8.10 5.00 3.50 7.02 9.44 15.30 21.85 70.20% 47.22% 30.60% 21.85% 30 s 216.70 34.10 24.40 16.30 10.80 6.35 8.88 13.29 20.06 63.50% 44.41% 26.59% 20.06% A related measure is the Efficiency of a parallel algorithm, which is calculated with the formula of Equation 2. An algorithm scales linearly (ideal) when it reaches a speedup SN=Nand hence, the parallel efficiency is EN= 100%. 5.2 Computational Effort As stated, to measure the parallel computing performance of our approach we have used the NSGA-II algorithm to solve a modified version of the ZDT1 problem. The running time of NSGA-II with those settings in a laptop equipped with an Intel i7 processor is less than a second, so we have artificially increased the computing time of the evaluation functions of ZDT1 (by adding an idle loop) to simulate a real scenario where the total computing time would be in the order of several hours. After a number of preliminary experiments, we set the evaluation time to two values, 10 and 30 seconds; this way, we estimated the total running time of the sequential NSGA-II algorithm to be around 76.5 and 216.7 hours, respectively. Table 1 shows the running time in hours used by the NSGA-II approach of jMetalSP running on 1, 10, 20, 50, and 100 cores, with regards to the two delays considered, applied in each solution evaluation. This table also contains the corresponding speedup and efficiency values to the resulting times. As mentioned, T1(one core) is 76.5 hours or 3.19 days in the case of the 10 seconds delay and 216.7 hours or 9.03 days with a delay of 30 seconds. As expected, these times are reduced in relation to the increase in the number of cores used in the parallel model. The highest reductions in time are obtained when our approach is configured with 100 cores in parallel, for which the running time is reduced to 3.5 hours (95.42%) in the case of a delay of 10 seconds and to 10.8 hours (95.02%) using a 30 seconds delay. In terms of efficiency, the highest percentage 70.2% is reached with 10 cores and it decreases as the number of resources gets larger, to reach 47.22%, 30.6%, and 21.85% with 20, 50, and 100 cores, respectively. This behavior was somewhat expected as the parallel model is based on alternating parallel and sequential steps. Considering the results, it is worth mentioning that both problem configurations yield similar speedup and efficiency values, which indicates that the bottleneck is due to both the parallel model and the parallel infrastructure, so increasing the evaluation time does not compensate the synchronization and communication costs. 9