WIXLIB

(a) Resource availability.Fig. 2.(b) FIFO Scheduler.(c) Fair Scheduler.(d) Ideal Scheduler.Performance of FIFO Scheduler, Fair Scheduler and an ideal scheduler in a dynamic Hadoop cluster.Knowing that more resources would be available in the 30thto 40th interval, the ideal scheduler prioritized J2 with moreresources from 10th to 30th . The flexible resource allocationeffectively guaranteed the deadlines of both jobs.We make two observations. First, deadline-oblivious schedulers perform poorly for jobs with deadlines in a dynamiccluster. Second, knowledge on future resource availabilityis critical to avoiding deadline misses. For example, if theresource level in the 20th to 30th interval further drops, adeadline-aware scheduler should allocate more resources toJ2 in the 10th to 20th interval. In summary, the informationon future resource availability affects a deadline-aware scheduler’s decisions on individual job resource allocations.For this job setting, there exists a schedule that meets bothjob’s deadline. For such a schedulable job set, earliest deadlinefirst scheduling (EDF) can also meet the deadlines. EDF isoptimal on preemptive uniprocessors [19]. However, EDF haspractical issues when used as a Hadoop job scheduler. First,in a resource constrained scenario or an overloaded system,where not all job deadlines can be met, the performance ofEDF is unpredictable and often quite bad. Figure 3(a) showsthe scheduling order of two jobs under EDF. The system become overloaded as the available resources dropped at the 10thminute. For this system, EDF is clearly not optimal because J2would have meet its deadline if it was scheduled before J1 (asshown in Figure 3(b)). Second, EDF does not determine howmuch resource is needed to meet job deadlines [24]. It maylead to over-provisioning or under-provisioning of resources.Finally, EDF overwrites user-defined job priorities making itless attractive in a multi-user environment.[Summary] We have shown that deadline-aware schedulershave significant advantage over deadline-oblivious ones. However, obtaining an optimal or near-optimal job scheduler isnot trivial in clusters with dynamically variable resources.We have shown that resource-oblivious schedulers such asEDF only performs scheduling optimizations based on currentobservation of resource availability. Looking forward into thefuture resource availability can guide schedulers in makingwise resource allocation decisions. These findings motivatedus to develop a resource and deadline-aware scheduler fordynamic Hadoop clusters.(a) EDF-preemptive.Fig. 3.(b) Ideal scheduler.Performance comparison in an overloaded system.estimations of job completion time and the predictions onfuture resource availability. To realize fine-grained resourcecontrol, RDS divides job execution into control intervals andbuilds performance models for estimating job progress, fromwhich overall job completion time can be inferred. Based onthe job completion time estimation and resource availabilityprediction, RDS derives the optimal job scheduling via anonline receding horizon control (RHC) algorithm.Figure 4 shows the architecture of RDS. We describe thefunctionality of each component as follows: Fuzzy performance model takes allocated resourcesand job size as inputs and generates the estimated jobcompletion time as outputs. The model is updated periodically based on the measured job progress at eachcontrol interval.Resource predictor takes the history information onresource availability and predicts the amount of availableresources for the next few intervals.Scheduling optimizer adjusts the number of slots allocated to each running job based on an online recedinghorizon control algorithm.We formulate the resource and deadline-aware schedulingas an optimization problem that minimizes job deadline misspenalty. We present detailed design of the self-adaptive fuzzymodel, the resource model and the scheduling optimizer.A. Problem FormulationWe consider a Hadoop cluster with dynamic resource availability ra . Consider that there are J jobs running in the systemand the control interval is t (t [1, ., T ]). Each job j hasrmap tasks allocated umj resource and reduce tasks allocated ujrefresource. yj is the actual completion time of job j and yjis the reference time that meets its deadline. The optimizationIII. RDS D ESIGNIn this section, we present the design of RDS, a resourceand deadline-aware Hadoop job scheduler. RDS determinesthe resource allocations to individual jobs based on the958

Fig. 4.allocation considering future resource availability. Δumj (t) andΔurj (t) are the resource adjustment for map and reduce tasksin order to meet the expected job progress in each interval,respectively. They together represent the control penalty andare weighted by penalty matrices P and Q. The first part of thefunction serves as a penalty for not meeting a job’s progresstarget and the second part requires the cost for changing job’sresource allocations be minimized.To consider future resource availability, the RHC controllerpredicts a job’s performance for the next Hp control intervals. It then computes a sequence of control actions Δu(t),Δu(t 1), . . . , Δu(t Hc ) over Hc control periods, calledthe control horizon, to keep the predicted performance closeto their expected targets. Thus, a performance model is neededto predict the progress of a job in a control interval given acertain amount of resources. The total amount of resourcesavailable in future intervals needs to be predicted.The architecture of RDS.problem is formulated as follows,minT J t 1 j 1s.t.J ωj (yj yjrefyjref) ,r(umj u j ) ra .(1)B. Estimating Job Execution ProgressWe use a multiple-input-single-output fuzzy model to predict a job’s execution progress based on its input sizeand resource allocation in each control interval. The fuzzymodel is often used to capture the complex relationshipbetween resource allocations and a job’s fine-grained executionprogress [14]. However, a job’s progress can be affected bymany factors. First, job progress is not uniform at differentexecution phases, e.g., map and reduce phases. Second, evenwithin the same phase, data skew among tasks leads todifferent task execution speed at different intervals. Finally,co-running jobs may unpredictably interfere with a job’sexecution, making the mapping of resource to job progressvariable. Therefore, we design an online self-adaptive fuzzymodel based on real-time measurements of job progress.1) Fuzzy Model: The job j execution progress in thecontrol interval t is represented as the input-output NARX type(Nonlinear Auto Regressive model with eXogenous inputs),(2)j 1The goal of RDS scheduler is to minimize the penalty ofdeadline misses. Objective Eq. (1) captures the lost revenuedue to the deadline misses. The penalty remains zero until thejob misses its deadline. ωj is a constant that represents thepriority of job j. Constraint Eq. (2) ensures that the sum ofresources assigned to all running jobs must be bounded by thetotal available resources in the cluster.However, a job’s completion time yj can only be measuredwhen the job finishes. It is often too late for the Hadoopscheduler to intervene if a job already misses its deadline. Tothis end, we break down a job’s execution into small intervalsand apply calibrated deadlines for each interval. Considera job’s deadline is 100 minutes away and the execution isdivided into 10 intervals. If the job can finish one tenth of thetotal work in each interval, it can meet the overall deadline.The Hadoop scheduler can adjust the resource allocation if ajob’s execution is considered slow based on its progress onindividual intervals. Such a breakdown of job execution alsoallows the scheduler to look forward into future intervals andapply optimization considering future resource availability.Specifically, we transform the optimization problem to areceding horizon control (RHC) problem that minimizes thefollowing objective function:J(t) Hp J i 0 j 1 Hc J yj (t) Rj (u(t), dj , ξ(t)).(4)R is the relationship between the input variables and the outputvariable. The input variables are the current resource allocationu(t), the job input size dj , and the regression vector ξ(t).Here, resource allocation, u(t) [um (t), ur (t)], includes bothmap resource allocation um (t) and reduce resource allocationur (t). The regression vector ξ(t) contains a number of laggedoutputs and inputs of the previous control periods. It isrepresented as yj (t i) yjref (t i) 2Wξ(t) [(y(t 1), y(t 2), · · · , y(t ny )),(3)(u(t), u(t 1), · · · , u(t nu ))]T2r2( Δumj (t i) P Δuj (t i) Q ),(5)where ny and nu are the number of lagged values for outputsand inputs, respectively. Let ρ denote the number of elementsin the regression vector ξ(t), that is,i 0 j 1where t is the measurement (or control) interval, yj (t) is theactual progress of the job in interval t and yjref (t) is theexpected progress that ensures meeting the job’s deadline. Wis the job priority matrix. The optimization looks forward intofuture Hp intervals and tries to derive the optimal resourceρ ny nu .(6)R is the rule-based fuzzy model that consists of Takagi-Sugenorules [5]. A rule Rj is represented as959

Rj : IF ξ1 (t) is Ωj,1 , ξ2 (t) is Ωj,2 , · · · , and ξρ (t) is Ωj,ρu(t) is Ωj,ρ 1 and dj is Ωj,ρ 2THEN yj (t) ζj ξ(t) ηj u(t) ωj dj θj .(7)TABLE II3- STEP PREDICTION OF ARIMASteps123Here, Ωj is the antecedent fuzzy set of the jth rule, whichis composed of a series of subsets: Ωj,1 , Ωj,2 , · · · , Ωj,ρ 2 . ζj ,ηj and ωj are parameters, and θj is the offset. Their valuesare obtained by offline training. Each fuzzy rule characterizesthe nonlinear relationship between allocated resources andperformance for a specific job type.2) Online Self-Learning: Due to the dynamics of MapReduce job behaviors (e.g., data skews, different phases andmulti-tenant interferences), we design an online self-learningmodule to adapt the fuzzy model. It aims to minimize theprediction error of the fuzzy model e(t), which is the errorbetween actual measured job progress and predicted value.If e(t) 0, we apply a recursive least squares (RLS)method [2] to adapt the parameters of the current fuzzyrule. The technique updates the model parameters as newmeasurements are sampled from the runtime system. It appliesexponentially decaying weights on the sampled data so thathigher weights are assigned to more recent observations.We express the fuzzy model output in Eq.(4) as follow:y(t) φ(t)X e(t)t (e(t)2 τ e(t 1)2 ).Inputs of ARIMAra (t 3 t), ra (t 2 t), ra (t 1 t)ra (t 2 t), ra (t 1 t), ra (t t)ra (t 1 t), ra (t t), ra (t 1 t)Outputra (t t)ra (t 1 t)ra (t 2 t)dow h N . Let ra (t h t) denote the hth step prediction ofra (t) knowing the last n observations, i.e., ra (t n), ., ra (t 1). The available resource series ra (t t), ra (t 1 t), ., ra (t h t) are obtained by iterating the one-step prediction. Table IIillustrates how one-step prediction is iterated to obtain a 3step prediction. We study the impact of different predictionhorizons in Section V.D. Scheduling OptimizerThe scheduling optimizer is invoked at each control interval.It solves the RHC control problem using quadratic programming and outputs a sequence of resource adjustments thatminimize fine-grained job deadline misses at each interval.Then, RDS applies the resource adjustment to individual jobsvia a two-level scheduling.Job management: RDS maintains two separate queues forcurrent running jobs and waiting jobs. By default, incomingjobs enter the waiting queue. The scheduling optimizer determines which job is to be moved to the running queue based onthe solution of the RHC control problem. For example, lowpriority jobs or jobs with distant deadlines may not be immediately allocated resources by the RHC control algorithm,i.e., uj (t) 0. These jobs wait until they receive resourceallocation from the scheduling optimizer. Since the resourceallocation is determined once for every control interval, it ispossible that short jobs whose deadline is earlier than thenext control interval may miss their deadlines due to the lateallocation of resources. To this end, we provide a fast pathfor short jobs in the job queue management, that is, RDSimmediately moves a short job to the running queue.Task scheduling: RDS applies the resource adjustment toindividual jobs by changing the number of execution slotsassigned to each job. Assigning more homogeneous slots toa job leads to more resources allocated to the job. Althoughthe actual resources allocated with different slots may vary,we found that the total number of slots assigned to a job is agood approximation of the job’s allocated resources. RDS usesa minimally invasive approach to realize dynamic number ofslots assigned to each job. Algorithm 1 shows how dynamicslots are realized via task scheduling. First, jobs are sortedaccording to their resource adjustment in the next controlinterval. At each heartbeat, the job with the largest resourceadjustment (line 5) is selected to run its task. After assigningone slot to this job, the job’s resource adjustment is updatedby subtracting the amount of resource equivalent to the sizeof one slot rslot (line 9). We calculate rslot based on the totalavailable resource ra and the total number of slots in thecluster. We discuss two scenarios in Section V, where rslotis treated differently.(8)where e(t) is the error between the actual output and predictedoutput. φ(t) [φT1 , φT2 , ., φTρ ] is a vector composed of themodel parameters. X [σ1 X(t), σ2 X(t), ., σρ X(t)] whereσj is the normalized degree of fulfillment or firing strengthof jth rule and X(t) [ξ(t)T , u(t)] is a vector containingthe previous outputs and inputs of the control system. Theparameter vector φ(t) is estimated so that the error functionin Eq.(9) is minimized. We apply both the current error e(t)and the previous error e(t 1) to estimate the parameter vector,Error MODEL .(9)t 1Where τ is called the discount factor as it gives higher weightson more recent samples in the optimization. It determines inwhat manner the current prediction error and old errors affectthe update of parameter estimation.C. Predicting Resource AvailabilityWe use a simple but effective Auto-Regressive IntegratedMoving Average (ARIMA) model [3] to predict the resourceavailability in future intervals based on history information.The ARIMA model has been used to predict resource consumption [26], and dynamic power supply [4].In ARIMA model, the available resource of the currentinterval ra (t) is predicted based on the last n observationsof resource availability, i.e., ra (t 1), . ra (t n 1).ra (t) a1 ra (t 1) a2 ra (t 2) . an ra (t n), (10)where a1 , a2 , . . . , an are coefficients obtained via model fitting. RDS predicts future resource availability over a time win-960

Brown energyGreen energyPower (Kw)2.52: repeat3:if Any slot is available then4:/*Select a job in the running job queue*/5:j arg max[Δuj (t)], j {1, · · · , J}Resource AvailabilityCPU resource (GHz)Algorithm 1 Task scheduling with dynamic slots.1: Update the running job queue21.510.56:Select a local task from the job j7:Assign selected task to the available slot8:/*Update control adjustment scheme*/9:Δuj (t) Δuj (t) rslot10:end ifJmr11: untilj 1 (uj (t) uj (t)) ra403020102j50468 10 12 14 16 18Control intervals (th)(a) Energy supply.Fig. 5.2468 10 12 14Control intervals (th)1618(b) Dynamic resource.A dynamic Hadoop cluster.MATLAB to compute the local control solution. We usedMATLAB Builder JA to create a Java class from the MATLABprogram calling quadprog. This Java class was integratedinto RDS and deployed in the master node of the cluster.Based on the observations, we empirically set the controlpenalty weight matrix P [0.0107, 0.0096, 0.0132] and Q [0.0173, 0.0168, 0.0194] for map and reduce task, respectively.We set the control interval to 10 minutes.The algorithm effectively changes the number of slots ofeach job. If Δuj (t) 0, the job eventually will be assignedfree slots. If Δuj (t) 0, the job actually gives up theopportunity to run tasks, which is equivalent to reducing itsnumber of allocated slots.C. WorkloadsIV. S YSTEM I MPLEMENTATIONFor performance evaluations, we used a set of representativeMapReduce applications from the PUMA benchmark [1], i.e.,Wordcount, Terasort and Grep. By default, we set the samepriority for each benchmark. We study the impact of differentjob priorities in Section V-D. For each application, we submitmultiple copies with different input sizes as shown in theTable III that contains jobs with widely varying executiontimes and data set sizes, emulating a scenario where the clusteris used to run many different types of MapReduce applications.Similarly as Natjam et al. [8], we set the expected executiontime of individual jobs to 2.5 times of the job completion timein dedicated cluster with sufficient resources. Then, we derivedjobs’ deadlines according to their arrival times. According tothe study [25], we set inter-arrival time of jobs to 10 minutes.A. TestbedWe built a dynamic Hadoop cluster in our university cloud,which consists of 108-core CPUs and 704 GB memory.VMware vSphere 5.1 was used for server virtualization.VMware vSphere module controls the CPU usage limits inMHz allocated to the virtual machines (VMs). It also providesan API to support the remote management of VMs. Hadoopversion 1.2.1 was deployed to the cluster with 21 VMs, i.e.,one master node and 20 slave nodes. We configured each slavenode with two map slots and one reduce slot. The block size isconfigured 64 MB in our experiments. Each VM was allocated1 virtual CPU and 2 GB memory. All VMs ran Ubuntu Server10.04 with Linux kernel 2.6.32.B. RDS ImplementationD. Dynamic Resource TraceTo implement RDS in the Hadoop environment, we added anew member mapred.job.deadline to store the deadlineof a job to the class JobConf. We applied the idea of Hadoopcapacity scheduler and refactoried its QueueManager classto implement RDS waiting and running job queues. Weimplemented the core components of RDS in the classSchedulerTaskScheduling.Resource Predictor: We used a sensor program providedby VMware vSphere 5.1 to collect the resource availability ofindividual VMs. The cluster information, e.g., number of slots,was monitored by the function getClusterStatus in theJobTracker. We applied the ARIMA approach combined withthe collected resource information of last 3 horizons to obtainthe predicted resource in the next 3 intervals.Performance Modeling: We used MATLAB Fuzzy LogicToolbox to apply subtractive clustering and ANFIS modelingtechnique on the data collected from cluster. At runtime, theperformance models were updated based on new measurements collected from the JobTracker using RLS algorithm.

using VMware's distributed resource scheduler to emulate the dynamics in resource availability of the cluster. The total resource was evenly distributed to each slave node. The master node was allocated a fixed capacity. We ran two differentword-count [1] jobs. Two jobs have different input sizes, resource demands and deadlines.