# PAS: Performance-Aware Job Scheduling for Big Data Processing Systems

Authors: Yiren Li; Tieke Li; Pei Shen; Liang Hao; Jin Yang; Zhengtong Zhang; Junhao Chen; Liang Bao
Journal: Security and Communication Networks
 (2022)

---

## Abstract

Big data analytics has become increasingly vital in many modern enterprise applications such as user profiling and business process optimization. Today’s big data processing systems, such as Hadoop MapReduce, Spark, and Hive, treat big data applications as a batch of jobs for scheduling. Existing schedulers in production systems often maintain fair allocation without considering application performance and resource utilization simultaneously. It is challenging to perform job scheduling in big data systems to achieve both low turnaround time and high resource utilization due to the high complexity in data processing logics and the dynamic variation in workloads. In this article, we propose a performance-aware scheduler, referred to as PAS, which dynamically schedules big data jobs in Hadoop YARN and Spark and autonomously adjusts scheduling policies to improve application performance and resource utilization. Specifically, PAS schedules multiple concurrent jobs using different policies based on the predicted job completion time and employs a greedy approach and a one-step lookahead strategy to opportunistically maximize the average job performance while still maintaining a satisfactory level of resource utilization. We implement PAS in Hadoop YARN and evaluate its performance with HiBench, a well-known big data processing benchmark. Experimental results show that PAS reduces the average turnaround time by 25% and the makespan by 15% in comparison with four state-of-the-art schedulers.


---

## Body

## 1. Introduction

With the prosperity of big data analytics and artificial intelligence, big data processing systems (BDPSs) are playing critical roles in modern enterprises’ applications. The job scheduler continues to be a key component to a BDPS, in which diverse coexisting jobs from many users and applications contend for resources in a shared environment. As the data volume increases and the demand for analytics jobs surges, typical production BDPSs are frequently resource-constrained. Therefore, efficient resource management comes as the top priority for cluster schedulers [1]. Recently, due to the prevalence of “data analysis as a service” (DAAS), the BDPSs running on the public cloud are providing data analysis abilities for different users.To balance interests of both users and service providers, performance (from user side) and resource efficiency (from provider side) need to be concerned simultaneously for the system design on those shared environments [2]. User-side performance is often measured using average turnaround time, which is the average time intervals from the time of each submission of their job to the time of its completion. Provider-side resource efficiency indicates the efficiency of the resource usage and it is usually measured with the resource utilization ratio and makespan of a set of jobs. These objectives are opposite, because of the conflict of interest between users and the service providers. Current production schedulers often settle for isolation guarantee as the primary objective and seek to maintain fair allocations at all time, which is neither necessary nor efficient [1].Motivated by this intuition, in this article, we propose a performance-aware scheduling algorithm to opportunistically improve the average job performance for fast job completion for users while still achieving good enough resource efficiency for service providers. To this end, we develop a performance prediction method for estimating completion time of a data analytical job under different resource utilization conditions, formulate a job scheduling problem for big data processing systems (JS-BDPS) in order to minimize average turnaround time under a user-specified resource utilization constraint, and propose a performance-aware scheduling solution. Specifically, our work makes the following contributions to the field:(1)
The method for performance prediction is based on generic computer and program models, which provide an accurate estimation of completion time for big data analytical jobs and make it directly applicable to different jobs running on various big data processing systems.(2)
The formulation of theJS-BDPS problem and the performance-aware scheduling approach that employs a greedy and a one-step lookahead strategy to solve the JS-BDPS problem effectively and efficiently.(3)
The performance prediction method is validated and justified by experimental results using a well-known big data benchmark on disparate computing nodes, and the performance superiority of the proposedPAS scheduling approach is illustrated by extensive simulation results in comparison with four state-of-the-art algorithms.The rest of the article is organized as follows. We discuss the related work in Section2. Section 3 formulates the JS-BDPS problem and discuss the objectives and constraints. In Section 5, we construct the regression model of a job for performance estimation and propose the performance-aware scheduling (PAS) policies to minimize the average turnaround time for a set of scheduling jobs. Section 6 presents the design and implementation of the PAS algorithm on top of Hadoop YARN and Spark. In Section 7, we describe the experimental setup and evaluates the performance model and scheduling algorithm using a well-known big data benchmark. We conclude with a discussion of our approach and a sketch of future work in Section 8.


## 2. Related Work

Job scheduling for BDPSs has received much attention from both industry and academia. From the perspective of scheduling goal, previous studies can be classified into two categories: performance-oriented and fairness-oriented approaches, as discussed next.


### 2.1. Performance-Oriented Scheduling

Maximizing resource utilization and minimizing makespan are two common goals for performance-oriented scheduling [2]. Yao et al. [3] proposed HaSTE to improve the resource utilization by using efficient task packing according to diverse resource requirements and dependencies between tasks. Their later work proposed OpERA [4] to leverage the knowledge of actual runtime resource utilizations as well as future resource availability for task assignments. Quasar [5] used classification techniques to find appropriate resource allocations to applications in order to fulfill their QoS requirements and maximize system resource utilization. Polo et al. [6] dynamically adjusted slots on each machine to maximize the cluster utilization. Verma et al. [7] allocated resource to jobs by using job profiles to estimate the requested resource that meets the deadline. Cheng et al. [8] proposed a deep reinforcement learning (DRL)-based job scheduler that dispatches the jobs in real time to deal with real-time workloads. Fan et al. [9] proposed an intelligent scheduling framework for different hardware resources and increasing diverse workloads in modern job scheduling. Zheng et al. [10] designed an online algorithm for SaaS providers to optimally purchase IaaS instances and schedule pleasingly parallel job.To minimize the makespan of a set of independent MapReduce jobs, Verma et al. [11] introduced a heuristic scheduling algorithm. Huang et al. [12] tried to optimize the makespan by estimating the completion time of jobs that are prone to error. Hou et al. [13] proposed a deadline-aware scheduling algorithm to reduce the average job execution time by checking the percentage of tasks and allocating resources. Wang et al. [14] proposed a workflow-based scheduling algorithm to satisfy the budget and the deadline constraints. Khan et al. [15] applied linear regression method to perform runtime estimation for Hadoop jobs. Lim et al. [16] proposed CP-Scheduler to estimate task execution time and handle MapReduce jobs with deadlines. Shao et al. [17] proposed an energy-aware greedy algorithm (EAGA) for fine-grained task placement to minimize the energy consumption and job execution time. Chen et al. [18] observed that with demand elasticity, a job requires significantly less amount of resources, only at the cost of a moderate performance penalty. Amer et al. [19] tackled the multi-objective scheduling problem and presents a modified Harris hawks optimizer (HHO)for multi-objective scheduling problem. Khan et al. [20] proposed a task scheduling method based on a hybrid optimization algorithm, which effectively schedules jobs with the least amount of waiting time. Meyer et al. [21] proposed a machine learning-driven classification scheme for dynamic interference-aware resource scheduling in cloud computing environments. They presented a classification approach to better represents the workload variations for resource scheduling. Chen et al. [22] considered the heterogeneous characteristics of data centers and modeled energy consumption based on the frequency and kernel number of the virtual machine CPU.


### 2.2. Fairness-Oriented Scheduling

Fairness is another important factor for a scheduling framework. Matei et al. [23] proposed a delay scheduling policy to improve the performance of Fair Scheduler by increasing the data locality of Hadoop. Dominant resource fairness (DRF) [24] is the first work to generalize the max-min fairness to multiple resource types on Hadoop YARN. Wang et al. [25] extended the DRF algorithm for the heterogeneous environment. Many production schedulers, such as Hadoop’s Fair Scheduler [26], Quincy [27], Mesos [28], and Choosy [29], support max-min fairness or its extensions. Liu et al. [30] presented a resource allocation mechanism to enable fair sharing multiple types of resource among multiple tenants. Huang et al. [31] calculated the approximated total workload according to the job’s runtime distribution and performed resource allocation accordingly in order to maximize the client-specified utilities regarding max-min fairness. Wang et al. [32] corrected the monopolizing behavior of long reduce tasks from large jobs and dynamically balanced the execution of different jobs for fair and fast completion.Some other studies consider the trade-off between performance and fairness simultaneously. In [33], a general meta-scheduler to leverage existing schedulers in Hadoop YARN to implement the efficiency-fairness trade-off was proposed. Wang et al. [34] utilized many metrics to efficiently balance the performance and the fairness, as well as to reduce the makespan of MapReduce tasks. Tang et al. [35] presented DynamicMR, a dynamic Hadoop slot allocation (DHSA) framework aiming to improve the performance of MapReduce workloads while maintaining the fairness. Pastorelli et al. [36] presented HFSP, a size-based scheduler with aging to implement fairness and near-optimal system response times on Hadoop. Niu et al. [37] presented an adaptive scheduler called Gemini, which adaptively decides the proper scheduling policy according to the running workload, in order to achieve better performance as well as fairness.


## 2.1. Performance-Oriented Scheduling

Maximizing resource utilization and minimizing makespan are two common goals for performance-oriented scheduling [2]. Yao et al. [3] proposed HaSTE to improve the resource utilization by using efficient task packing according to diverse resource requirements and dependencies between tasks. Their later work proposed OpERA [4] to leverage the knowledge of actual runtime resource utilizations as well as future resource availability for task assignments. Quasar [5] used classification techniques to find appropriate resource allocations to applications in order to fulfill their QoS requirements and maximize system resource utilization. Polo et al. [6] dynamically adjusted slots on each machine to maximize the cluster utilization. Verma et al. [7] allocated resource to jobs by using job profiles to estimate the requested resource that meets the deadline. Cheng et al. [8] proposed a deep reinforcement learning (DRL)-based job scheduler that dispatches the jobs in real time to deal with real-time workloads. Fan et al. [9] proposed an intelligent scheduling framework for different hardware resources and increasing diverse workloads in modern job scheduling. Zheng et al. [10] designed an online algorithm for SaaS providers to optimally purchase IaaS instances and schedule pleasingly parallel job.To minimize the makespan of a set of independent MapReduce jobs, Verma et al. [11] introduced a heuristic scheduling algorithm. Huang et al. [12] tried to optimize the makespan by estimating the completion time of jobs that are prone to error. Hou et al. [13] proposed a deadline-aware scheduling algorithm to reduce the average job execution time by checking the percentage of tasks and allocating resources. Wang et al. [14] proposed a workflow-based scheduling algorithm to satisfy the budget and the deadline constraints. Khan et al. [15] applied linear regression method to perform runtime estimation for Hadoop jobs. Lim et al. [16] proposed CP-Scheduler to estimate task execution time and handle MapReduce jobs with deadlines. Shao et al. [17] proposed an energy-aware greedy algorithm (EAGA) for fine-grained task placement to minimize the energy consumption and job execution time. Chen et al. [18] observed that with demand elasticity, a job requires significantly less amount of resources, only at the cost of a moderate performance penalty. Amer et al. [19] tackled the multi-objective scheduling problem and presents a modified Harris hawks optimizer (HHO)for multi-objective scheduling problem. Khan et al. [20] proposed a task scheduling method based on a hybrid optimization algorithm, which effectively schedules jobs with the least amount of waiting time. Meyer et al. [21] proposed a machine learning-driven classification scheme for dynamic interference-aware resource scheduling in cloud computing environments. They presented a classification approach to better represents the workload variations for resource scheduling. Chen et al. [22] considered the heterogeneous characteristics of data centers and modeled energy consumption based on the frequency and kernel number of the virtual machine CPU.


## 2.2. Fairness-Oriented Scheduling

Fairness is another important factor for a scheduling framework. Matei et al. [23] proposed a delay scheduling policy to improve the performance of Fair Scheduler by increasing the data locality of Hadoop. Dominant resource fairness (DRF) [24] is the first work to generalize the max-min fairness to multiple resource types on Hadoop YARN. Wang et al. [25] extended the DRF algorithm for the heterogeneous environment. Many production schedulers, such as Hadoop’s Fair Scheduler [26], Quincy [27], Mesos [28], and Choosy [29], support max-min fairness or its extensions. Liu et al. [30] presented a resource allocation mechanism to enable fair sharing multiple types of resource among multiple tenants. Huang et al. [31] calculated the approximated total workload according to the job’s runtime distribution and performed resource allocation accordingly in order to maximize the client-specified utilities regarding max-min fairness. Wang et al. [32] corrected the monopolizing behavior of long reduce tasks from large jobs and dynamically balanced the execution of different jobs for fair and fast completion.Some other studies consider the trade-off between performance and fairness simultaneously. In [33], a general meta-scheduler to leverage existing schedulers in Hadoop YARN to implement the efficiency-fairness trade-off was proposed. Wang et al. [34] utilized many metrics to efficiently balance the performance and the fairness, as well as to reduce the makespan of MapReduce tasks. Tang et al. [35] presented DynamicMR, a dynamic Hadoop slot allocation (DHSA) framework aiming to improve the performance of MapReduce workloads while maintaining the fairness. Pastorelli et al. [36] presented HFSP, a size-based scheduler with aging to implement fairness and near-optimal system response times on Hadoop. Niu et al. [37] presented an adaptive scheduler called Gemini, which adaptively decides the proper scheduling policy according to the running workload, in order to achieve better performance as well as fairness.


## 3. Problem Statement

This article focuses on the job scheduling problem for big data processing systems (short forJS-BDPS problem). Generally, the architecture of scheduling frameworks of BDPSs can be centralized, centralized two-level, distributed two-level, or shared-state [2]. This article focuses on the centralized two-level (CTL) architecture because it is easy to implement and can generate an optimal or near-optimal scheduling plan. Many practical scheduling frameworks, such as Yarn [38], Mesos [28], Fuxi [39], Teris [40], and Corral [41], have adopted the CTL architecture. In a typical scheduling framework with the CTL architecture, the scheduler is responsible for allocating resources to the various running jobs subject to constraints of capacities, queues, etc.The goal ofJS-BDPS problem is to find an optimal scheduling policy that minimize the average turnaround time (ATAT) of the submitted jobs within a predefined time period, given a set of jobs, and the underlying runtime environment. The reason we choose the turnaround time metric is because it can tell how long a job can take to finish execution since it arrives at the scheduler, which characterizes a most important capability of a scheduler. Other measurements such as energy, cost, makespan, and load balancing are integrated by turnaround time in some ways [2].Notably, theJS-BDPS problem has the following components:


### 3.1. Job

A job represents a big data processing application running on a specific BDPS. We model it as a 4-tuplej=f,r,tw,te, where f denotes the processing logic of j; r represents the required resource for running j; tw records the waiting time, that is the accumulated queuing time for scheduling, of j; and te indicates the execution time of j red and assumes the jobs obtain the same priority. Once submitted, jobs need to be queued for a while until they are scheduled. Thereafter, the scheduler assigns the necessary resource r to j and j will be continuously executed in the BDPS until completion.


### 3.2. Resource

A cluster of BDPS often contains different types of resource, that is, CPU, memory, and network bandwidth, that a job needs for executing, and a job usually need to describe the necessary resource it required to initiate its execution. For example, a MapReduce job running on a YARN scheduler has to declare the number of CPU core and the memory size when submitting. Without loss of generality, we model the resource requirement of a jobj as a 2-tuple r=c,m, where c is the number of CPU cores and m denotes the memory size, and R=C,M is used to indicate the total resource held by a BDPS cluster.


### 3.3. Turnaround Time

Turnaround time (TAT) is an important metric in evaluating the scheduling algorithms from the users’ perspective [2]. For a specific job ji, its turnaround time TATi is the time interval from the time of ji’s submission to the time of the ji’s completion, that is TATi=tiw+tie.


### 3.4. Resource Utilization Ratio

The resource utilization ratio (RUR) is an important performance metric that is used to measure the efficiency of a BDPS cluster from the service providers’ perspective [33]. Given any time point t, the RURt of a BDPS cluster is defined as a 2-tuple:(1)RURt=RusedtR=CusedtC,MusedtM,where Rused is the already used resource of the cluster at t and R is the total resource of the cluster.In summary, given a set ofn jobs J=j1,j2,…,jn submitted to the scheduler, the JS-BDPS problem can be stated as follows:(2)min ATAT=1n∑i=1nTATi,s.t.L≤RURt≤U,∀t.where equation (2) states that the goal of JS-BDPS problem is to find the optimal job scheduling sequence that minimizes the average turnaround time (ATAT) of J. The constraint (2) is that at any time point t, the RUR of any solution must be within L,U, the resource utilization bound predefined by the BDPS.


## 3.1. Job

A job represents a big data processing application running on a specific BDPS. We model it as a 4-tuplej=f,r,tw,te, where f denotes the processing logic of j; r represents the required resource for running j; tw records the waiting time, that is the accumulated queuing time for scheduling, of j; and te indicates the execution time of j red and assumes the jobs obtain the same priority. Once submitted, jobs need to be queued for a while until they are scheduled. Thereafter, the scheduler assigns the necessary resource r to j and j will be continuously executed in the BDPS until completion.


## 3.2. Resource

A cluster of BDPS often contains different types of resource, that is, CPU, memory, and network bandwidth, that a job needs for executing, and a job usually need to describe the necessary resource it required to initiate its execution. For example, a MapReduce job running on a YARN scheduler has to declare the number of CPU core and the memory size when submitting. Without loss of generality, we model the resource requirement of a jobj as a 2-tuple r=c,m, where c is the number of CPU cores and m denotes the memory size, and R=C,M is used to indicate the total resource held by a BDPS cluster.


## 3.3. Turnaround Time

Turnaround time (TAT) is an important metric in evaluating the scheduling algorithms from the users’ perspective [2]. For a specific job ji, its turnaround time TATi is the time interval from the time of ji’s submission to the time of the ji’s completion, that is TATi=tiw+tie.


## 3.4. Resource Utilization Ratio

The resource utilization ratio (RUR) is an important performance metric that is used to measure the efficiency of a BDPS cluster from the service providers’ perspective [33]. Given any time point t, the RURt of a BDPS cluster is defined as a 2-tuple:(1)RURt=RusedtR=CusedtC,MusedtM,where Rused is the already used resource of the cluster at t and R is the total resource of the cluster.In summary, given a set ofn jobs J=j1,j2,…,jn submitted to the scheduler, the JS-BDPS problem can be stated as follows:(2)min ATAT=1n∑i=1nTATi,s.t.L≤RURt≤U,∀t.where equation (2) states that the goal of JS-BDPS problem is to find the optimal job scheduling sequence that minimizes the average turnaround time (ATAT) of J. The constraint (2) is that at any time point t, the RUR of any solution must be within L,U, the resource utilization bound predefined by the BDPS.


## 4. Discussion

### 4.1. Performance-Aware Job Scheduling

The key idea of our performance-aware job scheduling (PAS) approach is to train a performance prediction model for each job according to their historical profiles and repeatedly assign resource to the job(s) with the lowest cost gain using the trained model. PAS also applies a heuristic one-step lookahead strategy to find potentially good scheduling policy.


### 4.2. Predicting Job Completion Time

A BDPS can support many data analytical jobs running on it simultaneously. We observe that the completion time of a specific jobj varies under different resource utilization ratio (RUR) of the BDPS. Suppose the completion time j is t when executing it exclusively, that is RUR0=0, on a BDPS. If RUR increases to RURk0<RURk<1 by adding more jobs to the BDPS, at each time slot Δt there are statistically 1−RURk⋅Δt share of processing time can be used for executing j. Furthermore, when RURk continues to grow, the swapping and scheduling time among different jobs becomes significantly large [42]. Based on these observations and the job complexity estimation method proposed in [43], we model the job completion time t and the resource utilization ratio (RUR) as a power-law function:(3)t=a⋅RURb+c,where a, b, and c are regression constants.The prediction model indicates that when resource utilization ratio grows, the completion time of a job grows in a power number. Such conclusion has been well verified in our experiments.


### 4.3. Scheduling Gain and ATAT

To schedule a set of jobs dynamically,PAS has to make decision periodically. More specifically, at each time point t, it needs to decide whether to schedule a set of jobs J from the queue (for decreasing the tw and te of all jobs in J), or to do nothing (for decreasing the execution time of all running jobs). To lucubrate the scheduling process, we discrete the time into many small and equal slots, that is Δt, and denote the kth time interval as τkτk=t0+k−1Δt,t0+kΔt, where t0 represents the start time of scheduling. Note that the discretized time slot Δt is a hyperparameter of PAS that is configurable by the administrator of the BDPS, and it defines the scheduling frequency of the PAS.Based on the discretized time slot, we can define thescheduling gain to quantify the gain by scheduling a set of jobs:Definition 1.
Scheduling gain atτk, denoted by Gk, is defined as the decrease of the waiting time for scheduling a subset of submitted jobs Jsk minus the increase of the execution time for the running jobs Jrk caused by the newly scheduled and added jobs Jsk. Gk can be formulated by:(4)Gk=Δt⋅|Jsk|+∑ji∈Jsk∪JrkTji,Rused0Tji,Rusedk−∑ji∈JrkTji,Rused0Tji,Rusedk−1,where Rused0, Rusedk−1, and Rusedk denote the used resource at τ0, τk−1, and τk, respectively; Rusedk=Rusedk−1+∑ji∈Jskri.Based on the definition of scheduling gain, we can define the relationship betweenGk and ATAT:Theorem 1.
Given a time period withN time slots for running a set of jobs J, the average value of accumulated scheduling gain (AASG), that is ∑k=1NGk/J, equals to the average turnaround time (ATAT) of J.Proof.
By definition, theATAT can be formulated as:(5)ATAT=∑ji∈Jtiw+tieJ=∑ji∈JtiwJ+∑ji∈JtieJ,where the average value of accumulated scheduling gain is defined as:(6)AASG=∑k=1NGkJ=Δt⋅∑k=1NJ−Jsk+∑ji∈Jsk∪JrkTji,Rused0Tji,Rusedk,(7)=∑k=1NΔt⋅|J−Jsk|+∑k=1N∑ji∈Jsk∪JrkΔt⋅Tji,Rused0Tji,Rusedk.
Comparing equations (5) and (7), we have(8)∑ji∈JtiwJ=∑k=1NΔt⋅J−Jsk.(9)∑ji∈JtieJ=∑k=1N∑ji∈Jsk∪JrkΔt⋅Tji,Rused0Tji,Rusedk.
Finally, we haveAASG=ATAT.We can see from Theorem1 that maximizing AASG equals to minimizing ATAT, one of the ultimate goal of our JS-BDPS problem, as defined in equation (2). In order to maximize AASG, we apply the greedy strategy in this article: at any time slot τk∈N, we try to find the scheduling job set Jsk that can maximize Gk.


### 4.4. Small Job First Policy

To minimize theATAT while achieving a better resource efficiency, the PAS algorithm tries to utilize as much resource as possible for running more jobs at any time slot τk on the one hand, and on the other hand it needs to compare the Gk values under different scheduling job sets and choose the scheduling sets with the best value of Gk. Given a fixed resource utilization ratio RURk at τk and a set of jobs Jk that needs to be scheduled, all possible number of scheduling job sets equals to 2Jk at the worst case. To reduce the searching space of this process, we propose a small job first policy, defined as follows:Theorem 2.
Given any two valid scheduling job setsJs1k and Js2k, if Js1k|≥|Js2k and ∑ji∈Js1kri=∑ji∈Js2kri, then we have Gs1k≥Gs2k.Proof.
According to the Definition1, the scheduling gain Gs1k over Js1k is defined as:(10)Gs1k=Δt⋅Js1k+∑ji∈Js1k∪JrkTji,Rused0Tji,Ruseds1k−∑ji∈JrkTji,Rused0Tji,Rusedk−1.
Similarly, theGs2k over Js2k is defined as:(11)Gs2k=Δt⋅Js2k+∑ji∈Js2k∪JrkTji,Rused0Tji,Ruseds2k−∑ji∈JrkTji,Rused0Tji,Rusedk−1.
Thus,Gs1k−Gs2k equals:(12)Gs1k−Gs2k=Δt⋅Js1k−Js2k+∑ji∈Js1k∪JrkTji,Rused0Tji,Ruseds1k−∑ji∈Js2k∪JrkTji,Rused0Tji,Ruseds1k,
Because∑ji∈Js1kri=∑ji∈Js2kri, we have(13)Ruseds1k=Ruseds2k.
Suppose that the performance prediction functions for all jobs share the similar shape, we have:(14)Tji,Rused0Tji,Rusedk−Tjp,Rused0Tjp,Rusedk≈0,∀ji,jp∈J.
Based on equations (12)–(14), we can substitute the second subtraction part of equation (12) with equation (14) and draw the final conclusion that the scheduling gain Gs1k over Js1k is greater than or equal to the scheduling gain Gs2k over Js2k.Theorem2 indicates that given a fixed resource r we can use for scheduling at any time slot τk, the scheduler should always schedule as many jobs as it can to minimize the ATAT. In another word, the small jobs with less resource requirements should be scheduled first, we call this strategy the small job first (SmJF) policy.


### 4.5. One-Step Lookahead

After usingSmJF policy, we can still obtain many scheduling job sets having the same size and the same Gk value. In this section, we propose a one-step lookahead (OSLA) policy that can find the scheduling job set(s) to achieve the possibly smaller ATAT value in the future, which is defined as follows:Definition 2.
One-step lookahead (OSLA) policy: Suppose the further scheduling policy is shortest job first (SJF) afterτk. According to the assumption, we have Js1k=Js2k, Gs1k=Gs2k. Let the following scheduling sequence for Js1k is S1 and for Js2k is S2, the average turnaround time under S1 is ATATs1S1 and under S2 is ATATs2S2, if ATATs1S1≤ATATs2S2, we should choose Js1 as our scheduling job set.Definition2 indicates that at any time slot τk, if two candidate job sets Js1 and Js2 have the same size and the same Gk value, our scheduler should lookahead for one step and apply a simple shortest job first (SJF) policy Js1 and Js2 to generate two scheduling sequences S1 and S2, respectively. Thereafter, we simulatively schedule Js1 under S1 and Js2 under S2 separately and estimate the ATATs1S1 and ATATs2S2. Finally, we choose the job set with the smaller ATAT value.The reason why we use the simpleOSLA policy here is because the following scheduling process is dynamic and too complex to predict—an approximate and greedy policy works well under this circumstance.


## 4.1. Performance-Aware Job Scheduling

The key idea of our performance-aware job scheduling (PAS) approach is to train a performance prediction model for each job according to their historical profiles and repeatedly assign resource to the job(s) with the lowest cost gain using the trained model. PAS also applies a heuristic one-step lookahead strategy to find potentially good scheduling policy.


## 4.2. Predicting Job Completion Time

A BDPS can support many data analytical jobs running on it simultaneously. We observe that the completion time of a specific jobj varies under different resource utilization ratio (RUR) of the BDPS. Suppose the completion time j is t when executing it exclusively, that is RUR0=0, on a BDPS. If RUR increases to RURk0<RURk<1 by adding more jobs to the BDPS, at each time slot Δt there are statistically 1−RURk⋅Δt share of processing time can be used for executing j. Furthermore, when RURk continues to grow, the swapping and scheduling time among different jobs becomes significantly large [42]. Based on these observations and the job complexity estimation method proposed in [43], we model the job completion time t and the resource utilization ratio (RUR) as a power-law function:(3)t=a⋅RURb+c,where a, b, and c are regression constants.The prediction model indicates that when resource utilization ratio grows, the completion time of a job grows in a power number. Such conclusion has been well verified in our experiments.


## 4.3. Scheduling Gain and ATAT

To schedule a set of jobs dynamically,PAS has to make decision periodically. More specifically, at each time point t, it needs to decide whether to schedule a set of jobs J from the queue (for decreasing the tw and te of all jobs in J), or to do nothing (for decreasing the execution time of all running jobs). To lucubrate the scheduling process, we discrete the time into many small and equal slots, that is Δt, and denote the kth time interval as τkτk=t0+k−1Δt,t0+kΔt, where t0 represents the start time of scheduling. Note that the discretized time slot Δt is a hyperparameter of PAS that is configurable by the administrator of the BDPS, and it defines the scheduling frequency of the PAS.Based on the discretized time slot, we can define thescheduling gain to quantify the gain by scheduling a set of jobs:Definition 1.
Scheduling gain atτk, denoted by Gk, is defined as the decrease of the waiting time for scheduling a subset of submitted jobs Jsk minus the increase of the execution time for the running jobs Jrk caused by the newly scheduled and added jobs Jsk. Gk can be formulated by:(4)Gk=Δt⋅|Jsk|+∑ji∈Jsk∪JrkTji,Rused0Tji,Rusedk−∑ji∈JrkTji,Rused0Tji,Rusedk−1,where Rused0, Rusedk−1, and Rusedk denote the used resource at τ0, τk−1, and τk, respectively; Rusedk=Rusedk−1+∑ji∈Jskri.Based on the definition of scheduling gain, we can define the relationship betweenGk and ATAT:Theorem 1.
Given a time period withN time slots for running a set of jobs J, the average value of accumulated scheduling gain (AASG), that is ∑k=1NGk/J, equals to the average turnaround time (ATAT) of J.Proof.
By definition, theATAT can be formulated as:(5)ATAT=∑ji∈Jtiw+tieJ=∑ji∈JtiwJ+∑ji∈JtieJ,where the average value of accumulated scheduling gain is defined as:(6)AASG=∑k=1NGkJ=Δt⋅∑k=1NJ−Jsk+∑ji∈Jsk∪JrkTji,Rused0Tji,Rusedk,(7)=∑k=1NΔt⋅|J−Jsk|+∑k=1N∑ji∈Jsk∪JrkΔt⋅Tji,Rused0Tji,Rusedk.
Comparing equations (5) and (7), we have(8)∑ji∈JtiwJ=∑k=1NΔt⋅J−Jsk.(9)∑ji∈JtieJ=∑k=1N∑ji∈Jsk∪JrkΔt⋅Tji,Rused0Tji,Rusedk.
Finally, we haveAASG=ATAT.We can see from Theorem1 that maximizing AASG equals to minimizing ATAT, one of the ultimate goal of our JS-BDPS problem, as defined in equation (2). In order to maximize AASG, we apply the greedy strategy in this article: at any time slot τk∈N, we try to find the scheduling job set Jsk that can maximize Gk.


## 4.4. Small Job First Policy

To minimize theATAT while achieving a better resource efficiency, the PAS algorithm tries to utilize as much resource as possible for running more jobs at any time slot τk on the one hand, and on the other hand it needs to compare the Gk values under different scheduling job sets and choose the scheduling sets with the best value of Gk. Given a fixed resource utilization ratio RURk at τk and a set of jobs Jk that needs to be scheduled, all possible number of scheduling job sets equals to 2Jk at the worst case. To reduce the searching space of this process, we propose a small job first policy, defined as follows:Theorem 2.
Given any two valid scheduling job setsJs1k and Js2k, if Js1k|≥|Js2k and ∑ji∈Js1kri=∑ji∈Js2kri, then we have Gs1k≥Gs2k.Proof.
According to the Definition1, the scheduling gain Gs1k over Js1k is defined as:(10)Gs1k=Δt⋅Js1k+∑ji∈Js1k∪JrkTji,Rused0Tji,Ruseds1k−∑ji∈JrkTji,Rused0Tji,Rusedk−1.
Similarly, theGs2k over Js2k is defined as:(11)Gs2k=Δt⋅Js2k+∑ji∈Js2k∪JrkTji,Rused0Tji,Ruseds2k−∑ji∈JrkTji,Rused0Tji,Rusedk−1.
Thus,Gs1k−Gs2k equals:(12)Gs1k−Gs2k=Δt⋅Js1k−Js2k+∑ji∈Js1k∪JrkTji,Rused0Tji,Ruseds1k−∑ji∈Js2k∪JrkTji,Rused0Tji,Ruseds1k,
Because∑ji∈Js1kri=∑ji∈Js2kri, we have(13)Ruseds1k=Ruseds2k.
Suppose that the performance prediction functions for all jobs share the similar shape, we have:(14)Tji,Rused0Tji,Rusedk−Tjp,Rused0Tjp,Rusedk≈0,∀ji,jp∈J.
Based on equations (12)–(14), we can substitute the second subtraction part of equation (12) with equation (14) and draw the final conclusion that the scheduling gain Gs1k over Js1k is greater than or equal to the scheduling gain Gs2k over Js2k.Theorem2 indicates that given a fixed resource r we can use for scheduling at any time slot τk, the scheduler should always schedule as many jobs as it can to minimize the ATAT. In another word, the small jobs with less resource requirements should be scheduled first, we call this strategy the small job first (SmJF) policy.


## 4.5. One-Step Lookahead

After usingSmJF policy, we can still obtain many scheduling job sets having the same size and the same Gk value. In this section, we propose a one-step lookahead (OSLA) policy that can find the scheduling job set(s) to achieve the possibly smaller ATAT value in the future, which is defined as follows:Definition 2.
One-step lookahead (OSLA) policy: Suppose the further scheduling policy is shortest job first (SJF) afterτk. According to the assumption, we have Js1k=Js2k, Gs1k=Gs2k. Let the following scheduling sequence for Js1k is S1 and for Js2k is S2, the average turnaround time under S1 is ATATs1S1 and under S2 is ATATs2S2, if ATATs1S1≤ATATs2S2, we should choose Js1 as our scheduling job set.Definition2 indicates that at any time slot τk, if two candidate job sets Js1 and Js2 have the same size and the same Gk value, our scheduler should lookahead for one step and apply a simple shortest job first (SJF) policy Js1 and Js2 to generate two scheduling sequences S1 and S2, respectively. Thereafter, we simulatively schedule Js1 under S1 and Js2 under S2 separately and estimate the ATATs1S1 and ATATs2S2. Finally, we choose the job set with the smaller ATAT value.The reason why we use the simpleOSLA policy here is because the following scheduling process is dynamic and too complex to predict—an approximate and greedy policy works well under this circumstance.


## 5. Algorithm Design and Implementation

Based on thesmall job first policy (SmJF) and the one-step lookahead (OSLA) policy, we can implement our PAS algorithm. The detailed process is specified in Algorithm 1.Algorithm 1:PAS algorithm: PAS Jk,Rusedk,L,U.
Require:Jk: the job set waiting for scheduling at τk; Rusedk: the used resource at τk; L,U: the resource boundary.Ensure:Jsk: scheduling job set at τk.(1
R′←U;(2
J′←∅;(3
whileR′≥Rusedkdo(4
J′←J′∪SmJFJk,R′;(5
R′←R′−ϵ;(6
end while(7
Jmax′←argmaxSGJ′;(8
Jsk←OSLAJk,Jmax′,Rusedk,U; returnJsk;PAS initially sets the available resource R′ to upper bound B (line 1) and starts to try different resource utilization values decreasingly (line 3 and 5). At each iteration, it applies the SmJF algorithm to select job sets according to R′ (line 4). After trying all possible R′, we choose the candidate scheduling job set(s) with the best scheduling gain value (line 7) and apply the OSLA algorithm to find the final scheduling job set Jsk (line 8).


### 5.1. Small Job First Algorithm

As Theorem2 shows, given a fixed resource utilization value R′, we should greedily choose jobs with smaller resource demand first to decrease the waiting time for these jobs without severely increasing the running time for the already running jobs. The SmJF algorithm is described in Algorithm 2.Algorithm 2:Small job first algorithm: SmJF(Jk, Rusedk, R′).
Require:Jk: the job set waiting for scheduling at τk; Rusedk: the used resource at τk; R′: the target resource utilization after scheduling.Ensure: All possible scheduling job sets Jsk(1
Jsk←∅;(2
Rc←Rusedk;(3
Sort the jobs inJk by their resource demands r in the increasing order to form a bin list B=B1,B2,…,Bl, in which each bin contains jobs with the same r;(4
fori=1:ldo(5
Rc←Rc+|Bi|⋅ri;(6
ifRc≤R′then(7
Jsk←Jsk∪Bi;(8
else(9
n←⌊R′−Rc−|Bi|⋅ri/ri⌋;(10
J′← all subset from Bi containing n jobs;(11
Jsk←Jsk∪J′; returnJsk;(12
end if(13
end for returnJsk;SmJF algorithm starts by sorting the jobs in Jk according to their resource demands r in the increasing order to generate a bin list B=B1,B2,…,Bl, in which each bin contains jobs with the same r (line 3). At each iteration step i, SmJF tries to add all jobs in the bin Bi first (line 5) and compares the updated resource utilization Rc with the target value R′ (line 6). If Rc is still smaller than or equal to R′, we can now safely add all jobs in Bi to the scheduling job set Jsk (line 7) and move to the next iteration. If Rc is already greater than R′, we have to release the resource possessed by jobs in Bi first and recalculate the desirable number of jobs n (line 9). We should then enumerate all possible subsets of Bi containing n jobs (line 10). The algorithm ends after adding subsets of some Bi in B to Jsk (line 11), or finally adding all jobs in Jk to Jsk (line 13).


### 5.2. One-Step Lookahead Algorithm

Given each candidate scheduling setJi in Jck, the OSLA algorithm first appends all jobs in Ji to the scheduling sequence S (line 2–5). Thereafter, other jobs in Jck are appended to S according to the shortest job first (SJF) policy (line 6). Once S is complete, we call the simulation algorithm to estimate the average turnaround time ATATi of S starting with Ji (line 7). Finally, the candidate scheduling set Ji with the smallest ATAT values are chosen and returned (line 9). The whole process is shown in Algorithm 3.Algorithm 3:One-step lookahead algorithm: OSLA(Jk, Jck, Rusedk, U).
Require:Jk: the job set waiting for scheduling at τk; Jck: a group of candidate scheduling sets; Rusedk: the used resource at τk; U: the upper bound of resource utilization.Ensure:Jsk: the scheduling set.(1
Ordered job scheduling sequence S←∅;(2
forJi∈Jckdo(3
forjp∈Jido(4
S←S∪jp;(5
end for(6
Append jobs inJk−Ji to S according to the shortest job first (SJF) policy;(7
ATATi←SimulationJk,Ji,Rusedk,S,U;(8
end for(9
Jsk←argminJi∈JckATATi; returnJsk;Given a possible job scheduling sequenceS, the simulation algorithm estimates the ATAT of S by greedily choosing jobs from the beginning of the S, putting them into the running job set, and updating their waiting and execution time. The detailed working process is shown in Algorithm 4.Algorithm 4:Job scheduling simulation algorithm: Simulation (Jk, Jc, Rusedk, S, U).
Require:Jk: the job set waiting for scheduling at τk; Jc: a group of candidate scheduling sets; Rusedk: the used resource at τk; S the scheduling sequence under SJF policy, U: the upper bound of resource utilization.Ensure:ATAT for Jk under the scheduling sequence S.(1
S′←S;(2
The running job set:Jr←∅;(3
The current used resource:R←Rusedk;(4
The turnaround time for a jobji: TATi←0;(5
whileS′≠∅do(6
forji∈Sdo(7
ifR+ri≤Uthen(8
S′.removeji;(9
Jr.appendji;(10
R←R+ri;(11
forjp∈Jrdo(12
tp.remain←tp.remain⋅Tjp,R/Tjp,R−ri;(13
end for(14
else(15
jfast←argmixji∈Jrti.remain;(16
forji∈Jrdo(17
ti.remain←ti.remain−tfast.remain;(18
TATi←TATi+tfast.remain;(19
ti.remain←ti.remain⋅Tji,R−rfast/Tji,R;(20
end for(21
TATi←TATi+tfast.remain;(22
R←R−rfast;(23
Jr.removejfast;(24
end if(25
end for(26
end while(27
ATAT←1/Jk⋅∑ji∈JkTATi; returnATAT;


### 5.3. Timeout Mechanism

One potential issue of ourPAS algorithm is that the long-running job with large resource demand will suffer from the starvation problem. To deal with it, we set a timeout value, that is the maximal waiting deadline, to each job. Once the waiting time of a job exceeds the timeout, it will be executed first without any delay.We can set the timeout value as the makespan of the current scheduling job setJsk; this ensures that the long-running job(s) with large resource demand has the opportunity to execute after all other jobs in Jsk complete their execution. The timeout value for each job can be set at the first time in the simulation algorithm, in which the job scheduling sequence S is given and the makespan value is able to estimate.


### 5.4. Implementation on Hadoop YARN

We incorporatePAS into Hadoop YARN (2.6.0) by implementing an independent PAS algorithm module and adding a PAS scheduler plugin to YARN. The implementation detail is shown in Figure 1.Figure 1
PAS implementation on Hadoop YARN.As shown in Figure1, PAS consists of two modules. The first module is called PAS algorithm, which is implemented by Python 3.5. At each time slot τk, PAS algorithm first receives the necessary scheduling information, for example Rusedk, Jk, etc., from the PAS scheduler, and then calculate the suggested scheduling job sets by calling the proposed PAS algorithm with the scheduling information. The communication between PAS algorithm and PAS scheduler is implemented by TCP protocol.We have implemented thePAS scheduler as a customized plugin of YARN. PAS scheduler focuses on sending the scheduling information periodically to the PAS algorithm and allocating resource for the scheduling job set returned by the PAS algorithm.More specifically, at each time slotτk, PAS scheduler needs to determine whether there still has more resource to allocate in comparison with U, the upper bound of resource utilization:(i)
If so,PAS scheduler sends the information including current used resource Rusedk and the job set waiting for scheduling Jk to the PAS algorithm and requests for the scheduling job set. Once received the Jsk, PAS scheduler will attach all the jobs in Jsk to the execution queue of YARN.(ii)
If not,PAS scheduler will do nothing and wait for the next time slot.


## 5.1. Small Job First Algorithm

As Theorem2 shows, given a fixed resource utilization value R′, we should greedily choose jobs with smaller resource demand first to decrease the waiting time for these jobs without severely increasing the running time for the already running jobs. The SmJF algorithm is described in Algorithm 2.Algorithm 2:Small job first algorithm: SmJF(Jk, Rusedk, R′).
Require:Jk: the job set waiting for scheduling at τk; Rusedk: the used resource at τk; R′: the target resource utilization after scheduling.Ensure: All possible scheduling job sets Jsk(1
Jsk←∅;(2
Rc←Rusedk;(3
Sort the jobs inJk by their resource demands r in the increasing order to form a bin list B=B1,B2,…,Bl, in which each bin contains jobs with the same r;(4
fori=1:ldo(5
Rc←Rc+|Bi|⋅ri;(6
ifRc≤R′then(7
Jsk←Jsk∪Bi;(8
else(9
n←⌊R′−Rc−|Bi|⋅ri/ri⌋;(10
J′← all subset from Bi containing n jobs;(11
Jsk←Jsk∪J′; returnJsk;(12
end if(13
end for returnJsk;SmJF algorithm starts by sorting the jobs in Jk according to their resource demands r in the increasing order to generate a bin list B=B1,B2,…,Bl, in which each bin contains jobs with the same r (line 3). At each iteration step i, SmJF tries to add all jobs in the bin Bi first (line 5) and compares the updated resource utilization Rc with the target value R′ (line 6). If Rc is still smaller than or equal to R′, we can now safely add all jobs in Bi to the scheduling job set Jsk (line 7) and move to the next iteration. If Rc is already greater than R′, we have to release the resource possessed by jobs in Bi first and recalculate the desirable number of jobs n (line 9). We should then enumerate all possible subsets of Bi containing n jobs (line 10). The algorithm ends after adding subsets of some Bi in B to Jsk (line 11), or finally adding all jobs in Jk to Jsk (line 13).


## 5.2. One-Step Lookahead Algorithm

Given each candidate scheduling setJi in Jck, the OSLA algorithm first appends all jobs in Ji to the scheduling sequence S (line 2–5). Thereafter, other jobs in Jck are appended to S according to the shortest job first (SJF) policy (line 6). Once S is complete, we call the simulation algorithm to estimate the average turnaround time ATATi of S starting with Ji (line 7). Finally, the candidate scheduling set Ji with the smallest ATAT values are chosen and returned (line 9). The whole process is shown in Algorithm 3.Algorithm 3:One-step lookahead algorithm: OSLA(Jk, Jck, Rusedk, U).
Require:Jk: the job set waiting for scheduling at τk; Jck: a group of candidate scheduling sets; Rusedk: the used resource at τk; U: the upper bound of resource utilization.Ensure:Jsk: the scheduling set.(1
Ordered job scheduling sequence S←∅;(2
forJi∈Jckdo(3
forjp∈Jido(4
S←S∪jp;(5
end for(6
Append jobs inJk−Ji to S according to the shortest job first (SJF) policy;(7
ATATi←SimulationJk,Ji,Rusedk,S,U;(8
end for(9
Jsk←argminJi∈JckATATi; returnJsk;Given a possible job scheduling sequenceS, the simulation algorithm estimates the ATAT of S by greedily choosing jobs from the beginning of the S, putting them into the running job set, and updating their waiting and execution time. The detailed working process is shown in Algorithm 4.Algorithm 4:Job scheduling simulation algorithm: Simulation (Jk, Jc, Rusedk, S, U).
Require:Jk: the job set waiting for scheduling at τk; Jc: a group of candidate scheduling sets; Rusedk: the used resource at τk; S the scheduling sequence under SJF policy, U: the upper bound of resource utilization.Ensure:ATAT for Jk under the scheduling sequence S.(1
S′←S;(2
The running job set:Jr←∅;(3
The current used resource:R←Rusedk;(4
The turnaround time for a jobji: TATi←0;(5
whileS′≠∅do(6
forji∈Sdo(7
ifR+ri≤Uthen(8
S′.removeji;(9
Jr.appendji;(10
R←R+ri;(11
forjp∈Jrdo(12
tp.remain←tp.remain⋅Tjp,R/Tjp,R−ri;(13
end for(14
else(15
jfast←argmixji∈Jrti.remain;(16
forji∈Jrdo(17
ti.remain←ti.remain−tfast.remain;(18
TATi←TATi+tfast.remain;(19
ti.remain←ti.remain⋅Tji,R−rfast/Tji,R;(20
end for(21
TATi←TATi+tfast.remain;(22
R←R−rfast;(23
Jr.removejfast;(24
end if(25
end for(26
end while(27
ATAT←1/Jk⋅∑ji∈JkTATi; returnATAT;


## 5.3. Timeout Mechanism

One potential issue of ourPAS algorithm is that the long-running job with large resource demand will suffer from the starvation problem. To deal with it, we set a timeout value, that is the maximal waiting deadline, to each job. Once the waiting time of a job exceeds the timeout, it will be executed first without any delay.We can set the timeout value as the makespan of the current scheduling job setJsk; this ensures that the long-running job(s) with large resource demand has the opportunity to execute after all other jobs in Jsk complete their execution. The timeout value for each job can be set at the first time in the simulation algorithm, in which the job scheduling sequence S is given and the makespan value is able to estimate.


## 5.4. Implementation on Hadoop YARN

We incorporatePAS into Hadoop YARN (2.6.0) by implementing an independent PAS algorithm module and adding a PAS scheduler plugin to YARN. The implementation detail is shown in Figure 1.Figure 1
PAS implementation on Hadoop YARN.As shown in Figure1, PAS consists of two modules. The first module is called PAS algorithm, which is implemented by Python 3.5. At each time slot τk, PAS algorithm first receives the necessary scheduling information, for example Rusedk, Jk, etc., from the PAS scheduler, and then calculate the suggested scheduling job sets by calling the proposed PAS algorithm with the scheduling information. The communication between PAS algorithm and PAS scheduler is implemented by TCP protocol.We have implemented thePAS scheduler as a customized plugin of YARN. PAS scheduler focuses on sending the scheduling information periodically to the PAS algorithm and allocating resource for the scheduling job set returned by the PAS algorithm.More specifically, at each time slotτk, PAS scheduler needs to determine whether there still has more resource to allocate in comparison with U, the upper bound of resource utilization:(i)
If so,PAS scheduler sends the information including current used resource Rusedk and the job set waiting for scheduling Jk to the PAS algorithm and requests for the scheduling job set. Once received the Jsk, PAS scheduler will attach all the jobs in Jsk to the execution queue of YARN.(ii)
If not,PAS scheduler will do nothing and wait for the next time slot.


## 6. Experiments

### 6.1. Experimental Setup

We have implemented our approach and conducted extensive experiments under different testing scenarios. The source code and dataset can be found in an anonymous website:https://github.com/anon4review/pas. In this section, we first describe our experiment setup and then report the experimental results to prove the efficiency and effectiveness of the proposed approach.


#### 6.1.1. Running Environment

We use Hadoop YARN (2.6.0) on Spark (1.6.0) and conduct our experiments on a local cluster of five physical servers. Each server is equipped with two 8-core Intel XeonE5-2650v2 2.6 GHz processors, 256GiB RAM, 1.5 TB disk, and running CentOS 6.0, Java 1.7.0_55 and Python 3.5. All of servers are connected via a high-speed 1.5 Gbps LAN. To avoid interference and comply with the actual deployment, we run the YARN on Spark, the workload generators, and thePAS algorithm on different physical servers at each experiment.


#### 6.1.2. Baseline Algorithms

To evaluate the performance ofPAS, we compare it with four state-of-the-art algorithms, namely FIFO [38], AHP [44], SJF [45], and DRF [24]. We provide a brief description for each algorithm as follows and compare the scheduling goals for the five algorithms in Table 1:Table 1
Scheduling goals for the five algorithms.
AlgorithmResourcePerformanceFairnessPriorityFIFO✓AHP✓✓SJFDRF✓✓PAS✓✓FIFO sorts all jobs in the order of submission (first in, first out), and it is the default scheduler of the YARN.AHP is an improvement in priority-based job scheduling algorithm in cloud computing, which is based on multiple criteria and multiple attribute decision-making model.SJF sorts all jobs in the order of execution time, the shortest will be sorted first.DRF is an extension of classic fair scheduling [26] for multiple types of resource. DRF determines CPU and memory resource shares based on the availability of those resources and the job requirements.InPAS algorithm, we set the lower and upper bounds of CPU and memory utilization ratios to 0.5,0.9 and 0.75,1, respectively.


#### 6.1.3. Workloads

In our experiment, we use HiBench, a well-known big data benchmark, to generate Spark workloads. More specifically, we choose 10 different workloads with three micro benchmarks—SparkPi (SP), WordCount (WC), and Sort (ST)—and seven machine learning workloads— support vector machine (SVM), Bayesian classification (BC), K-means clustering (KM), gradient boosting trees (GBT), linear regression (LR), principal component analysis (PCA), and random forest (RF). For each workload, we have two different scales of datasets and three different two-dimensional resource demands (e.g.,2C,4GB means two CPU cores and 4 GB memory), as shown in Table 2.Table 2
Ten different candidate workloads in our experiment.
#Job nameScaleResource demand1SPTiny, large2C,4GB,3C,6GB,4C,8GB2WCTiny, large2C,4GB,3C,6GB,4C,8GB3STTiny, large2C,4GB,3C,6GB,4C,8GB4SVMTiny, small2C,4GB,3C,6GB,4C,8GB5BCTiny, small2C,4GB,3C,6GB,4C,8GB6KMTiny, small2C,4GB,3C,6GB,4C,8GB7GBTTiny, small2C,4GB,3C,6GB,4C,8GB8LRTiny, small2C,4GB,3C,6GB,4C,8GB9PCATiny, small2C,4GB,3C,6GB,4C,8GB10RFTiny, large2C,4GB,3C,6GB,4C,8GBWe construct six groups of jobs for scheduling with different job numbers, that is 15, 30, 45, 60, 75, and 90, by randomly choosing these jobs from the candidate workloads listed in Table2. For each testing group and a scheduling algorithm, we conduct 10 independent runs and record the results separately.


#### 6.1.4. Evaluation Metrics

We consider three well-known metrics in our experiments for performance evaluation, namely average turnaround time (ATAT), resource utilization ratio (RUR), and makespan (MS).Average turnaround time (ATAT). ATAT can measure the performance of a scheduler from the user’s perspective. The detailed definition of ATAT can be found in Section 3. Here we define the ATAT improvement in an algorithm over the baseline algorithm in comparison as:(15)Impbaseline=ATATbaseline−ATATATATbaseline×100%,where ATATbaseline is the baseline ATAT, and ATAT is that of the algorithm being evaluated.Resource utilization Ratio (RUR). RUR measures the resource efficiency of a BDPS from the service providers’ perspective. The detailed definition of RUR can also be found in Section 3.Makespan (MS). Makespan defines the time difference between the start and finish of a sequence of jobs. It measures the resource efficiency of a BDPS from the service providers’ perspective [2].


### 6.2. Experimental Results

#### 6.2.1. Estimation of Job Completion Time under Different RUR

We choose three workloads, that is a CPU-intensive workload (SP), a memory-intensive workload (WC), and a machine learning workload (SVM), to verify the prediction model for job completion time. We first run each of these three workloads on the experimental environment with a series ofRUR. For each workload, we perform regressions on the estimated expressions, that is equation (3), and provide in Table 3 the results of the root mean square error (RMSE), normalized root mean square error (NRMSE), and R-square R2, which clearly indicate that a high level of goodness-of-fit is achieved in these regressions.Table 3
Regression errors in performance estimation.
Job nameRMSENRMSE (%)R2SP0.70132.060.9730WC2.59116.180.7496SVM1.72582.550.9692For a better illustration, we plot three fitted curves of these three workloads in Figure2. By fitting the data points, we find that the job execution time is an approximate cubic function under RUR. With these fitted curves, we are able to provide a good estimate of the completion time of a given workload on the different RUR values.Figure 2
Boxplots ofATAT values under five algorithms over 10 independent runs.


#### 6.2.2. ATAT

Table4 shows the ATAT (ms) values of five different scheduling algorithms. As expected, our algorithm outperforms the default FIFO scheduler of YARN by 25.3% to 42.0%, Furthermore, PAS outperforms all other three algorithms: 13.6%–36.4% improvement over AHP, 10.2%–28.8% improvement over SJF, and 4.1%–12.1% improvement over DRF.Table 4
ATAT performance of five algorithms.
#PASAHPSJFFIFODRF1566234.488424.893058.13103481.3733183085147.7106808.894857.4125558.491477.466745109096.1129421.9130781.3174098.5118586.660138780.8205562.2161795.8239613.8157879.475130243.3150723177455.8174463.3135836.690123840.5194835147746.5210194.8136140.944Figure3 shows the boxplots of the ATAT performance under six groups of job sets and five algorithms over 10 independent runs. In these boxplots, the bottom and top of the box are the first and third quartiles, the bands inside the boxes represent robust estimates of the uncertainty about the medians for box-to-box comparison. The ends of the whiskers represent possible alternative values and the symbol “+” denotes outliers. As shown by the ATAT boxplots, PAS has lower ATAT values than other four scheduling algorithms on all of the six job sets, which indicates the superiority and the robustness of our PAS algorithm.Figure 3
Boxplots ofATAT values under six groups of job sets and five algorithms over 10 independent runs.


#### 6.2.3. RUR

For a better illustration, we plot the CPU and memory utilization ratio for running a group of 45 jobs under five scheduling algorithms in Figures4 and 5, respectively. We can see from Figure 4 that the CPU utilization ratios under PAS algorithm (red line) are more stable over the time than others: the CPU utilization ratios are between 50% and 90% (our predefined boundary) during the 69.04% of time slots. Figure 5 shows that the difference of memory utilization ratios among these algorithms is not significative, but the memory utilization ratios under PAS can be still kept in between 75% and 100% during 76.19% of time slots. Note that the resource utilization jitter exists under all scheduling algorithms because there might be a delay between the release of resource for already finished jobs and the allocation of resource for a set of newly added jobs.Figure 4
CPU trends w.r.t. the time slots.Figure 5
Memory trends w.r.t. the time slots.


#### 6.2.4. MS

Finally, Figure6 shows the makespan values for six groups of workloads under five scheduling algorithms. We can see from Figure 6 that PAS outperforms all other three algorithms in term of makespan, with the exception of DRF algorithm. The reason why DRF has a good performance in makespan is probably because it allows schedulers to take into account the heterogeneous resource demands, leading to both fairer allocation of resources and higher utilization [24].Figure 6
Makespan values under five algorithms over six groups of workloads.


## 6.1. Experimental Setup

We have implemented our approach and conducted extensive experiments under different testing scenarios. The source code and dataset can be found in an anonymous website:https://github.com/anon4review/pas. In this section, we first describe our experiment setup and then report the experimental results to prove the efficiency and effectiveness of the proposed approach.


### 6.1.1. Running Environment

We use Hadoop YARN (2.6.0) on Spark (1.6.0) and conduct our experiments on a local cluster of five physical servers. Each server is equipped with two 8-core Intel XeonE5-2650v2 2.6 GHz processors, 256GiB RAM, 1.5 TB disk, and running CentOS 6.0, Java 1.7.0_55 and Python 3.5. All of servers are connected via a high-speed 1.5 Gbps LAN. To avoid interference and comply with the actual deployment, we run the YARN on Spark, the workload generators, and thePAS algorithm on different physical servers at each experiment.


### 6.1.2. Baseline Algorithms

To evaluate the performance ofPAS, we compare it with four state-of-the-art algorithms, namely FIFO [38], AHP [44], SJF [45], and DRF [24]. We provide a brief description for each algorithm as follows and compare the scheduling goals for the five algorithms in Table 1:Table 1
Scheduling goals for the five algorithms.
AlgorithmResourcePerformanceFairnessPriorityFIFO✓AHP✓✓SJFDRF✓✓PAS✓✓FIFO sorts all jobs in the order of submission (first in, first out), and it is the default scheduler of the YARN.AHP is an improvement in priority-based job scheduling algorithm in cloud computing, which is based on multiple criteria and multiple attribute decision-making model.SJF sorts all jobs in the order of execution time, the shortest will be sorted first.DRF is an extension of classic fair scheduling [26] for multiple types of resource. DRF determines CPU and memory resource shares based on the availability of those resources and the job requirements.InPAS algorithm, we set the lower and upper bounds of CPU and memory utilization ratios to 0.5,0.9 and 0.75,1, respectively.


### 6.1.3. Workloads

In our experiment, we use HiBench, a well-known big data benchmark, to generate Spark workloads. More specifically, we choose 10 different workloads with three micro benchmarks—SparkPi (SP), WordCount (WC), and Sort (ST)—and seven machine learning workloads— support vector machine (SVM), Bayesian classification (BC), K-means clustering (KM), gradient boosting trees (GBT), linear regression (LR), principal component analysis (PCA), and random forest (RF). For each workload, we have two different scales of datasets and three different two-dimensional resource demands (e.g.,2C,4GB means two CPU cores and 4 GB memory), as shown in Table 2.Table 2
Ten different candidate workloads in our experiment.
#Job nameScaleResource demand1SPTiny, large2C,4GB,3C,6GB,4C,8GB2WCTiny, large2C,4GB,3C,6GB,4C,8GB3STTiny, large2C,4GB,3C,6GB,4C,8GB4SVMTiny, small2C,4GB,3C,6GB,4C,8GB5BCTiny, small2C,4GB,3C,6GB,4C,8GB6KMTiny, small2C,4GB,3C,6GB,4C,8GB7GBTTiny, small2C,4GB,3C,6GB,4C,8GB8LRTiny, small2C,4GB,3C,6GB,4C,8GB9PCATiny, small2C,4GB,3C,6GB,4C,8GB10RFTiny, large2C,4GB,3C,6GB,4C,8GBWe construct six groups of jobs for scheduling with different job numbers, that is 15, 30, 45, 60, 75, and 90, by randomly choosing these jobs from the candidate workloads listed in Table2. For each testing group and a scheduling algorithm, we conduct 10 independent runs and record the results separately.


### 6.1.4. Evaluation Metrics

We consider three well-known metrics in our experiments for performance evaluation, namely average turnaround time (ATAT), resource utilization ratio (RUR), and makespan (MS).Average turnaround time (ATAT). ATAT can measure the performance of a scheduler from the user’s perspective. The detailed definition of ATAT can be found in Section 3. Here we define the ATAT improvement in an algorithm over the baseline algorithm in comparison as:(15)Impbaseline=ATATbaseline−ATATATATbaseline×100%,where ATATbaseline is the baseline ATAT, and ATAT is that of the algorithm being evaluated.Resource utilization Ratio (RUR). RUR measures the resource efficiency of a BDPS from the service providers’ perspective. The detailed definition of RUR can also be found in Section 3.Makespan (MS). Makespan defines the time difference between the start and finish of a sequence of jobs. It measures the resource efficiency of a BDPS from the service providers’ perspective [2].


## 6.1.1. Running Environment

We use Hadoop YARN (2.6.0) on Spark (1.6.0) and conduct our experiments on a local cluster of five physical servers. Each server is equipped with two 8-core Intel XeonE5-2650v2 2.6 GHz processors, 256GiB RAM, 1.5 TB disk, and running CentOS 6.0, Java 1.7.0_55 and Python 3.5. All of servers are connected via a high-speed 1.5 Gbps LAN. To avoid interference and comply with the actual deployment, we run the YARN on Spark, the workload generators, and thePAS algorithm on different physical servers at each experiment.


## 6.1.2. Baseline Algorithms

To evaluate the performance ofPAS, we compare it with four state-of-the-art algorithms, namely FIFO [38], AHP [44], SJF [45], and DRF [24]. We provide a brief description for each algorithm as follows and compare the scheduling goals for the five algorithms in Table 1:Table 1
Scheduling goals for the five algorithms.
AlgorithmResourcePerformanceFairnessPriorityFIFO✓AHP✓✓SJFDRF✓✓PAS✓✓FIFO sorts all jobs in the order of submission (first in, first out), and it is the default scheduler of the YARN.AHP is an improvement in priority-based job scheduling algorithm in cloud computing, which is based on multiple criteria and multiple attribute decision-making model.SJF sorts all jobs in the order of execution time, the shortest will be sorted first.DRF is an extension of classic fair scheduling [26] for multiple types of resource. DRF determines CPU and memory resource shares based on the availability of those resources and the job requirements.InPAS algorithm, we set the lower and upper bounds of CPU and memory utilization ratios to 0.5,0.9 and 0.75,1, respectively.


## 6.1.3. Workloads

In our experiment, we use HiBench, a well-known big data benchmark, to generate Spark workloads. More specifically, we choose 10 different workloads with three micro benchmarks—SparkPi (SP), WordCount (WC), and Sort (ST)—and seven machine learning workloads— support vector machine (SVM), Bayesian classification (BC), K-means clustering (KM), gradient boosting trees (GBT), linear regression (LR), principal component analysis (PCA), and random forest (RF). For each workload, we have two different scales of datasets and three different two-dimensional resource demands (e.g.,2C,4GB means two CPU cores and 4 GB memory), as shown in Table 2.Table 2
Ten different candidate workloads in our experiment.
#Job nameScaleResource demand1SPTiny, large2C,4GB,3C,6GB,4C,8GB2WCTiny, large2C,4GB,3C,6GB,4C,8GB3STTiny, large2C,4GB,3C,6GB,4C,8GB4SVMTiny, small2C,4GB,3C,6GB,4C,8GB5BCTiny, small2C,4GB,3C,6GB,4C,8GB6KMTiny, small2C,4GB,3C,6GB,4C,8GB7GBTTiny, small2C,4GB,3C,6GB,4C,8GB8LRTiny, small2C,4GB,3C,6GB,4C,8GB9PCATiny, small2C,4GB,3C,6GB,4C,8GB10RFTiny, large2C,4GB,3C,6GB,4C,8GBWe construct six groups of jobs for scheduling with different job numbers, that is 15, 30, 45, 60, 75, and 90, by randomly choosing these jobs from the candidate workloads listed in Table2. For each testing group and a scheduling algorithm, we conduct 10 independent runs and record the results separately.


## 6.1.4. Evaluation Metrics

We consider three well-known metrics in our experiments for performance evaluation, namely average turnaround time (ATAT), resource utilization ratio (RUR), and makespan (MS).Average turnaround time (ATAT). ATAT can measure the performance of a scheduler from the user’s perspective. The detailed definition of ATAT can be found in Section 3. Here we define the ATAT improvement in an algorithm over the baseline algorithm in comparison as:(15)Impbaseline=ATATbaseline−ATATATATbaseline×100%,where ATATbaseline is the baseline ATAT, and ATAT is that of the algorithm being evaluated.Resource utilization Ratio (RUR). RUR measures the resource efficiency of a BDPS from the service providers’ perspective. The detailed definition of RUR can also be found in Section 3.Makespan (MS). Makespan defines the time difference between the start and finish of a sequence of jobs. It measures the resource efficiency of a BDPS from the service providers’ perspective [2].


## 6.2. Experimental Results

### 6.2.1. Estimation of Job Completion Time under Different RUR

We choose three workloads, that is a CPU-intensive workload (SP), a memory-intensive workload (WC), and a machine learning workload (SVM), to verify the prediction model for job completion time. We first run each of these three workloads on the experimental environment with a series ofRUR. For each workload, we perform regressions on the estimated expressions, that is equation (3), and provide in Table 3 the results of the root mean square error (RMSE), normalized root mean square error (NRMSE), and R-square R2, which clearly indicate that a high level of goodness-of-fit is achieved in these regressions.Table 3
Regression errors in performance estimation.
Job nameRMSENRMSE (%)R2SP0.70132.060.9730WC2.59116.180.7496SVM1.72582.550.9692For a better illustration, we plot three fitted curves of these three workloads in Figure2. By fitting the data points, we find that the job execution time is an approximate cubic function under RUR. With these fitted curves, we are able to provide a good estimate of the completion time of a given workload on the different RUR values.Figure 2
Boxplots ofATAT values under five algorithms over 10 independent runs.


### 6.2.2. ATAT

Table4 shows the ATAT (ms) values of five different scheduling algorithms. As expected, our algorithm outperforms the default FIFO scheduler of YARN by 25.3% to 42.0%, Furthermore, PAS outperforms all other three algorithms: 13.6%–36.4% improvement over AHP, 10.2%–28.8% improvement over SJF, and 4.1%–12.1% improvement over DRF.Table 4
ATAT performance of five algorithms.
#PASAHPSJFFIFODRF1566234.488424.893058.13103481.3733183085147.7106808.894857.4125558.491477.466745109096.1129421.9130781.3174098.5118586.660138780.8205562.2161795.8239613.8157879.475130243.3150723177455.8174463.3135836.690123840.5194835147746.5210194.8136140.944Figure3 shows the boxplots of the ATAT performance under six groups of job sets and five algorithms over 10 independent runs. In these boxplots, the bottom and top of the box are the first and third quartiles, the bands inside the boxes represent robust estimates of the uncertainty about the medians for box-to-box comparison. The ends of the whiskers represent possible alternative values and the symbol “+” denotes outliers. As shown by the ATAT boxplots, PAS has lower ATAT values than other four scheduling algorithms on all of the six job sets, which indicates the superiority and the robustness of our PAS algorithm.Figure 3
Boxplots ofATAT values under six groups of job sets and five algorithms over 10 independent runs.


### 6.2.3. RUR

For a better illustration, we plot the CPU and memory utilization ratio for running a group of 45 jobs under five scheduling algorithms in Figures4 and 5, respectively. We can see from Figure 4 that the CPU utilization ratios under PAS algorithm (red line) are more stable over the time than others: the CPU utilization ratios are between 50% and 90% (our predefined boundary) during the 69.04% of time slots. Figure 5 shows that the difference of memory utilization ratios among these algorithms is not significative, but the memory utilization ratios under PAS can be still kept in between 75% and 100% during 76.19% of time slots. Note that the resource utilization jitter exists under all scheduling algorithms because there might be a delay between the release of resource for already finished jobs and the allocation of resource for a set of newly added jobs.Figure 4
CPU trends w.r.t. the time slots.Figure 5
Memory trends w.r.t. the time slots.


### 6.2.4. MS

Finally, Figure6 shows the makespan values for six groups of workloads under five scheduling algorithms. We can see from Figure 6 that PAS outperforms all other three algorithms in term of makespan, with the exception of DRF algorithm. The reason why DRF has a good performance in makespan is probably because it allows schedulers to take into account the heterogeneous resource demands, leading to both fairer allocation of resources and higher utilization [24].Figure 6
Makespan values under five algorithms over six groups of workloads.


## 6.2.1. Estimation of Job Completion Time under Different RUR

We choose three workloads, that is a CPU-intensive workload (SP), a memory-intensive workload (WC), and a machine learning workload (SVM), to verify the prediction model for job completion time. We first run each of these three workloads on the experimental environment with a series ofRUR. For each workload, we perform regressions on the estimated expressions, that is equation (3), and provide in Table 3 the results of the root mean square error (RMSE), normalized root mean square error (NRMSE), and R-square R2, which clearly indicate that a high level of goodness-of-fit is achieved in these regressions.Table 3
Regression errors in performance estimation.
Job nameRMSENRMSE (%)R2SP0.70132.060.9730WC2.59116.180.7496SVM1.72582.550.9692For a better illustration, we plot three fitted curves of these three workloads in Figure2. By fitting the data points, we find that the job execution time is an approximate cubic function under RUR. With these fitted curves, we are able to provide a good estimate of the completion time of a given workload on the different RUR values.Figure 2
Boxplots ofATAT values under five algorithms over 10 independent runs.


## 6.2.2. ATAT

Table4 shows the ATAT (ms) values of five different scheduling algorithms. As expected, our algorithm outperforms the default FIFO scheduler of YARN by 25.3% to 42.0%, Furthermore, PAS outperforms all other three algorithms: 13.6%–36.4% improvement over AHP, 10.2%–28.8% improvement over SJF, and 4.1%–12.1% improvement over DRF.Table 4
ATAT performance of five algorithms.
#PASAHPSJFFIFODRF1566234.488424.893058.13103481.3733183085147.7106808.894857.4125558.491477.466745109096.1129421.9130781.3174098.5118586.660138780.8205562.2161795.8239613.8157879.475130243.3150723177455.8174463.3135836.690123840.5194835147746.5210194.8136140.944Figure3 shows the boxplots of the ATAT performance under six groups of job sets and five algorithms over 10 independent runs. In these boxplots, the bottom and top of the box are the first and third quartiles, the bands inside the boxes represent robust estimates of the uncertainty about the medians for box-to-box comparison. The ends of the whiskers represent possible alternative values and the symbol “+” denotes outliers. As shown by the ATAT boxplots, PAS has lower ATAT values than other four scheduling algorithms on all of the six job sets, which indicates the superiority and the robustness of our PAS algorithm.Figure 3
Boxplots ofATAT values under six groups of job sets and five algorithms over 10 independent runs.


## 6.2.3. RUR

For a better illustration, we plot the CPU and memory utilization ratio for running a group of 45 jobs under five scheduling algorithms in Figures4 and 5, respectively. We can see from Figure 4 that the CPU utilization ratios under PAS algorithm (red line) are more stable over the time than others: the CPU utilization ratios are between 50% and 90% (our predefined boundary) during the 69.04% of time slots. Figure 5 shows that the difference of memory utilization ratios among these algorithms is not significative, but the memory utilization ratios under PAS can be still kept in between 75% and 100% during 76.19% of time slots. Note that the resource utilization jitter exists under all scheduling algorithms because there might be a delay between the release of resource for already finished jobs and the allocation of resource for a set of newly added jobs.Figure 4
CPU trends w.r.t. the time slots.Figure 5
Memory trends w.r.t. the time slots.


## 6.2.4. MS

Finally, Figure6 shows the makespan values for six groups of workloads under five scheduling algorithms. We can see from Figure 6 that PAS outperforms all other three algorithms in term of makespan, with the exception of DRF algorithm. The reason why DRF has a good performance in makespan is probably because it allows schedulers to take into account the heterogeneous resource demands, leading to both fairer allocation of resources and higher utilization [24].Figure 6
Makespan values under five algorithms over six groups of workloads.


## 7. Conclusion and Future Work

In this article, we proposePAS algorithm to optimize average turnaround time and resource efficiency simultaneously. PAS scheduler constructs a performance prediction models for an accurate estimation of completion time of big data analytical jobs, it then dynamically schedules multiple jobs concurrently using different policies based on the prediction model and employs a greedy and a one-step lookahead strategy to opportunistically improve the average job performance for fast job completion while still achieving good enough resource efficiency.It is of our future work to refine our prediction model by supporting the automatic selection of the appropriate parameters for any jobs given the hardware and software settings of a BDPS. For practical applications, we will take the priority of jobs and fairness into consideration. We will also investigate the performance dynamics of BDPSs and design better scheduling approaches to account for such dynamics in our algorithm design.



---

*Source: 8598305-2022-04-12.xml*