HybSMRP: a hybrid scheduling algorithm in Hadoop MapReduce framework

Distributed and parallel processing is one of the best intelligent ways to store and compute big data [1]. Most definitions defined big data as characterized by the 3Vs: the extreme volume of data, the wide variety of data types and the velocity at which the data must be processed. MapReduce [2] is a programming model for big data processing. MapReduce programs are intrinsically parallel [3, 4]. MapReduce executes the programs in two phases, map and reduce, so that each phase is defined by a function called mapper and reducer. A MapReduce framework consists of a master and multiple slaves. The master is responsible for the management of the framework, including user interaction, job queue organization and task scheduling. Each slave has a fixed number of map and reduce slots to perform tasks. The job scheduler located in the master assigns tasks according to the number of free task slots reported by each slave through a heartbeat signal [5]. Each job is split into a large number of map and reduce tasks before being started. The runtime is in charge of running tasks for every job until they are completed. The tasks are actually executed in any of the slave nodes which comprise the MapReduce cluster. In particular, the task scheduler is responsible for deciding what tasks are run at each moment in time, as well as what slave node will host the task execution [6]. A number of scheduling algorithms are available which help Hadoop [7] to improve its performance in different factors such as Data locality rate and job completion time. Each of the traditional algorithms improves the performance regarding certain factors. In Hadoop, data locality is the process of moving the computation close to where the actual data resides on the node, instead of moving large data to computation. This minimizes network congestion and increases the overall throughput of the system.

Hadoop provides three built-in scheduling modules [7]: First In First Out (FIFO) scheduler, Fair scheduler, and Capacity scheduler. These schedulers have different performances such as execution time and waiting time in different situations. Limitations of Hadoop FIFO occur when short jobs have to wait too long behind long running jobs, thus negatively affecting the job response time. The Hadoop Fair scheduler, developed by Zaharia et al. [8], was the first to addresses this limitation in depth through a fair share mechanism between multiple concurrent jobs. Over time, fair scheduler assigns resources such that all jobs get, on average, an equal share of resources. Additionally, fair scheduler extends the data locality of FIFO using a delayed execution mechanism. The most negative point of this algorithm is that it does not consider the job priority of each node. There are many Hadoop MapReduce scheduling algorithms which focus on MapReduce scheduling issues, some of which use only one technique to achieve high data locality rate while some others use dynamic job priority to decrease response and completion time. For example, matchmaking algorithm [9] uses a locality ID to increase the data locality rate, while HybS [10] uses a dynamic job priority. Due to the fact that many of the existing MapReduce scheduling algorithms do not use these techniques together, the current paper has proposed a new Hybrid Scheduling Algorithm, called HybSMRP (Hybrid Scheduling MapReduce priority), which uses a data localization technique and Dynamic job priority to increase data locality rate and decrease completion time.

The rest of this paper is organized as follows. In the section on research Background, an overview of Hadoop MapReduce is described. The section entitled Related Works provides a review of the related works. It describes existing approaches and introduces the background of Hadoop MapReduce scheduling algorithms. The section on Proposed Method presents the scheduling algorithm introduced in this work. Evaluation setting and results are given in the Results and Discussion Section. Finally, the section on Conclusion and Future Work concludes the paper with remarks on the limitations and possible future works.

In this section, firstly an overview of Hadoop MapReduce is presented and, then, MapReduce scheduling issues and Hadoop default scheduling algorithms are introduced.

There are many scheduling algorithms which address the main issues of MapReduce scheduling with different techniques and approaches. As it has already been mentioned, some of these algorithms have been focused on improving data locality and some aim to provide synchronization processing. Also, many of these algorithms have been designed to decrease the completion time. LATE [16], SAMR [17], CREST [18], LARTS [19], Maestro [20] and Matchmaking algorithms have focused on data locality. What follows is a brief description of some of the most important algorithms: in Longest Approximate Time to End (LATE) scheduler, backup tasks are used for the tasks that have longer remaining execution time. LATE uses a set of fixed weights to estimate the remaining execution time. This scheduler tries to identify the slow running tasks, and once identified, sends them to another node for execution. If this node is able to perform the task faster, then the system performance increases. The advantage of this method is the calculation of the remaining execution time of the task, together with the calculation of the rate of job progress, which leads to an increase in the rate of system response. In contrast, one of the disadvantages of LATE is that the task selection for re-execution is carried out incorrectly in some cases, which is due to the wrong calculation of the remaining execution time of the task. As a result, Chen et al. [17] recommended Self-Adaptive MapReduce (SAMR) scheduling algorithm, inspired by LATE scheduling algorithm. In this algorithm, the history of job executions is used to calculate the remaining execution time more accurately. The task tracker reads the history information and adjusts the parameters using this information. The SAMR scheduling algorithm improves MapReduce by saving execution time and system resources. It defined fast nodes and slow nodes to be nodes, which can finish a task in a shorter time and longer time than most other nodes. It will focus on how to account for data locality when launching backup tasks, because data locality may remarkably accelerate the data load and store. Lei et al. [18] proposed a novel approach, CREST (Combination Re-Execution Scheduling Technology), which can achieve the optimal running time for speculative map tasks and decrease the response time of MapReduce jobs. To mitigate the impact of straggler tasks, it is common to run a speculative copy of the straggler task. The main idea is that re-executing a combination of tasks on a group of computing nodes may progress faster than directly speculating the straggler task on target node, due to data locality. Hammoud and Sakr [19] presented another approach discussing the data locality problem. It deals specifically with reduce tasks. The scheduler, named Locality-Aware Reduce Task Scheduler (LARTS), uses a practical strategy that leverages network locations and sizes of partitions to exploit data locality. LARTS attempts to schedule reducers as close as possible to their maximum amount of input data and conservatively switches to a relaxation strategy seeking a balance among scheduling delay, scheduling skew, system utilization, and parallelism. Ibrahim et al. [20] developed a scheduling algorithm called Maestro to avoid the non-local map tasks execution problem that relies on replica aware execution of map tasks. To address this, Maestro keeps track of the chunks and replica locations, along with the number of other chunks hosted by each node. In this way, Maestro was able to schedule map tasks with low impact on other nodes local map tasks execution by calculating the probabilities of executing all the hosted chunks locally. Maestro, with the use of these two waves, provided a higher locality in the execution of map tasks and a more balanced intermediate data distribution for the shuffling phase.

He et al. [9] proposed a matchmaking-scheduling algorithm, which is close to our HybSMRP. Local map tasks are always preferred over non-local map tasks, no matter which job a task belongs to, and a locality marker is used to mark nodes and to ensure each node has a fair chance to grab its local tasks.

Another algorithm, the one closest to our proposed algorithm, is HybS, which was proposed by Nguyen et al. [10]. HybS is based on dynamic priority, in order to reduce the delay for variable length concurrent jobs, and relax the order of jobs to maintain data locality. Also, it provides a user-defined service level value for quality of service. This algorithm is designed for data-intensive workloads and tries to maintain data locality during job execution.

Wang et al. [21] proposed a map task scheduling algorithm in MapReduce with data locality, which is derived an outer bound on the capacity region and a lower bound on the expected number of backlogged tasks in steady state. Their focus is to strike the right balance between data-locality and load-balancing to simultaneously maximize throughput and minimize delay. Naik et al. [22] proposed a novel data locality based scheduling algorithm which enhances the MapReduce framework performance in heterogeneous Hadoop cluster. The proposed scheduler dynamically divides the input data and assigns the data blocks according to the node processing capacity. It also schedules the map and reduce tasks according to the processing capacity of nodes in the heterogeneous Hadoop cluster. Liu et al. [23] proposed that the available resources in the Hadoop MapReduce cluster are partitioned among multiple organizations who collectively fund the cluster based on computing needs. MapReduce adopts a job-level scheduling policy for a balanced distribution of tasks and effective utilization of resources. The current paper introduces a two-level query scheduling which can maximize the intra-query job-level concurrency, and at the same time speed up the query-level completion time based on the accurate prediction and queuing of queries.

Tran et al. [24] proposed a new data layout scheme that can be implemented for HDFS. The proposed layout algorithm assigns data blocks to the high-performance set and the energy-efficient set based on the data size, and keeps the replicas of data blocks in inefficient servers.

The proposed algorithm is a data distribution method, via which input data is dynamically distributed among the nodes on the base of their processing capacity in the cluster.

Recently, several scholarly papers have proposed the use of scheduling models to minimize communication costs. As an instance, an offline scheduling algorithm based on graph models was proposed by Selvitopi et al. [25], which correctly encodes the interactions between map and reduce tasks. Choi et al. [26] addressed a problem in which a map split consisted of multiple data blocks distributed and stored in different nodes. Two data-locality-aware task scheduling algorithms were proposed by Beaumont et al. [27], which optimized makespan. Makespan is defined as the time required for completing a set of jobs from the beginning to the end; i.e. the maximum completion time of all jobs. Furthermore, the scheduling algorithms proposed by Li et al. [28] were aimed at optimizing the locality-enhanced load balance and the map, local reduce, shuffle, and final reduce phases. Unlike the approaches which maximize data locality, the aim of the approach presented in the current paper is to minimize the job completion time through higher data locality rate.

In the new scheduling algorithm, the focus has been placed on two aspects: dynamic jobs priority and data localization. The main ideas of this algorithm are increasing data locality rate and decreasing average completion time of map tasks. To this end, the proposed, algorithm has been named HybSMRP. The underlying ideas were inspired by two papers. In HybS algorithm, due to the fact that direct localization technique is not used, the locality rate is not high. Furthermore, in matchmaking scheduling algorithm, since dynamic job priority is not employed, when there is a large number of map tasks in the scheduling system, the response and completion times are increased. Wordcount and Terasort benchmarks are used in the experiments carried out in the present paper. The result of Terasort and Wordcount are shown in Table 1 as x and y values, where x values correspond to Terasort and y values correspond to Wordcount. As shown in Table 1, with the increase in the number of maps, the run time of jobs is also increased.

The Job Tracker will launch one map Task for each map split. Typically, there is a map split for each input file. If the input file is too big, then there are two or more map splits associated with the same input file. This is the pseudocode used inside the method get Splits() of the File Input Format class, as shown in algorithm 1. This algorithm is used to determine the number of splits stored on HDFS.

The priority of a job affects the choice of the task assignment. The scheduling policies for the job priority can be assigned by an administrator by tuning the system parameters of the job’s workload. HybSMRP uses dynamic priority and proportional share assignment to determine which tasks from which jobs should be assigned to the available resource node. Similar to HybS, it uses Eq. (1) to take priority of jobs based on three parameters: runtime, waiting time and job size. The proposed algorithm ignores the order of job execution to preserve data locality.where td is the duration that a job waits in the queue, avgWaitTime denotes the average job waiting time in the system, r shows the average run time for the map task of MapReduce job, avgRunTime is the average run time for the map tasks for all jobs, n stands for the number of remaining tasks for the job and avgNumTask denotes the average number of tasks in all jobs in the queue. In this equation, α, β and γ are priority parameters, which can have zero or 1 values. Determining the best non zero α, β, and γ sets for a given system and workload is an interesting problem because they give applications a choice of scheduling policies. These parameters with different values can be made up of different policies, e.g. if α = 1 and β =  γ = 0, then this policy works like the FCFS (First Come First Serve) scheduler. With the increase in job waiting time, priority is increased; hence, the jobs which have been added to the queue earlier, get services earlier. This is similar to FCFS policy.

The waiting time is increased as the job waits in the queue, whereas the remaining size of the job is reduced when its individual tasks are completed. Moreover, if α = 0, β = − 1 and γ = 0, the scheduling policies can work like SJF (Shortest Job First). These policies were used to test the proposed algorithm.

The objective of the proposed scheduling algorithm is to shorten the expected response time of any given job. The proposed hybrid scheduling algorithm was evaluated with FIFO and Fair scheduling algorithm based on two parameters: map tasks’ data locality rate and completion time. These parameters are the most important criteria for evaluating scheduling algorithms in Hadoop. The localization rate is calculated from Eq. (2).

In this equation, L is the number of local map tasks, and N is the total number of map tasks.

Cloudera was used to test and evaluate our algorithm. Cloudera provides a pre-packaged, tested and enhanced open-source distribution of the Hadoop platform. Cloudera is one of the leading innovators in Hadoop space and the largest contributor to the open-source Apache Hadoop ecosystem. In this cluster, there is one master node and 20 slave nodes. The configuration details of the cluster are shown in Table 2. Wordcount and Terasort were used as benchmarks. To evaluate the proposed HybSMRP algorithm, a submission schedule was created, which generated a submission schedule for 100 jobs. Inter-arrival times were roughly exponential with a mean of 14 s.

In this study, 6400 MB input data was generated. The block size corresponding to HDFS was considered to be 64 MB and the replication level was set as 3. TeraGen benchmark was used to generate random data. In these experiments, the cluster is always configured to have just one job queue. As already mentioned, these jobs were prioritized by Eq. 1, and this priority was based on three parameters: job size, waiting time and run time. Table 3 shows the parameters of the configuration environment for tests.

Firstly, the data locality rate was used to measure the performance of the following three schedulers: Hadoop FIFO scheduler, Hadoop Fair scheduler and HybSMRP algorithm. Table 4 shows comparisons of HybSMRP, fair and FIFO algorithms based on Data locality rate on Wordcount. In the experiment in the current study, the average locality rate percentage of FIFO is 38.4%, the average of HybSMRP is 58.7% and the average of Fair is 58.4%. From Table 4, it can be seen that the HybSMRP scheduling improves the data locality by 20.3% and 0.3%, respectively, while comparing with FIFO and Fair Scheduler on average, in the running job of Wordcount. Table 5 shows comparison of HybSMRP with FIFO and Fair algorithms based on Data locality rate on running Terasort jobs. From Table 5, it can be seen that the HybSMRP scheduling improves the data locality by 14.8% and 18.9%, respectively, while compared with FIFO and Fair Scheduler on average.

In order to evaluate the completion time of HybSMRP, its performance was compared with FIFO and Fair on Wordcount and Terasort. In this test, the number of tasks was increased so that the running time of the jobs could clearly be determined. Tables 6 and 7 show the experiment results. As shown in Table 6, the proposed method is very similar to Fair Scheduler; however, HybSMRP is approximately 2.19% faster than FIFO and 0.79% faster than Fair scheduling algorithm. It can be seen from Table 7 that the HybSMRP improves completion time by about 11.51% compared to Fair and 29.15% compared to FIFO scheduling algorithm.

In this paper, a new hybrid scheduling algorithm, HybSMRP, has been proposed in Hadoop MapReduce framework to improve data locality rate and decrease completion time with two techniques; namely dynamic priority and localization ID. The proposed algorithm was compared with Hadoop default scheduling algorithms. Experimental results demonstrate that the proposed Hybrid scheduling algorithm can often obtain high data locality rate and low average completion time for map tasks. In the future, we will integrate our algorithm with a technique that determines the number of local tasks and mixing with delay scheduling algorithm. Delay scheduling is a simple technique for achieving locality and fairness in cluster scheduling. We also plan to focus on evaluating the performance of the proposed scheduler at scale by deploying it on a large cluster to clearly show our contribution.