How Heterogeneity Affects the Design of Hadoop MapReduce Schedulers: A State-of-the-Art Survey and Challenges

MR programming paradigm and its Hadoop implementation

The MR programming paradigm, ³ originally developed at Google, is an algorithmic design technique like divide and conquer. It is well suited for a wide variety of structured, semistructured, and unstructured data processing applications. The Map and Reduce functions in classical functional programming (e.g., Lisp) inspire the basic design idea of Google's MR. An MR computation is defined as a collection of user-defined Map and Reduce functions, which expect the input data to be a set of key–value pair \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $$\left\langle { \; \left. {k , v} \right\rangle } \right.$$ \end{document} . First, the input data are partitioned into multiple chunks and each is assigned to a user-defined map function that generates another set of intermediate key–value pair \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $$\left\langle { \left. {k \prime , {v^ \prime }} \right\rangle } \right.$$ \end{document} . Next, the MR library joins all the values associated with the same intermediate key (say \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $$k \prime \prime$$ \end{document} ) and forwards it to user-defined reduce function. Lastly, the reduce function merges the values to form a possibly smaller set of values. Typically, one output is produced per reduce function. Many real-world computational tasks can be easily expressed through this model. A typical MR computation can be written as follows, where \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $$\left\langle { \left. {{k_i} , {v_i}} \right\rangle } \right.$$ \end{document} and \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $$\; \left[ {{V_x}} \right] \;$$ \end{document} represent a key–value pair and a set of values, respectively. \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} \begin{align*} map: & \left\{ { \left\langle { \left. {{k_1} , {v_1}} \right\rangle } \right. , \; \left\langle { \left. {{k_2} , {v_2}} \right\rangle } \right. , \; \ldots \ldots \left\langle { \left. {{k_i} , {v_i}} \right\rangle } \right.} \right\} \to \\ & \quad \{ {\langle {k^\prime}_1} , {{v^\prime }_1} \rangle , \langle{{{k^\prime}_2}}, {{v^\prime}_2} \rangle, \ldots.. \langle{{k^\prime}_j}, {{v^\prime}_j}\rangle \} \end{align*} \end{document} \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} \begin{align*} reduce: & \left\{ { \left\langle { \left. {{{k^{\prime\prime}}_1} , \; \left[ {{V_1}} \right] } \right\rangle } \right. , \; \left\langle { \left. {{{k^{\prime\prime}}_2} , \; \left[ {{V_2}} \right] } \right\rangle } \right. \ldots \ldots \left\langle { \left. {{{k^{\prime\prime}}_x} , \; \left[ {{V_x}} \right] } \right\rangle } \right.} \right\} \to \\ & \{ {v^{\prime\prime}_1} , {{v^{\prime\prime}_2}} , \ldots \ldots.{{v^{\prime\prime}_y}} \}. \end{align*} \end{document}

In the literature, various implementations of MR programming paradigm are available. The right decision depends upon the application and environment. Google implemented this programming methodology for large clusters of commodity PCs (i.e., distributed memory multicomputer architecture) connected with switched ethernet. ³ In its master-slave architecture, data are stored in the GFS.^3,38 Other than Google MR, various frameworks that implement Google's MR methodology in their own way are Apache Hadoop, QtConcurrent, ³² Phoenix, ³³ Skynet, ³⁷ Twister, ³⁴ Disco, ³⁵ Misco, ³⁶ etc., specifically for either multiprocessor or multicomputer architecture. Among these, Hadoop is the most widely used implementation. A key benefit of Hadoop is its automatic failure handling feature, thus, hiding the complexity of fault tolerance from a programmer. In case a node crashes, Hadoop reruns its tasks on a different machine.

Scheduling in Hadoop MR

A small submodule within the JT/RM logic known as Scheduler performs the scheduling in Hadoop. It plays an important role in the overall performance of Hadoop and schedules three main entities, namely, users, jobs, and map/reduce tasks. The process of scheduling is to allocate computing resources (either in the form of slots or containers) to various scheduling entities. The MR scheduler is usually triggered by two events: (i) whenever a new job is submitted to job queue of the system and (ii) whenever JT receives a heartbeat message from a worker node. The heartbeat messages are periodically sent by worker nodes to inform the number of free slots it has.

Whenever a scheduler is triggered, it first selects a user for scheduling followed by job selection from a list of jobs submitted by that particular user. Afterward, the scheduler allocates resources to map or reduce tasks of the selected job. Depending upon the Hadoop version, a resource may be coarse grained or fine grained. On the basis of the scheduled entity, an MR scheduler may be classified as level 1 (user level), level 2 (job level), and level 3 (task level). Different scheduling algorithms can be used at each level based on the objectives to be achieved. It is noted that a single scheduling algorithm may or may not work at more than one level, for example, FAIR and CS work at the user and job level, whereas FIFO works only at job level in the scheduling hierarchy.

An MR scheduler caters various QoS requirements of two main stakeholders, Hadoop user and administrator, in a Hadoop system. Both stakeholders have their own quality requirements. For instance, a Hadoop administrator is concerned about throughput and resource utilization of a system, whereas an end user (customer, domain/business analyst, or auditor) expects good response time, service level agreement (SLA) adherence, minimum completion time, etc. for their jobs. Usually, an MR scheduling problem is mathematically formulated as optimization problem, for example, Linear Programming (LP) in which QoS requirements are either maximized or minimized. ¹⁷ Scheduling in Hadoop is a complex task and shown to be NP-hard in nature ²⁰ even in the restricted environment where all nodes are homogeneous and all resource requirements of tasks are similar.

From the mentioned discussion, we may precisely conclude that MR scheduler allocates computing resources in the form of slots/containers to various users, jobs, and map/reduce tasks to achieve various QoS objectives.

Heterogeneity factors in Hadoop

A number of users mostly share a single large Hadoop cluster, where each user has a different priority and resource demand to execute his or her jobs, thus leading to user heterogeneity in the system. Furthermore, each user has different types of jobs (e.g., CPU- or I/O intensive) to be executed on the large cluster that causes workload heterogeneity. Furthermore, given the large size of clusters dedicated to data-intensive applications, it is not always possible or even desirable to have a cluster with a single type of machine. Over time, machines with different configurations are inducted into the existing Hadoop cluster, ultimately causing hardware heterogeneity. Through the literature survey, a heterogeneous Hadoop system can be defined on the basis of the following factors (Fig. 1).

OPEN IN VIEWER

FIG. 1.

Heterogeneous Hadoop factors.

A Hadoop system may consist of partial or full hardware heterogeneity. It is possible that nodes with a similar type of CPUs may have memory units with varying bandwidths. The cluster-level heterogeneity (partial or full) may also occur in virtual cluster setup that can be built upon private data center or over clouds. Although virtualization technology can isolate CPU and memory performance effectively between Virtual Machines (VMs), heterogeneity may arise because I/O devices (disk and network) are shared between VMs. ²² For instance, only disk level heterogeneity has been considered in Zaharia et al., ²² while neglecting CPU and memory heterogeneity.

Scheduling challenges in heterogeneous Hadoop environment

In the beginning, the Hadoop was homogeneous in all three heterogeneity aspects. However, as Hadoop became popular, its scheduler's implicit assumption of homogeneous system has become obsolete in modern data centers, especially in the virtualized environment. Considering such heterogeneity factors by Hadoop schedulers while making scheduling decision is a tedious task. In this subsection, we discuss various scheduling challenges faced by MR scheduler in a heterogeneous environment. Following challenges have been identified through the literature survey.

Scheduling algorithms for user heterogeneity

Earlier Apache Hadoop was mainly designed for a single user to run large batch jobs such as generating web indexes and mining log data. A single FIFO queue was maintained for the same. As Hadoop is to be shared by multiple users now, single job queue (i.e., FIFO queue) becomes a bottleneck in performance and responsiveness that further affects the fairness of the system. In a shared cluster environment, a user with a small job may get starved for a long time if the job is submitted after a long batch job. In large systems, simple algorithms like FIFO can cause severe performance degradation, particularly when multiple users sharing the cluster having heterogeneous needs. ⁵⁵ Two popular MR schedulers, namely, FAIR and CS, work on a similar approach to cater the need of heterogeneous users to maintain fairness. The standard Hadoop distribution includes these two schedulers in all versions. Fairness is the main performance objective in heterogeneous multiuser environment, which otherwise cannot be achieved by FIFO. Although fairness implies that each user and job get a proportional number of slots to run its tasks, the same does not hold true for heterogeneous cluster/hardware environment where different slots may have different computational power [discussed in Scheduling algorithm for cluster (hardware) heterogeneity subsection].

FAIR ⁴⁸ and CS ⁵¹ are two early pluggable schedulers as soon as the scheduler in Hadoop became a pluggable component. ⁵⁶ Separating the scheduler logic from JT module led to innovation in this domain. Developed as a special multiuser scheduler for in-house usage by Facebook and Yahoo, the two schedulers were later included in Hadoop distribution and undergone several improvements ranging from security to better resource utilization. FAIR and CS both are designed separately for slot-based and YARN-based Hadoop. The description here is in the context of slot-based Hadoop framework. The difference between FAIR and CS is minimalist. The FAIR as well as the CS equally distributes the compute resources among users/jobs to achieve fairness. FAIR allocates one pool to each user and the resources (slots) are divided equally among all the pools. It uses a generalization of max-min fairness ⁵⁷ with a minimum guarantee to allocate slots across pools. However, CS uses queues instead of pools, in which each queue is assigned to an organization, and resources (slots) are divided among these queues. Additional security mechanisms are built to control access to the queues so that each organization can access only its queue and cannot interfere with another organization's queues, jobs, or tasks.

Both FAIR and CS guarantee a minimum share of resources for a pool or queue. If a pool does not receive a minimum share for a long time, FAIR pre-empts the most recently started task of an over-allocated job from some other pool. This ensures that long batch jobs do not block the execution of some production jobs and also that a smaller number of resources are wasted (those spent in the pre-empted task). Similarly, CS limits running/pending tasks and jobs from a single queue. To maximize resource utilization, it allows reallocation of resources of a free queue to queues with full capacity. Whenever jobs arrive in that queue, running tasks are completed and resources are given back to the original queue. It also allows priority-based scheduling in an organization queue. FAIR schedules jobs within a pool based on priority, first come first serve, or fair-share basis. Both schedulers aim to assign a fair share to users, which means resources are assigned to jobs such that all users receive, on average, an equal share of resources over time. Therefore, the fair sharing and capacity scheduling algorithm only takes user heterogeneity into account.

FAIR ⁴⁸ scheduler improves the response time and throughput in comparison with FIFO scheduler. However, it suffers from the data locality issue because in the FAIR scheme, tasks are assigned to slots as soon as they become available to maintain the fairness among users compromising with data locality as identified by Zaharia et al.^48,55 To address the conflict between locality and fairness, Zaharia et al. ⁵⁵ extended their FAIR ⁴⁸ scheduler, used a simple algorithm called Delay scheduling, ⁵⁵ and proposed a scheduler essentially known as Hadoop Fair Scheduler (HFS). ⁵⁵ According to Delay scheduling scheme, if data locality is not achieved, job, which is scheduled in next turn, cannot launch its map or reduce task to a free slot. Rather, it waits for some time and allows other jobs to launch their task. After some time, if data locality constraint is met, the job is scheduled immediately. Thus, Delay scheduling achieves nearly optimal data locality by deferring the immediate allocation of slots to tasks while preserving fairness in a variety of workloads and can double the throughput. Delay scheduling can also be combined with FIFO, BalancedPools, ⁵⁸ HScheduler, ⁵⁹ or any other queuing policy that produces a sorted list of jobs. HFS was developed as an extension to FAIR and later included in a standard Hadoop 0.21 framework.

In fine-grained YARN cluster manager (i.e., in Hadoop 2.x and Hadoop 3.x), achieving fairness for multiple heterogeneous users is a complex task and well received by DRF ⁴¹ scheduler. Initially, DRF was proposed for Mesos cluster manager, ⁴² however, later adopted in Hadoop framework after inclusion of general purpose YARN cluster manager. DRF further generalizes the concept of max-min fairness to multiple resource types that leads to better throughput. It is noted that max-min fairness generalization is also used by FAIR ⁴⁸ for only single resource type. The intuition behind DRF is that in a multiresource environment, the allocation of a user should be determined by the user's dominant share, which is the maximum share that the user has been allocated for any resource. The maximum among all shares of a user is called that user's dominant share and the resource corresponding to the dominant share is called the dominant resource. Different users may have different dominant resources. For example, if user A runs CPU-heavy tasks and user B runs memory-heavy tasks, DRF attempts to equalize user A's share of CPUs with user B's share of memory. DRF satisfies a number of desirable properties identified for any fairness scheme in the multiresource fine-grained environment. In particular, DRF fulfills all four essential properties that include sharing incentive, strategy proofness, Envy freeness, and Pareto efficiency. ⁴¹ Each scheduling decision in DRF can be made in \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${O} \left( {log \ n} \right)$$ \end{document} for n users.

Although FAIR, CS, and Delay schedulers were initially built for slot-based Hadoop framework, all were modified later and included in YARN-based Hadoop framework where multiple resource types are available for allocation as compared with single slot-based abstraction. Table 1 summarizes the already discussed scheduling algorithms on the basis of QoS objectives, SLA adherence, scheduling entity, etc. where QoS objective(s) refer to most important QoS metric(s) aimed to improve by particular scheduling algorithm. In case, the scheduling problem is formulated mathematically as an optimization problem, it refers to the parameter being minimized or maximized. The static/dynamic characteristic describes whether the particular scheduler is static (in which all jobs arrive at time zero) or dynamic (where jobs arrive over time). Mathematical formulation column describes whether the scheduling problem has been formulated as an integer linear program or not. Table 2 summarizes the various quality metrics and base algorithm(s) that have been used in comparison while experimentally evaluating the algorithms. The mentioned discussion of table attributes applies to Tables 3–8.

Scheduling algorithm for cluster (hardware) heterogeneity

A different configuration of computer hardware is a common phenomenon in data centers as a new machine becomes available every year. The induction of new machines in data centers results in heterogeneous Hadoop cluster at hardware level. The Hadoop implicitly assumes that all nodes in the cluster are homogenous and make scheduling decision accordingly. In the literature, various scheduling algorithms have been proposed to overcome the problem of hardware heterogeneity. In this subsection, such types of scheduling algorithms are discussed. We note that straggler detection in heterogeneous hardware environment is a difficult task. Most of the algorithms in this subsection address the straggler detection problem especially in a heterogeneous cluster environment to improve the overall response of the system.

To improve the response time of jobs, Hadoop incorporates the mechanism of speculative execution. It identifies the slow tasks called stragglers and speculatively runs its copy (also called “backup task”) on different nodes to finish the computation faster. The tasks may be slow due to different reasons such as heavy system load, faulty hardware, and process deadlock. To identify a slow task, Hadoop monitors tasks progress using a progress score between 0 and 1 using Equation (1), ²² where M is the number of key–value pairs that have been processed and N is the number of key–value pairs that need to be processed for map task. In the same manner, M′ is the number of key–value pairs that have been processed and N′ is the number of key–value pairs that need to be processed in any particular phase of reduce task. A task is marked as a straggler, in case it has run for at least 1 minute and its ProgressScore (PS) is less than the average for its category minus 0.2 (standard Hadoop distribution uses value 0.2). While computing the PS for reduce task, a multiplicative term “1/3” is used because a reduce task has three phases, namely, shuffle (copy), sort, and reduce. Hadoop implicitly assumes that cluster is homogeneous and each node executes the task at the same speed, that is, tasks make progress linearly. It further uses these assumptions to identify the stragglers and decides when to speculatively re-execute stragglers. Hence, in a heterogeneous cluster environment, Hadoop scheduler(s) need to be redesigned considering hardware heterogeneity. A heterogeneous cluster consists of diverse CP machines. Hence, the reason for a slow task may also be a low CP node if the task is scheduled on it. Given this, straggler detection is a challenging task due to ambiguity in reason for its creation. This problem also arises in clouds because cluster heterogeneity frequently occurs in virtualized cloud computing infrastructure such as Amazon EC2 and Google. ²² \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} \begin{align*} { \rm{PS}} = \; \left\{ {\begin{matrix}{M / N \qquad\qquad\qquad\qquad{ \rm{for \;}}map \;task} \\ 1/3 * \left( {K + M^\prime / N^\prime } \right) \quad { \rm{for \;}}reduce \;task\end{matrix}} \right.. \tag{1} \end{align*} \end{document}

Longest Approximate Time to End (LATE), proposed by Zaharia et al., ²² is one of the early MR scheduling algorithms proposed for the heterogeneous cluster to identify and schedule straggler tasks. The LATE scheduler decides re-execution of stragglers speculatively to improve response time in a cloud environment. Native Hadoop scheduler considers all stragglers equally low in terms of speed. However, LATE prioritizes the speculative tasks. It selects the fast node for straggler execution and binds the number of speculative tasks using a threshold to prevent thrashing. The algorithm first calculates the PS of a task using Equation (1) (like native Hadoop scheduler). Thereafter, it calculates the ProgressRate of each task as ProgressScore/T, where T is the amount of time the task has been running for. Then, a new heuristic, “time to completion” (commonly referred as TimeToEnd) of the task, is estimated as (1 − ProgressScore)/ProgressRate. The heuristic serves to prioritize the stragglers, that is, tasks with high completion time are speculatively re-executed first. A different technique to estimate the completion time may also be incorporated into LATE. It is noted that authors have created heterogeneity through virtualized Hadoop cluster over the cloud and private data center. Furthermore, LATE also improves the performance of speculative execution in a homogeneous environment.

By focusing on estimated time left rather than ProgressRate as explained earlier, LATE speculatively executes only those tasks that improve job response time, rather than any slow task. However, LATE scheduler adopts a static method to compute the TimeToEnd heuristic, which is further used to identify slow tasks. As a result, LATE does not perform so well in a heterogeneous environment. The Hadoop default, LATE, and BASE ⁶⁰ speculative scheduling algorithms assume equal execution time at all stages of reduce tasks and consider 33% completion of a reduce task while calculating its ProgressScore. The algorithms further assume only one stage in map task. Such assumptions may not always hold true because the time taken by each map and reduce stage depends on the type of node.

To further improve the performance, Self-Adaptive MapReduce (SAMR) scheduling algorithm, proposed by Chen et al., ⁶¹ estimates the time left for tasks to complete on a given node using dynamic stage weights. The algorithm adjusts the weight of map and reduce tasks using previous (historical) information. The TT reads the historical stage weight information stored on every node in xml format and utilizes it to tune the stage weights of currently running task and finally updates the historical database with current value. SAMR speculatively schedules the slow tasks with highest estimated time left.

The SAMR algorithm also identifies slow and fast TTs (i.e., nodes) dynamically using the mathematical expression established in Chen et al. ⁶¹ and classifies them into slow and fast TTs using the parameter SLOW_TRACKER_CAP. The scheduler never launches backup tasks on slow nodes, ensuring that the backup tasks will no longer be slow. The experimental results show that SAMR significantly decreases the execution time up to 25% and 14% when compared with Hadoop default scheduler and LATE scheduler, respectively. ⁶¹

Chen et al. ⁶² proposed another variant of SAMR, called HAT ⁶² (History-based Auto-Tuning), which uses historical data on each node to adjust the weights of the map and reduce phases. Using the updated value of weights, the progress score of tasks is calculated (refer Chen et al. ⁶² for a detailed description). Based on the accurately calculated progress of tasks, HAT estimates the remaining time of tasks and further launches backup tasks for those with longest remaining time. The limitation of SAMR and HAT scheduling algorithms is consideration of heterogeneous hardware only while calculating the stage weight of tasks.

Yang and Chen ⁶³ make an improvement on the original speculative execution method of Hadoop (called as Hadoop speculative in the literature) and LATE Scheduler by proposing a new scheduling scheme known as Adaptive Task Allocation Scheduler (ATAS). The ATAS adopts more accurate methods (due to refined Approximate End Time [AET] calculation) to determine the response time and backup tasks that effectively increase the system's ability to respond. It finds out the straggler through more reasonably calculated AET heuristic of each task. The speculative execution is based on the principle of launching backup tasks for map/reduce tasks that have longest expected execution time in the computing process. AET of the ith task is computed as shown in Equation (2), where AvgProgress represents the average speed of task execution and k is the serial number of the reduce stage. AvgProgress of a task is computed as Progress/T where T represents the task execution time. \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} \begin{align*} { \rm { AET } } = \; { \frac { 1 - \sum \nolimits_ { i = 1 } ^k Progress \left[ i \right] / T } { AvgProgress } } . \tag { 2 } \end{align*} \end{document}

The scheduler divides the cluster nodes into two sets, namely, QuickNode and SlowNode, based on speed. The QuickNode is prioritized with the allocation of backup tasks. The authors conducted three different simulation experiments where the execution time of LATE Scheduler and ATAS is reduced by a maximum of 12% and 30%, respectively. Furthermore, the average tasks throughput is increased by a maximum of 13% and 33%, and the average tasks latency is reduced by a maximum of 25% and 36%, respectively, in comparison with Hadoop speculative. Thus, ATAS can effectively enhance the overall processing performance of MR in a heterogeneous cloud environment.

Mashayekhy et al. ⁶⁴ proposed a framework to improve the energy efficiency of MR applications over heterogeneous machines while satisfying the SLA. The problem of energy-aware scheduling of a single MR job has been modeled as an Integer Program (IP) called Energy-aware MapReduce Scheduling (EMRS-IP) with a deadline as a constraint (refer Mashayekhy et al. ⁶⁴ for detailed expressions for constraints). Equation 3 shows the objective function of the formulated IP where M and R are the set of map and reduce tasks of the application, A and B are the set of map and reduce slots, respectively, on heterogeneous machines, \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${e_{ij \;}}$$ \end{document} is the difference between energy consumption of slot j when executing task i and its idle energy consumption. \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} \begin{align*} \begin{split} & { \rm{min \;}}{ \sum \nolimits_{j \in A}}{ \sum \nolimits_{i \in M}}{e_{ij}}{X_{ij}} + \\ & \quad \;{ \sum \nolimits_{j \in B}}{ \sum \nolimits_{i \in R}}{ \sum \nolimits_{t \in M \cup R}}{ \delta _{ij}}{e_{ij}}{Y_{ij}}.\end{split} \tag{3} \end{align*} \end{document}

Two heuristic algorithms have been proposed to solve EMRS-IP, called EMRS algorithms: EMRSA-I and EMRSA-II. The algorithms find the assignment of the map and reduce tasks to the machine slots to minimize the overall energy consumption while executing the application. Both heuristics take the energy efficiency differences of machines into account and use a metric called energy consumption rate (ecr), which is calculated for each slot. The metric characterizes the energy consumption of each machine and induces an order relation among the machines. EMRSA-I calculates energy consumption rate metrics \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${ \rm{ecr}}_j^m \;$$ \end{document} and \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${ \rm{ecr}}_j^r$$ \end{document} for each map and reduce slot as the minimum ratio of energy consumption and processing time of tasks when executed on slot j. In particular, ^‡ \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${ \rm{ecr}}_j^m$$ \end{document} is calculated using the expression \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${ \rm{ecr}}_j^m = { \mathop { \min } \nolimits_{ \forall i \in M}}{e_{ij}} / {p_{ij}} , \; \forall j \in A$$ \end{document} , where symbols have usual meaning already explained. In contrast, EMRSA-II calculates energy consumption rate metrics \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${ \rm{ecr}}_j^m \;$$ \end{document} and \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${ \rm{ecr}}_j^r$$ \end{document} on the basis of average ratio of energy consumption and processing time of tasks when executed on slot j. In case of EMRSA-II, particularly ^‡ \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${ \rm{ecr}}_j^m$$ \end{document} is calculated as \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $$[ { \sum \nolimits_{ \forall i \in M}}{e_{ij}} / {p_{ij}} ] / M , \; \forall j \in A$$ \end{document} . Therefore, both algorithms differ in terms of calculation of energy consumption rate metric. Being an IP, EMRS-IP is an NP-hard problem and the proposed approaches find a near-optimum solution. The results show that EMRSA-I and EMRSA-II can find near-optimal job schedules consuming ∼40% less energy on average than the schedules obtained by a common practice scheduler that minimizes the makespan. While formulating the problem, energy consumption and execution time of map and reduce tasks are kept different for each slot due to consideration of heterogeneous machines. It is noted that heterogeneous workload has been used while evaluating the EMRSA algorithms. However, the same has not been considered in scheduling decision of the algorithms.

The MapReduce Task Scheduler for Deadline (MTSD) algorithm presented by Zhou and colleagues ⁶⁵ is first of its kind that prioritizes and schedules jobs to meet their response time SLAs in a heterogeneous environment. The algorithm takes hardware heterogeneity into account while making scheduling decisions. MTSD introduces a classification algorithm that classifies the nodes in a multilevel hierarchy according to their computing capacity. The fastest nodes are placed on the highest level, whereas slowest node is assigned to the lowest level. Furthermore, the data are distributed on different levels proportionally, that is, the nodes at the highest level will get a maximum proportion of data that may improve data locality and completion time. The algorithm further uses the remaining time of partially completed tasks and estimated task execution time on the slowest nodes to compute the required number of map and reduce slots. Tables 3 and 4 summarize various attributes of studied schedulers in this subsection.

Scheduling algorithm for workload heterogeneity

A workload is a collection of various MR jobs that could be either CPU or I/O intensive, or both. As discussed in Heterogeneity factors in Hadoop subsection, jobs in a workload may have different size and priority. The heterogeneity in workload may complicate the scheduling decision in Hadoop. One simple technique to achieve better performance is to distribute the different types of jobs uniformly on every node in the cluster. For example, if the map tasks of two CPU-intensive jobs are scheduled on one node, both of them will compete for CPU and take a longer time to finish. In contrast, if the tasks of a CPU-intensive job and an I/O-intensive job are scheduled on the same node, both can run in parallel without having an impact on each other's performance.

Polo et al. proposed RAS ⁶⁶ (Resource-aware Adaptive Scheduling for MapReduce Clusters) scheduler in which a slot is fixed for a specific type of job. The algorithm extends the abstraction concept of “task slot” to “job slot” and uses profiling information from previous execution of jobs. The job slot is an execution slot bound to a particular job as well as task type (either map or reduce) within that job. The algorithm considers completion time goals as soft deadlines being submitted by users. RAS has four main components: (i) Job Status Updater, (ii) Job Time Completion Estimator, (iii) Placement Algorithm, (iv) Job Utility Calculator. The Job Status Updater keeps track of average task length of job and later used in estimation of completion time for each job. The second component, job time completion estimator, estimates the number of map tasks allocated simultaneously to meet the deadline for a job. The placement algorithm forms possible placement matrices after examining the tasks on different TTs and their resource allocation. In contrast, the job utility calculator gives a utility value for the placement matrix P. The matrix value P_j,tt represents a number of tasks of job j placed (scheduled) on TT tt. The placement algorithm selects best placement choice and thereafter task scheduler enforces the placement decision. The entire focus of RAS is on Memory, I/O, and CPU; however, it can be easily extended to include parameters such as storage capacity and network bandwidth.

Tian et al. ⁶⁷ proposed a scheduler called Triple-Queue that overlaps CPU- and I/O-bound tasks to improve the overall utilization of the cluster. As the name implies, Triple-Queue scheduler maintains three separate queues, namely, wait queue, CPU-bound queue, and I/O-bound queue for incoming jobs. It employs a workload predict mechanism called MR-Predict, which automatically predicts the class of a new coming job and inserts the jobs into the corresponding queue. Whenever a new job arrives, it is first kept in the wait queue. The Triple-Queue scheduler takes a different approach by first allocating the map tasks of a new job to an available slot. The scheduler then moves the job to a CPU-bound or a disk I/O-bound queue using the execution data of job (precisely if a job spends most of its time doing computation on CPU, it is moved to CPU-bound queue otherwise to I/O-bound queue). The scheduler continuously monitors the execution times of tasks to move jobs dynamically from one type of queue to another. Such a scheme has been shown to achieve a 30% improvement in the overall throughput over the default single-queue scheduler. However, the scheme explored a limited form of workload heterogeneity.

The Job-Aware Task Scheduling (JATS) algorithm, proposed by Nanduri et al., ⁶⁸ uses a task vector \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${ \vec T_k}$$ \end{document} to represent the proportion of each resource type consumed by the task (map or reduce) while executing on a node. Task vector is represented as \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${ \vec T_k} = \;{E_{{ \rm{cpu}}}}{e_1} + {E_{{ \rm{mem}}}}{e_2} + {E_{{ \rm{disk}}}}{e_3} + {E_{{ \rm{nw}}}}{e_4}$$ \end{document} , where \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${E_{{ \rm{cpu}}}} , \;{E_{{ \rm{mem}}}} , \;{E_{{ \rm{disk}}}}{ \rm{ \;and}} \;{E_{{ \rm{nw}}}}$$ \end{document} are resource usage patterns for CPU, memory, disk, and network bandwidth, respectively, of a particular job. How these resource usage patterns are calculated is not explained by Nanduri et al. ⁶⁸ Furthermore, \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${e_1} , \;{e_2} , \;{e_{3 \;}}$$ \end{document} , and e₄ are basis vectors. In this scheme, the scheduler is made aware of different types of jobs running on the cluster. It tries to allocate a task on a node if the incoming task does not affect the tasks already running on that node, thus avoiding node overloading. To find out whether incoming job will affect or not, both heuristic and machine learning-based approaches have been used. In machine learning approach, supervised incremental Naïve Bayes classifier^69,70 has been employed. It tries to maintain a resource balance on the cluster by not overloading any of the nodes, thereby reducing the overall runtime of the jobs.

Nanduri et al. ⁶⁸ considered the following features to train the classifier: (i) hardware specifications of TT (Φ), (ii) network distance between the nodes of task execution and the corresponding data split of the task ( \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${ \rm{ \Sigma }}$$ \end{document} ), (iii) TaskVector of incoming task ( \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${ \vec T_k}$$ \end{document} ), and (iv) TaskVectors of tasks running on TT [ \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $${ \vec T_{{ \rm{compound}} \left( i \right) }}$$ \end{document} ], collectively forming the feature set \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $$F = \left\{ {{ \rm{ \Phi }} , { \rm{ \; \Sigma }} , \; \;{{ \vec T}_k} , \; \;{{ \vec T}_{{ \rm{compound}} \left( i \right) }}} \right\} .$$ \end{document} The heuristic-based approach runs a Task Compatibility Test to check whether a task is compatible with the node or not so that it may not overload the node once scheduled. It is noted that JATS proposes a scheduler for slot-based Hadoop; however, task vector resembles the resource request like YARN.

Another algorithm that considers workload heterogeneity into account is LsPS, ⁷¹ which is actually an improvement of FAIR scheduler. FAIR maintains a separate job queue for each user and divides the resources fairly among all users so that each user gets an equal amount of slots that leads to better response time. However, one major drawback of the FAIR scheme is that it does not consider job size of the user while distributing slots. Multiple users have heterogeneous resource demand and to maintain better fairness to improve overall response time, slots must be allocated to users according to job size. Therefore, LsPS adaptively adjusts the slot shares among all active users such that the share ratio is inversely proportional to the ratio of their average job sizes. For example, in a simple case of two users, if their average job size ratio is equal to 1:2, the number of slots assigned to the first user will be twice that to the second user. Consequently, LsPS implicitly assumes higher priority to users with smaller jobs, resulting in better response time. At job level, LsPS dynamically tunes the scheduling scheme for jobs within an individual user by leveraging the knowledge of job size distribution. Whenever job size has high variance, the LsPS algorithm uses FAIR scheduler at the job level. As soon as the job sizes become close to each other, FIFO scheduler is used.

To the best of our knowledge, PRISM ⁷² (Fine-Grained Resource-Aware Scheduling for MapReduce) scheduler is first of its kind that considers phase-level scheduling within a task. The phase-level scheduling refers to the consideration of different characteristics of various phases (mapping, shuffling, sorting, merging, and reduce) of a map and reduce task in scheduling decision. Other heterogeneous workload schedulers such as RAS, ⁶⁶ HaSTE, ⁵⁴ and JATS ⁶⁸ perform scheduling at the task level only. Various phases of map and reduce tasks are the finest level of granularity possible for scheduling. Zhang et al. ⁷² find out that even a single task of a job usually has highly varying resource requirements during their lifetime, for example, shuffle phase of reduce task is highly I/O intensive, whereas reduce phase is CPU intensive. Motivated by this observation, they proposed a fine-grained resource-aware scheduler that coordinates task execution at the level of phases. Job profiling plays a significant role in these techniques as in ARIA ⁷³ and BalancedPools. ⁵⁸ Whenever a particular phase of a task finishes execution, it asks permission from the NM to start the next phase. Afterward, NM forwards the request to the scheduler that uses the current progress information and the phase level resource requirement to take a decision. On the basis of the information, the scheduler either allows the task to continue its next phase or launches a new task.

The drawback of the PRISM scheduler is that it pauses a task at runtime and thus delays the completion of the current task as well as all the subsequent tasks. Besides this, the authors claim that the performance of PRISM is dependent on the accuracy of the profiler. Any standard profiler such as Starfish ⁷⁴ can be plugged into PRISM. A less efficient profiler that extracts wrong insights from profiled data could degrade the performance. It is worth mentioning that it is not clear what kind of heterogeneous jobs were used during the evaluation of PRISM. Tables 5 and 6 summarize the various attributes of scheduling algorithms discussed in this subsection.

Scheduling algorithm for multiple heterogeneity factors

Consideration of multiple heterogeneity factors in Hadoop system while making scheduling decision is gradually becoming a common phenomenon. In the literature, few algorithms such as DySacle, ⁵⁰ Medley, ⁵² and COSHH ⁴⁷ consider more than one heterogeneity factor simultaneously. One type of heterogeneity is usually exploited with the help of another heterogeneity factor to increase the performance of scheduling process. For instance, to exploit the hardware heterogeneity of cluster nodes, a mixed workload of jobs with diverse resource demand is used. In this scenario, jobs with heterogeneous resource demand can be efficiently allocated the suitable resources. We illustrate a potential advantage of considering multiple heterogeneities by an individual scheduling technique as follows. Assume a heterogeneous Hadoop cluster consisting of two types of machines. The first type of machines (say type 1) have a high-performance CPU with a moderate speed optical disk drive, whereas the second type of machines (say type 2) have a low-performance CPU and a fast SSD. If a workload comprising two jobs is executed on the cluster where one of the jobs is CPU intensive and other is disk intensive, scheduling a CPU-intensive job to type 1 machines and disk-intensive job to type 2 machines will lead to better performance despite data transfer overhead. The default Hadoop scheduler does not consider hybrid heterogeneity and assigns jobs favoring only data locality.

In Scheduling algorithm for cluster (hardware) heterogeneity subsection, we discussed the drawback of SAMR ⁶¹ scheduling algorithm that it only considers the hardware heterogeneity. However, different types of jobs may have a different map and reduce stage weights. Even, for the same type of jobs, different data sets may lead to different weights. SAMR algorithm does not consider stage weight variations due to workload heterogeneity. To overcome the drawbacks of SAMR, Sun et al. ⁷⁵ proposed an ESAMR algorithm to improve the speculative re-execution of slow tasks in MR. To identify slow tasks more accurately, ESAMR considers the type of job and the input data size along with the node type while calculating weights for map and reduce task stages. The algorithm divides the historical stage weights information of tasks on each node into k clusters using a machine learning technique, that is, k-means clustering algorithm. The number k depends on characteristics of the data set (i.e., stage weights information). When executing a job on a node, ESAMR classifies the job's tasks into one of the clusters and uses the cluster's weights to estimate the time left to completion (i.e., TimeToEnd) of job's tasks on the node. Experiments show that taking the heterogeneity of workload along with cluster into account while estimating the TimeToEnd heuristic of a task produces least error as compared with SAMR and LATE. One drawback associated with ESAMR is that the k-means clustering algorithm introduces time overhead. Moreover, the algorithm does not consider the work done by the straggler task and starts the execution of the new task from scratch.

Yan et al. ⁵⁰ also considered hardware (CPU type) and workload heterogeneity while proposing DyScale scheduling algorithm for a Hadoop cluster in which each node has a heterogeneous multicore processor(s). DyScale aims at exploiting these heterogeneous cores within a single chip for achieving a variety of performance objectives. Authors have assumed that each chip has few fast cores and more slow cores. They further considered two types of jobs that are as follows.

The two different pools of virtual resources are constructed from heterogeneous physical cores. One is made up of fast cores of every node in the cluster and second is made up of slow cores. Each pool has its own job queue, namely, InteractiveJobQueue for short interactive jobs and BatchJobQueue for long batch jobs. Jobs in InteractiveJobQueue are scheduled over a fast pool of resources so that they can be completed in short time, whereas jobs submitted to BatchJobQueue are assigned resources from a slow pool of resources. The job ordering/scheduling within both job queues can be performed through any one of the existing techniques, for example, FIFO and HFS. ⁵⁵ Furthermore, resource allocation within the queue may be handled by ARIA SLO-driven scheduler. ⁷³ It is worth mentioning that heterogeneous core chips are not yet available, although expected to be used in the near future. ⁵⁰

Fairness metric needs to be redefined in the situation where the cluster is made up of heterogeneous machines. The number of slots might not be an appropriate metric of the share in a heterogeneous cluster because all the slots are not the same. Even some slots are more suitable for a particular job due to its microarchitecture. Furthermore, slots of a node with added hardware accelerator like Graphical Processing Unit (GPU) are more suitable for the specific type of job, for example, Cypto. ⁵² Lee et al. ⁵² redefined the fairness metric of delay scheduler ⁵⁵ and proposed a new scheduler called Medley for a multiheterogeneous environment. The Medley scheduler exploits the fact that a particular job runs faster on a specific node. Before achieving the fairness among the jobs, the scheduler establishes the relationship between job and slot through a fitness metric called Computing Power (CP). It is defined as the relative processing power of a slot for a particular task. It is worth mentioning that the CP of any slot is job dependent. Furthermore, the scheduler proposes a new fairness metric called Progress Rate (PR) as a ratio of the aggregate CP of slots. Mathematically, PR of a job j is defined in Equation (5). \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} \begin{align*} { \rm { PR } } \left( j \right) = \; { \frac { \sum \nolimits_ { s^\prime \in { S_j } } { \rm { CP } } \left( { s^\prime , j } \right) } { \sum \nolimits_ { s^{\prime\prime} \in S } { \rm { CP } } \left( { S^{\prime\prime}, j } \right) } } , \tag { 5 } \end{align*} \end{document}

where S_j is a set of all slots currently allocated to job j and S is a set of all slots available in the cluster. Medley scheduler allocates the appropriate number of specific slots to the tasks of a particular job, so that PR of each job reaches 1. It is worth mentioning that combined PR of all jobs may be >1.

The same fairness and fitness (job affinity) metric used in Lee et al. ⁵² are further utilized by Lee et al. ⁵³ with just different names, namely, PS and Computing Rate (CR). The authors have proposed two schemes, one for VMs allocation (i.e., for virtual cluster setup) and second for MR job scheduling in the cloud. Given the fluctuating resource demands of MR workloads, the data analytics cluster must scale according to demands. To that end, authors suggested a resource allocation strategy similar to the previous work of Chohan et al. ¹⁰ The scheme divides available machines into two virtual pools, namely, core nodes and accelerator nodes virtual pool. Core nodes are general purpose VMs with a disk for storage, whereas accelerator nodes are diskless machines with dedicated GPUs. Size of each pool is dynamically adjusted according to needs of incoming jobs to reduce cost and improve utilization. Job affinity (job/instance type relationship) has been taken into consideration while allocating resources for the cluster. It is quantized using the relative speed of each instance type i for a particular job j, which is called computing rate and denoted as CR(i, j). If a task of job j runs twice faster on the instance i₁ than on i₂ (i.e., finishes in half of the time), then CR(i₁, j) is twice of CR(i₂, j). CR is determined during the calibration phase of a job execution in which at least one map task is scheduled on every slot. By using the computing rate information and the cost of each instance type, the cloud driver can make a decision on which instance type to use. It guides to have a good mix of different instances that minimize the cost to maintain the cluster. Once the cluster has been set up, jobs are scheduled achieving fairness using ProgressShare metric of Medley scheduler in a heterogeneous environment.

Like DyScale, ⁵⁰ Medley ⁵² and ProgressShare ⁵³ scheduling algorithms, ThroughputScheduler proposed by Gupta and colleagues ⁷⁶ also tries to schedule the best task for a slot through job/task affinity or fitness property. The algorithm reduces the overall job completion time on a cluster of heterogeneous nodes. It schedules the task on the node whose capability (e.g., relative CPU and disk I/O speeds) optimally matches the task requirement. The algorithm works in two phases (i) explore phase and (ii) exploit phase. In the explore phase, the algorithm learns node characteristics in offline manner by running probe jobs and then job resource profile is built using a Bayesian active learning scheme in online mode without interrupting production use of the cluster. The characteristics/capabilities of the server are described by a corresponding vector \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $$k = \left[ {{k_1} , \;{k_2} , \; \ldots \ldots. {k_N}} \right]$$ \end{document} , which represents rates for processing the respective operation type (e.g., FLOPS or I/O per second). Furthermore, the task resource requirements are represented by a vector \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $$\theta = \left[ {{ \theta _1} , \;{ \theta _2} , \; \ldots \ldots..{ \theta _N}} \right]$$ \end{document} , where each component represents the total requirement for an operation type (number of instructions to process, bytes of I/O to read, etc.). For simplicity, the authors have assumed only two types of resources, namely, CPU and disk I/O bandwidth, which are represented by a two dimensional vector \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $$\;k = \left[ {{k_c} , \;{k_d}} \right]$$ \end{document} . In the same manner, task requirement is represented by \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $$\; \theta = \left[ {{ \theta _c} , \;{ \theta _d}} \right]$$ \end{document} . Once the resource profile of a job is learned, algorithm switches from explore to exploit phase and applies a real-time heuristic to optimally schedule tasks onto available cluster nodes. For each job J from the list of jobs ready for execution, a score value (i.e., heuristic) is calculated for a particular node N as shown in Equation (6). In the equation, \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $$\left[ {k_c^N , k_d^N} \right]$$ \end{document} and \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} $$\; [ \theta _c^{ \;J} , \; \theta _c^{ \;J} ] \;$$ \end{document} represent the resource capability of node N and corresponding resource requirements of job J, respectively. A job with minimum score value is picked for execution on node N. \documentclass{aastex}\usepackage{amsbsy}\usepackage{amsfonts}\usepackage{amssymb}\usepackage{bm}\usepackage{mathrsfs}\usepackage{pifont}\usepackage{stmaryrd}\usepackage{textcomp}\usepackage{portland, xspace}\usepackage{amsmath, amsxtra}\usepackage{upgreek}\pagestyle{empty}\DeclareMathSizes{10}{9}{7}{6}\begin{document} \begin{align*} { \rm { score } } = \; { \frac { { \rm { norm } } \left( { \theta _c^ { \;J } } \right) } { { \rm { norm } } \left( { k_c^N } \right) } } + { \frac { { \rm { norm } } \left( { \theta _d^ { \;J } } \right) } { { \rm { norm } } \left( { k_d^N } \right) } } . \tag { 6 } \end{align*} \end{document}

The Context-Aware Scheduler for Hadoop (CASH) ⁷⁷ assigns tasks to the nodes that are most capable of satisfying the tasks' resource requirements like ThroughputScheduler. CASH also learns resource capabilities and resource requirements, however, in offline mode, to enable efficient scheduling. The scheduler categorizes each node into “CPU-resources” and “I/O-resources” buckets by comparing their CPU and I/O benchmark results with the results of a baseline machine. It also classifies map and reduce tasks of jobs into CPU-bound and disk I/O-bound classes by comparing their execution rates (from history logs) with the speed/bandwidth of available resources. It is observed that this scheduler relies heavily on historical and benchmark data for appropriate scheduling, which may not exactly match the current job and resource scenario.

Rasooli and Down ⁴⁷ proposed a scheduling algorithm called COSHH, which considers all three heterogeneity factors in making scheduling decisions. First of all, the algorithm classifies the jobs using k-means clustering algorithm on the recorded execution times. When a new job arrives, the classification approach specifies the job class to which the job belongs, and stores the job in the corresponding queue. If the job does not fit any of the current classes, the list of classes is updated to add a class for the incoming job. Next, the optimization approach is used to find an appropriate matching of job classes and resources. An optimization problem is defined based on the properties of the job classes and features of the resources. The algorithm determines the allocation matrix for jobs using an LP approach by maximizing resource utilization. Constraints for LP formulation are execution rates of different job classes on different resources, arrival rates of these job classes, and capacity of the available nodes. At the time of a scheduling decision, the COSHH algorithm uses the set of suggested job classes for each resource and considers the priority, required minimum share, and a fair share of users to make a scheduling decision. In this scheme, data are replicated on the resource where the task will be executed to increase data locality. We note that COSHH also uses the concept of job affinity to exploit the hardware heterogeneity.

Yigitbasi et al. ⁷⁸ proposed an energy-efficient algorithm called EESched for scheduling heterogeneous workload to the heterogeneous cluster consisting of high- and low-power machines. This algorithm provides an opportunity to save energy by intelligently placing jobs on its corresponding energy-efficient machine. Based on the initial experiments, the authors derived the fact that I/O bound workloads have better energy efficiency on low power nodes, whereas CPU bound workloads achieve better energy efficiency on high-power nodes. Another fact was derived that a map task is more CPU intensive, whereas a reduce task is more I/O intensive. The energy-efficient scheduler EESched used the mentioned two facts to select an energy-efficient node for a given task. The metrics used for measuring the energy efficiency of nodes are defined as records/joule and IOPS/watt.

Other than EESched, Yigitbasi et al. ⁷⁸ have proposed two more energy-efficient scheduling heuristics, namely, EESched+Locality and Run Reduce Phase on Wimpy Nodes (RoW). ⁷⁸ In EESched, a TT tt (i.e., worker nodes) is assigned a task t if it is a most energy-efficient task for tt and tt also has the input split required by the task t. In contrast, RoW scheme has been proposed only for scheduling reduce tasks. It schedules the reduce tasks on wimpy nodes that are more energy efficient for I/O bound workloads/tasks. The main advantageous feature of the EESched scheme is that it does not require workload characteristics as a priori. All discussed schedulers in this subsection are summarized in Tables 7 and 8.