Adaptive task scheduling of Hadoop in a virtualized environment

A control module is introduced to communicate with an application workload scheduler of a distributed computing application, such as a Job Tracker node of a Hadoop cluster, and with the virtualized computing environment underlying the application. The control module periodically queries for resource consumption data, such as CPU utilization, and uses the data to calculate how MapReduce task slots should be allocated on each task node of the Hadoop cluster. The control module passes the task slot allocation to the application workload scheduler, which honors the allocation by adjusting task assignments to task nodes accordingly. The task nodes may also activate and deactivate task slots according to the changed slot allocation. As a result, the distributed computing application is able to scale up and down when other workloads sharing the virtualized computing environment change.

BACKGROUND

Distributed computing platforms, such as Hadoop or other MapReduce-related frameworks, include software that allocates computing tasks across a group, or “cluster,” of distributed software components executed by a plurality of computing devices, enabling large workloads (e.g., data sets) to be processed in parallel and more quickly than is generally feasible with a single software instance or a single device. Such distributed computing platforms typically utilize a distributed file system that can support input/output-intensive distributed software components running on a large quantity (e.g., on the order of thousands) of computing devices to access a large quantity (e.g., petabytes) of data. For example, a data set to be analyzed by Hadoop may be stored within a Hadoop Distributed File System (HDFS) that is typically used in conjunction with Hadoop, which enables various computing devices running Hadoop software to simultaneously process different portions of the file.

SUMMARY

One or more embodiments disclosed herein provide a method for managing resources for a first application comprising a distributed computing application. The method includes receiving a first performance metric for a host computer having a first plurality of virtual machines (VMs) controlled by the distributed computing application and a second plurality of VMs controlled by a second application. The method further includes determining a state of resource contention between the distributed computing application and the second application based on the first performance metric. The method includes modifying an allocation of task slots associated with the host computer based on the first performance metric. Each task slot represents a capability to handle a unit of work for the distributed computing application. The method includes transmitting the modified allocation of task slots to a scheduler, wherein the scheduler is configured to assign a plurality of tasks to be executed in the first plurality of VMs controlled by the distributed computing application based on the modified allocation of task slots.

Another embodiment disclosed herein provides a method for managing resources for a first application comprising a distributed computing application. The method includes receiving an allocation of task slots associated with a host computer having a first plurality of virtual machines (VMs) controlled by the distributed computing application and a second plurality of VMs controlled by a second application. The first plurality of VMs may comprise a first task node. The method further includes determining an assignment of a first task for the distributed computing application to a first task slot on the first task node based on the received allocation of task slots. The received allocation of task slots includes an allocation of task slots for the first task node. The method further includes transmitting, to the first task node, a heartbeat message having the assignment of the first task and the allocation of task slots for the first task node.

Another embodiment disclosed herein provides a method for managing resources of a host computer having a first plurality of virtual machines (VMs) controlled by a distributed computing application and a second plurality of VMs controlled by a second application. The first plurality of VMs may comprise a first task node. The method includes receiving an allocation of task slots for a first task tracker executing on one of the first plurality of VMs, and modifying at least one of a plurality of task slots managed by the first task tracker based on the allocation of task slots. Each task slot may represent a capability to handle a unit of work for the distributed computing application. The method further includes transmitting, to a job tracker, a heartbeat message indicating an updated status and availability of the plurality of task slots.

Further embodiments of the present invention include a non-transitory computer-readable storage medium that includes instructions that enable a processing unit to implement one or more of the methods set forth above or the functions of the computer system set forth above.

DETAILED DESCRIPTION

One or more embodiments disclosed herein provide methods, systems, and computer programs for adaptive task scheduling of a distributed computing application, such as Hadoop, on computing resources in a virtualized environment having a set of virtual machines. Due to the dynamic nature of the consolidated virtual environment, the distributed computing application may need to account for the resources consumed by virtual machines not under its management and for its workload's importance relative to those other virtual machines. A control module queries cluster-wide resource consumption data (e.g., CPU and memory utilization) and detects whether the distributed computing application is contending with other workloads in the virtualized environment for resources. The control may adjust an allocation of “slots” for performing tasks on each node of the distributed computing application based on the availability of memory and CPU resources.

FIG. 1is a block diagram that illustrates a computing system100with which one or more embodiments of the present invention may be utilized. As illustrated, computing system100includes a host group106of host computers, identified as hosts108-1,108-2,108-3, and108-4, and referred to collectively as hosts108. Each host108is configured to provide a virtualization layer that abstracts processor, memory, storage, and networking resources of a hardware platform118into multiple virtual machines (VMs)112that run concurrently on the same host108. The VMs112run on top of a software interface layer, referred to herein as a hypervisor116, that enables sharing of the hardware resources of host108by the VMs112. One example of hypervisor116that may be used in an embodiment described herein is a VMware ESXi hypervisor provided as part of the VMware vSphere solution made commercially available from VMware, Inc.

In one embodiment, VMs112are organized into a plurality of resource pools, identified as resource pool114-1,114-2, and114-3, which logically partitions available resources of hardware platforms118, such as CPU and memory. Resource pools114may be grouped into hierarchies; resource pools114provide resources to “child” resource pools and virtual machines. Resource pools114enable a system administrator to organize resources of computing system100, isolate VMs and computing resources from one resource pool to another, abstract resources from the actual hosts108that contribute the resources, and manage sets of VMs112associated with a resource pool114. For example, a system administrator may control the aggregate allocation of resources to the set of VMs112by changing settings on the VMs' enclosing resource pool114.

As shown, VMs112of hosts108may be provisioned and used to execute a number of workloads that deliver information technology services, including web services, database services, data processing services, and directory services. Some VMs112may be used to execute a workload of a distributed computing application124, while other VMs112may be used for other workloads122. In one embodiment, one or more VMs112are configured to serve as a node128of a cluster134generated and managed by a distributed computing application124configured to elastically distribute its workload over the nodes128. Distributed computing application124may be configured to receive processing requests, and distribute the processing requests to available nodes128for processing. VMs112executing as nodes128on host108are shown in greater detail inFIG. 2.

FIG. 2is a block diagram that illustrates a host computer108supporting one or more virtual machines112, according to one embodiment of the present invention. As shown, hardware platform118of each host108may include conventional components of a computing device, such as a memory202, a processor204, local storage206, a storage interface208, and a network interface210. Processor204is configured to execute instructions, for example, executable instructions that perform one or more operations described herein and may be stored in memory202and in local storage206. Memory202and local storage206are devices allowing information, such as executable instructions, cryptographic keys, virtual disks, configurations, and other data, to be stored and retrieved. Memory202may include, for example, one or more random access memory (RAM) modules; local storage206may include, for example, one or more hard disks, flash memory modules, solid state disks, and optical disks. Storage interface208enables host108to communicate with one or more network data storage systems that may, for example, store “virtual disks” that are accessed by VM nodes. Examples of storage interface208are a host bus adapter (HBA) that couples host108to a storage area network (SAN) or a network file system interface. Network interface210enables host108to communicate with another device via a communication medium, such as network110. An example of network interface210is a network adapter, also referred to as a Network Interface Card (NIC). In some embodiments, a plurality of NICs is included in network interface210.

As described earlier, virtual machines (e.g., VMs112-1to112-N) run on top of a hypervisor116that enables sharing of the resources of hardware platform118of host108by the virtual machines. Hypervisor116may run on top of the operating system of host108or directly on hardware components of host108. Hypervisor116provides a device driver layer configured to map physical resource of hardware platforms118to “virtual” resources of each VM112such that each VM112-1to112-N has its own corresponding virtual hardware platform (e.g., a corresponding one of virtual hardware platforms214-1to214-N). Each such virtual hardware platform214provides emulated hardware (e.g., memory202A, processor204A, local storage206A, networked storage208A, network interface210A, etc.) that may, for example, function as an equivalent, conventional hardware architecture for its corresponding VM112. Virtual hardware platforms214-1to214-N may be considered part of virtual machine monitors (VMMs)212-1to212-N which implement virtual system support to coordinate operations between hypervisor116and corresponding VMs112-1to112-N. In the embodiment depicted inFIG. 2, each VM112includes a guest operating system (OS)216(e.g., Microsoft Windows, Linux) and one or more guest applications running on top of guest OS216. In one embodiment, each VM112includes a runtime environment218, such as a Java Virtual Machine (JVM), that supports execution of a distributed software component code220(e.g., Java code) for distributed computing application124. For example, if distributed computing application124is a Hadoop application, a VM112may have a runtime environment218(e.g., JVM) that executes distributed software component code220implementing a “Job Tracker” function, “TaskTracker” function, “Name Node” function, “Secondary Name Node” function, and “Data Node” function. In another embodiment of distributed computing application124having a next-generation Hadoop data-processing framework (i.e., YARN), a VM112may have a runtime environment218(e.g., JVM) that executes distributed software component code220implementing a “Resource Manager” function (which includes a workload scheduler function), “Node Manager” function, “Task Container” function, “Application Master” function, “Name Node” function, “Data Node” function, and “Journal Node” function. Alternatively, each VM112may include distributed software component code220for distributed computing application124configured to run natively on top of guest OS216.

Referring back toFIG. 1, computing system100includes a virtualization management module130that may communicate to the plurality of hosts108via network110. In one embodiment, virtualization management module130is a computer program that resides and executes in a central server, which may reside in computing system100, or alternatively, running as a VM in one of hosts108. One example of a virtualization management module is the vCenter® Server product made available from VMware, Inc. Virtualization management module130is configured to carry out administrative tasks for the computing system100, including managing hosts108, managing VMs running within each host108, provisioning VMs, migrating VMs from one host to another host, load balancing between hosts108, creating resource pools114comprised of computing resources of hosts108and VMs112, modifying resource pools114to allocate and de-allocate VMs and physical resources, and modifying configurations of resource pools114. In one embodiment, virtualization management module130is configured to communicate with hosts108to collect performance data and generate performance metrics (e.g., counters, statistics) related to availability, status, and performance of hosts108, VMs112, and resource pools114.

In one embodiment, virtualization management module130is configured to provide virtual environment scheduler functionality that balances the VMs across hosts108of the host group106. For example, if the resource usage on one of the VMs in a resource pool drastically changes, the virtualization management module130moves around VMs among the physical hosts to optimize distribution of virtual machines across the hosts. Further, if the overall workload of all VMs decreases, the virtualization management module130may power down some of the physical hosts and consolidate the VMs across the remaining physical hosts. One example of a virtual environment scheduler is the VMware Distributed Resource Scheduler (DRS®) product made available from VMware, Inc.

As mentioned above, distributed computing application124is configured to elastically distribute its workload over a plurality of nodes128managed by the distributed computing application124. In one embodiment, distributed computing application124includes an application workload scheduler126(e.g., executing in a VM112) configured to manage execution of workloads running one or more within VM nodes128controlled by the distributed computing application. During operation, application workload scheduler126may query VM nodes128allocated to the distributed computing application to determine their status and the availability of resources for processing additional workloads. For example, application workload scheduler126may query VMs112allocated to the distributed computing application to determine if the VMs are up, and if they are up, determine available resources from each VM for executing a portion of the workload performed by the distributed computing application as a whole.

Conventional workload schedulers for distributed computing applications (e.g., Hadoop application) are designed to manage execution of a workload on a dedicated set of physical computing elements, under an assumption that the full set of dedicated computing resources (e.g., memory and CPU) are available, as well as based on other pre-determined attributes that are relevant to the application workload's performance, including data storage and networking locality. However, such an application workload scheduler may face challenges when attempting to schedule execution of a workload within a virtualized environment, as depicted inFIG. 1, that may have computing resources consumed by other VMs not controlled by the distributed computing application. For example, application workload scheduler126may accept jobs (e.g., from a user) for execution by distributed computing application124within the virtualized environment of computing system100. Application workload scheduler126may then schedule execution a received job within VM nodes128by splitting the job into small tasks and distributing the tasks, a process sometimes referred to as task placement, on the nodes128based on a scheduling or placement policy. Scheduling and placement policies typically factor in CPU and memory utilization of each node, for example, to balance use of computing resources. However, the scheduled use of computing resources for the distributed application workload does not factor in resource contention with other workloads running within the virtualized environment. For sake of discussion, a workload for distributed computing application124may be referred to herein as a “distributed application workload” or “application workload”, where as other workloads running within the computing system100that are not for distributed computing application124(e.g., workload122) may be referred to as “non-distributed application workloads” or “non-application workloads”. For example, application workload scheduler126may be attempting to place a task based on availability of nodes128while the virtualization management module130is attempting to load balance both application workload VMs and non-application workload VMs across hosts108, thereby resulting in inconsistent results. In another example, other applications may suffer a slowdown in performance when executed within a same virtualized environment as the distributed computing application, which is undesirable for higher-priority, business-critical application.

Accordingly, embodiments of the present invention provide a control module132configured to communicate with application workload scheduler126to automatically adjust task allocation based on resource availability in a shared virtualized environment. In one embodiment, control module132is configured to communicate (e.g., via API call) with virtualization management module130to obtain performance metrics of resources in VMs112, resource pools114, and hosts108, and determine a state of resource contention between the application workloads and non-application workloads. Control module132is further configured to communicate with application workload scheduler126to determine how available task “slots” for performing work in the distributed computing application may be allocated among nodes128based on the resource metrics. In one embodiment, control module132is configured to maintain a counter of the number of task slots allocated to the distributed computing application, including a count of the number of task slots allocated to each host108and a count of the number of task slots allocated to each node128executing on a particular host108.

While control module132is depicted inFIGS. 1 and 3as a separate component that resides and executes on a separate server or virtual machine, it is appreciated that control module132may alternatively reside in any one of the computing devices of the virtualized computing system100, for example, such as the same central server where the virtualization management module130resides. In one implementation, control module132may be embodied as a plug-in component configured to extend functionality of virtualization management module130. Further, while control module132is depicted as a separate entity from application workload scheduler126, it should be appreciated that in some alternative embodiments, some or all of functionality of control module132may be incorporated into application workload scheduler126.

Example Hadoop Application with Adaptive Task Scheduling

FIG. 3is a block diagram that illustrates a virtualized computing system300having a control module132configured to support an example distributed computing application, according to one embodiment of the present invention. In the embodiment shown inFIG. 3, the distributed computing application is a Hadoop application302configured to process a large set of data using a distributed set of workload nodes (e.g., VMs112) allocated to Hadoop application302. It should be recognized that alternative architectures for a Hadoop application having some form of an application workload scheduler may be utilized with the techniques described herein. It should be further recognized that, while embodiments of present invention are described in terms of a Hadoop installation, other distributed computing applications, such as web applications having a front end scheduler or large scalable database system (e.g., MongoDB, Apache Cassandra), may be configured and utilized according to the techniques provided herein.

In one embodiment, Hadoop application302includes an application workload scheduler, referred to herein as a job tracker304, which accepts Hadoop jobs from clients and schedules corresponding workloads for execution on a plurality of task nodes306that are part of Hadoop application302. Each task node306(e.g., a VM112) is a worker node that carries out tasks (e.g., map tasks, reduce tasks of a MapReduce job) provided by job tracker304. Each task node306may handle multiple tasks in parallel. As shown, each task node306may have a number of available task slots308that represent a capability to handle a unit of work for the Hadoop application. Each task (e.g., map or reduce task) performed by the node takes up one task slot308. In some embodiments, each task slot308may be implemented as an instance of a runtime environment (e.g., Java Virtual Machine) executing distributed software component code (e.g., code220) for completing a single task. As such, each task node306may execute multiple instances of the runtime environment to execute in parallel multiple tasks assigned to the task node by the job tracker304.

When job tracker304receives a request to execute a job within Hadoop application302, job tracker304considers what resources (e.g., task slots308) should be considered as available for executing the requested job and the availability of those resources on a per-node basis. Additionally, job tracker304tracks task slot308availability and other scheduling invariants when determining how to distribute work among the task nodes306. If a task node306fails due to software error, network problems, or other issues, job tracker304is able to adjust its scheduling of the application workload accordingly. For example, job tracker304may mark failed task nodes as “unavailable” for accepting tasks, and schedule placement of subsequent tasks to other slots in other task nodes based on the reduced amount of available resources.

According to one embodiment, the control module132is configured to communicate with the job tracker304to adjust task scheduling based on available resources in the virtualized computing system300as determined to be available by metrics provided by virtualization management module130. In some embodiments, the control module132and job tracker304coordinate to determine a modified number and distribution of task slots308within task nodes306based on resource contention detected in hosts108between application workloads and non-application workloads. As such, the control module132and job tracker304may change the number of tasks that a task node306can handle simultaneously depending on the resource load of its underlying host108.

In one embodiment, a task node306includes a per-node agent referred to as a task tracker310configured to manage execution of multiple tasks in parallel in one or more task slots308in coordination with job tracker304. In one embodiment, each task tracker310is configured to communicate with the centralized job tracker304through the exchange of a periodic “heartbeat” message or other suitable message bus. Using the heartbeat message, a task tracker310may transmit task status (e.g., progress) and slot availability to job tracker304, and receive modified slot allocations, as well as new task assignments, from job tracker304. In one embodiment, a task tracker310is configured to modify a number of available task slots308based on slot allocations received from job tracker304via the heartbeat message.

FIG. 4is a flow diagram that illustrates steps for a method400of managing a distributed computing application within a virtualized environment, according to an embodiment of the present invention. It should be recognized that, even though the method400is described in conjunction with the system ofFIGS. 1 and 3, any system configured to perform the method steps is within the scope of embodiments of the invention.

The method400begins at step402, where the control module132retrieves performance metrics (e.g., via an API call) from virtualization management module130that indicate a level of resource consumption by VMs112executing on hosts108. Control module132may retrieve the performance metrics on a periodic basis, for example, every 5 seconds, to continually monitor resource availability for the Hadoop application. Control module132may track which VMs112within the virtualized computing system are task nodes controlled by the Hadoop application and retrieve performance metrics relating to those VMs112. In one embodiment, the performance metrics may include metrics describing CPU and memory utilization per host108, per VM112, per resource pool114, and per resource of hardware platform118. In some embodiments, the performance metrics may include disk metrics of I/O performance, such as latency, read/write speeds, storage bandwidth consumption, and disk utilization per host, VM, or datastore. In some embodiments, the performance metrics may include performance statistics related to virtual environment scheduler functionality configured for clusters and resource pools114by virtualization management module130. For sake of discussion, embodiments are described relating to resource contention of CPU resources, though it should be recognized that embodiments of the present invention may be utilized to take into account memory utilization, storage bandwidth consumption, and other resources.

At step404, control module132detects a state of resource contention between workloads for the distributed computing application (e.g., Hadoop workloads) and non-distributed computing application workloads (e.g., non-Hadoop workloads122) sharing the same computing resources of the physical hosts. In some embodiments, control module132may deem a state of resource contention if an overall system load for one or more physical hosts running VMs controlled by Hadoop application302exceeds for a certain threshold (e.g., 70% CPU load). Control module132may further check whether the overall system load for physical hosts discounting CPU resource consumption by VMs controlled by Hadoop application302exceeds a certain threshold. As such, control module132may deem there is no resource contention if VMs controlled by Hadoop application302are the primary contributor to the system load. In some embodiments, multiple threshold values may be used for detecting resource contention, as described in greater detail later in conjunction withFIG. 6.

At step406, responsive to determining a state of resource contention, control module132determines changes to task slot allocation based on the performance metrics. Control module132may determine changes from an initial slot allocation given to each task node306during provisioning of the Hadoop application. In one embodiment, the initial number of task slots that a task node may be allocated may be determined based on the amount of virtual resources, such as a number of virtual CPU cores and amount of guest physical memory, the task node has been allocated. In one example, each task node306(e.g., embodied as a VM112) may be given an initial task slot allocation of one task slot308per virtual CPU core allocated to the task node VM. As such, a task node VM having 4 virtual CPU cores may initially start with 4 task slots308for processing Hadoop jobs. It should be recognized that other alternative initialization schemes may be used. For example, each task node may be given an initial number of task slots (e.g., one task slot), and if resource contention is not detected during operation, the initial number of tasks slots for each task node may be doubled until a threshold value (e.g., a resource contention threshold t1described below) is reached. In some cases, the initial number of task slots may be increased until reaching a certain threshold value t, and then the system performs an additive increment after the threshold value t.

In one embodiment, control module132may calculate a reduced number of task slots for a given task node based on the performance metrics. In some embodiments, control module132may identify one or more task nodes running on a particular physical host under resource contention and calculate a reduced number of task slots for those task nodes. In some cases, control module132may calculate a reduced number of task slots for one or more task nodes running in a resource pool of which the underlying physical hosts are under resource contention. In other cases, control module132may calculate a reduced number of slots for all task nodes controlled by the Hadoop application to reduce the overall resource consumption of the Hadoop application within the virtualized computing system. In some embodiments, if resource contention continues, control module132may continue to reduce the number of task slots allocated for task nodes. Once control module132detects that resource contention is abating, control module132may calculate an increased number of slots for task nodes, as discussed later.

At step408, control module132transmits the updated task slot allocation to job tracker304, which receives the updated task slot allocation, at step410. In some embodiments, the updated task slot allocation may indicate an updated number of slots for a specified task node. In other embodiments, the updated task slot allocation may indicate an updated number of slots per each task node.

At step412, job tracker304transmits a heartbeat message having the updated task slot allocation to one or more task trackers310. In one embodiment, job tracker304may transmit the heartbeat message at a periodic interval having a duration selected to enable the job tracker to effectuate changes in task slot allocation responsively to resource contention. For example, job tracker304may transmit a heartbeat message (and seek to receive a heartbeat response) every 5 seconds. At step414, one of task trackers310receives the heartbeat message.

At step416, task tracker310(e.g., executing on a task node) modifies the task slot allocation for that task node based on the received heartbeat message. In one embodiment, the task tracker310updates an internal counter indicating the number of task slots made available at the task node for performing work. The update may result in an increase in the number of task slots or a decrease in the number of task slots available at the task node.

In one embodiment, the new number of task slots (e.g., as indicated by the received heartbeat message) may be less than the current number of task slots. In this case, task tracker310marks one or more task slots as “inactive” to reach the new number of task slots and does not advertise their availability (e.g., in a heartbeat response back to the job tracker). Task tracker310may select and mark as inactive any task slot(s) currently unoccupied by a task, if there are any. In embodiments where each slot is implemented as an instance of a runtime environment (e.g., JVM) executing distributed software component code, task tracker310may suspend operation of the runtime environment instances corresponding to inactive slots. In other embodiment, task tracker310may simply allow the runtime environment instance corresponding to a task slot to idle, for example, in cases where CPU resources are highly contended.

In some cases, multiple task slots are already occupied with tasks, and task tracker310has insufficient free task slots to mark as inactive to satisfy the new number of task slots (i.e., the new number of task slots is less than a number of currently occupied task slots). In one embodiment, task tracker310selects and marks one or more occupied task slots as inactive. In some embodiments, task tracker310permits the work corresponding to the selected occupied task slots to reach completion, and does not advertise its availability after completion. In other embodiments, task tracker310preemptively removes the selected occupied task slot, for example by killing the task and aborting the work corresponding to the selected occupied task slot.

In one embodiment, the new number of task slots (e.g., as indicated by the received heartbeat message) may be greater than the current number of task slots. In this case, task tracker310adds additional slots to reach the new number of task slots. In some embodiments, task tracker310may remove an “inactive” status from one or more previously marked task slots and resume advertising their availability (e.g., to job tracker304).

At step418, task tracker310transmits a heartbeat response to job tracker304indicating the updated slot status and availability. The heartbeat message announces the availability of the task node and provides job tracker304with a status report of tasks in progress on the task node. In one embodiment, the heartbeat message may indicate an updated number of open task slots available for performing work that takes into account the updated task slot allocation. In some embodiments, the heartbeat message may include the current number of task slots at the task node and the number of “occupied” task slots at the task node, and may further confirm the number of task slots received from job tracker304in a previous heartbeat message.

At step420, job tracker304receives the heartbeat response from one or more task tracker(s). At step422, job tracker304determines task scheduling based on the received task slot allocation based on the updated task slot allocation and slot availability (e.g., as provided by task trackers310and control module132). Job tracker304assigns tasks to be performed by each task node with an available slot according to any of a variety of scheduling algorithms, such as first-in-first-out (FIFO), fair scheduling, and capacity scheduling. Conventional Hadoop schedulers are configured to run a fixed set of physical computers and schedule tasks based on a static number of slots pre-defined in one or more configuration files processed during start-up. According to one embodiment, job tracker304assigns and schedules tasks in available slots of task nodes306while adjusting the task assignments to honor the updated number of task slots received from control module132. For example, if job tracker304receives an update indicating a given task node's task slot count is reduced, job tracker304may schedule fewer tasks to be performed by that task node.

At step424, job tracker304transmits a heartbeat message to task tracker310that includes the determined task assignments for that task node as well as any new updated task slot allocations (e.g., as provided by control module132). At step426, task tracker310receives the heartbeat message indicating new task assignments and may proceed to run the assigned tasks with available task slots, for example, by processing work within an instance of a runtime environment corresponding to the task slot.

FIGS. 5A-5Bis a block diagram that illustrates an example of dynamically allocating task slots, according to one embodiment of the present invention. In the embodiment shown inFIG. 5A, VMs112-1and112-2are configured as task nodes for Hadoop application302and include task trackers310-1and310-2, respectively. Task tracker310-1includes one active task slot308-1and two inactive task slots308-2and308-3(depicted in dash lines), while task tracker310-2includes three active task slots308-4,308-5, and308-6. Task slots308-1,308-4,308-5are also currently occupied with a task (depicted with hash marks). As shown, job tracker304transmits a heartbeat message to both task trackers indicating an updated allocation of two task slots (e.g., newNumSlots=2).

FIG. 5Bshows changes at the task trackers responsive to the updated task slot allocation. To reach the increased number of task slots, task tracker310-1marks as active task slot308-2to reach the new number of task slots (e.g., 2 task slots) as active. Task slot308-2may be advertised in the heartbeat response as available for accepting a task assignment from job tracker304. Task slot308-3remains inactive. Task tracker310-1transmits a heartbeat response to job tracker304indicating the current number of tasks slots (i.e., currentNumSlots=2), the number of occupied task slots (i.e., occupiedNumSlots=1), and a confirmation of the number of task slots received in a previous task slot allocation (i.e., receivedNumSlots=2). With regards to task tracker310-2, one task slot must be marked inactive to reach the decreased number of task slots. As shown, the task tracker310-2selects task slot308-6which is currently unoccupied and marks the task slot as inactive. Task tracker310-2transmits a heartbeat response to job tracker304indicating the current number of task slots (i.e., currentNumSlots=2), the number of occupied task slots (i.e., occupiedNumSlots=2), and a confirmation of the number of task slots received in a previous heartbeat message (i.e., receivedNumSlots=2).

According to one embodiment, once control module132detects resource contention is appearing or disappearing, the control module may calculate task nodes' task slots accordingly. As such, control module132may increase the task slot allocation to prevent resources of the system from being under-utilized. To avoid thrashing and other cyclical issues, multiple threshold values may be used for detecting resource contention. In one embodiment, two threshold values t1and t2(t1>t2) may be used to detect resource contention, and two threshold values t3and t4(t3<t4) may be used to detect resource contention relief, as shown inFIG. 6. In some embodiments, the threshold values t1, t2, t3, t4may be selected such that the threshold for resource contention is higher than the threshold for resource contention relief (i.e., t1>t3), multiple thresholds for resource contention are higher than the threshold for resource contention relief (i.e., t3<t2<t1), and multiple thresholds for resource contention relief are lower than the threshold for resource contention relief (i.e., t3<t4<t1).

FIG. 6is a flow diagram that illustrates steps for a method600for calculating the number of task slots based on resource utilization, according to an embodiment of the present invention. It should be recognized that the steps of method600correspond to sub-steps of one embodiment of steps404and406ofFIG. 4(i.e., steps for detecting a state of resource contention and determining changes to task slot allocation). It should be further recognized that, even though the method is described in conjunction with the system ofFIGS. 1 and 3, any system configured to perform the method steps is within the scope of embodiments of the invention.

The method600begins at step602, where control module132detects when CPU utilization on one or more hosts108goes above a first threshold value t1that defines an upper threshold of resource contention. If the CPU utilization exceeds t1, control module132deems there to be a state of resource contention within the one or more hosts and, at step604, reduces the number of available Hadoop task slots on that host. It is appreciated that CPU utilization may decrease when the number of task slots (and therefore the number of tasks running in parallel for the Hadoop application) is reduced, and conversely, CPU utilization may increase when the number of task slots is increased.

In one embodiment, control module132may continue to reduce the task slot number until either the resource contention is no longer detected (e.g., as defined by threshold t2), or the number of task slots on that host has been reduced to zero. As shown inFIG. 6, at step606, if no task slots remain for the host, control module132may infer that the distributed computing application workload is not contributing to the high CPU load because Hadoop application is not consuming any resources on the host. Control module132may proceed to step610to wait until CPU utilization returns to normal (e.g., a spike from other workloads passes).

Otherwise, if there are still task slots left on the host, at step608, control module132detects whether CPU utilization has dropped below a second threshold value t2that defines a lower threshold of resource contention. If CPU utilization has been reduced to threshold value t2, control module132returns to the start of method600to continue monitoring CPU utilization. If CPU utilization has not been reduced to less than or equal to the second threshold value t2, control module132determines this to be a continued state of resource contention and continues to (e.g., at step604) reduce the number of task slots for that host.

In some embodiments, control module132may reduce the number of task slots associated with the host determined to have resource contention by a fixed increment (e.g., reduced by 2 slots each time). In other embodiments, control module132may dynamically change the increment by which the number of task slots is reduced. Control module132may reduce the number of task slots associated with the host determined to have resource contention in an exponential manner to rapidly relieve any resource contention on the host caused by the Hadoop workload. For example, control module132may at first reduce the number of task slots by 8 slots, and when resource contention persists, reduce by 16 slots, then 32 slots, and so forth.

In one embodiment, control module132may use threshold values (e.g., t3and t4) to detect a state of resource contention relief and decide when Hadoop task slots may be reactivated and restored to full use. At step610, if control module132does not detect or no longer detects resource contention, control module132determines whether CPU utilization is less than a third threshold value t3defining a lower bound of contention relief. If so, control module132determines that the system may be under-utilized, and at step612, increases available Hadoop task slots on that host. In some embodiments, control module132may incrementally increase the number of tasks slots associated with the under-utilized host by a fixed increment (e.g., increase by 2 slots each time). In other embodiments, control module132may dynamically change the increment by which the number of task slots is increased (e.g., increased exponentially.)

Control module132may continue to increase the task slot number until either the system is no longer under-utilized (e.g., as defined by threshold t4), or until the resources consumed by Hadoop application302reaches the total amount of resources allocated to task nodes at the time of VM provisioning. In some embodiments, control module132may determine the system is no longer under-utilized when CPU utilization exceeds a fourth threshold value t4that defines an upper bound of contention relief, such as at step614. Accordingly, embodiments of the present invention enable the Hadoop application to shrink down gracefully when business-critical applications are present and to scale up elegantly during low demand periods to maintain a high level of utilization across the virtualized computing system.

In one embodiment, the rate of increase and rate of decrease in the number of task slots may be based on a lower priority level of application workloads relative to non-application workloads. To prevent Hadoop tasks from business-critical application, control module132may be conservative when increasing the number of task slots and aggressive when reducing task slots. According to one embodiment, as shown inFIG. 6, control module132may decrease slots exponentially when contention is detected, thereby causing Hadoop to quickly give up resources to other contending applications, and increase slots incrementally to avoid re-triggering resource contention and cyclical effects.

According to one embodiment, control module132may maintain a log that records when resource contention is detected, what changes to task slots are made, and when resource contention is relieved. Control module132may incorporate this historical log in its decision making process when determining changes (e.g., how many slots, which slots) to task slots. In one embodiment, control module132tracks the increase and decrease in the number of task slots and corresponding changes to Hadoop nodes resource consumption, and uses the data to learn how fast it should increase and decrease the task slots. For example, if a control module detects a host's CPU consumption is above target 10%, and historical data shows decreases one slot in average decreases 2% CPU utilization, the control module may determine to reduce by 5 task slots (i.e., 10%/2%=5). As such, control module132may be self-learning and learn how changes to the Hadoop application (e.g., changes in slot allocation) affect a particular infrastructure (e.g., computing system100,300) hosting the Hadoop application. Accordingly, embodiments of the present invention enable a distributed computing application having a control module132to advantageously adapt to a variety of virtualized computing systems, which could have different hardware resources, different system requirements, and shared workloads.

While method600describes a control flow for controlling CPU resources, it should be recognized that the techniques described may be applied to other computing resources, such as memory utilization, storage bandwidth, I/O resources, etc. In one embodiment, control module132may include separate workflows (e.g., method600) for CPU and memory utilization which may independently cause changes to the number of task slots. In some embodiments, at times when both resource utilizations (e.g., CPU and memory) indicate changes are need to the task slot number, control module132may choose a smaller slot number of the two determined number of task slots.