Let's explore job priorities by changing them dynamically and watching the result of killing a job.
- Start a relatively long-running job on the cluster.
$ hadoop jar hadoop-examples-1.0.4.jar pi 100 1000 - Open another window and submit a second job.
$ hadoop jar hadoop-examples-1.0.4.jar wordcount test.txt out1 - Open another window and submit a third.
$ hadoop jar hadoop-examples-1.0.4.jar wordcount test.txt out2 - List the running jobs.
$ Hadoop job -listYou'll see the following lines on the screen:
3 jobs currently running JobId State StartTime UserName Priority SchedulingInfo job_201201111540_0005 1 1326325810671 hadoop NORMAL NA job_201201111540_0006 1 1326325938781 hadoop NORMAL NA job_201201111540_0007 1 1326325961700 hadoop NORMAL NA
- Check the status of the running job.
$ Hadoop job -status job_201201111540_0005You'll see the following lines on the screen:
Job: job_201201111540_0005 file: hdfs://head:9000/var/hadoop/mapred/system/job_201201111540_0005/job.xml tracking URL: http://head:50030/jobdetails.jsp?jobid=job_201201111540_000 map() completion: 1.0 reduce() completion: 0.32666665 Counters: 18
- Raise the priority of the last submitted job to
VERY_HIGH.$ Hadoop job -set-priority job_201201111540_0007 VERY_HIGH - Kill the currently running job.
$ Hadoop job -kill job_201201111540_0005 - Watch the other jobs to see which begins processing.
We started a job on the cluster and then queued up another two jobs, confirming that the queued jobs were in the expected order by using hadoop job -list. The hadoop job -list all command would have listed both completed as well as the current jobs and hadoop job -history
would have allowed us to examine the jobs and their tasks in much more detail. To confirm the submitted job was running, we used hadoop job -status
to get the current map and reduce task completion status for the job, in addition to the job counters.
We then used hadoop job -set-priority
to increase the priority of the job currently last in the queue.
After using hadoop job -kill
to abort the currently running job, we confirmed the job with the increased priority that executed next, even though the job remaining in the queue was submitted beforehand.
Manually modifying job priorities in the FIFO queue certainly does work, but it requires active monitoring and management of the job queue. If we think about the problem, the reason we are having this difficulty is the fact that Hadoop dedicates the entire cluster to each job being executed.
Hadoop offers two additional job schedulers that take a different approach and share the cluster among multiple concurrently executing jobs. There is also a plugin mechanism by which additional schedulers can be added. Note that this type of resource sharing is one of those problems that is conceptually simple but is in reality very complex and is an area of much academic research. The goal is to maximize resource allocation not only at a point in time, but also over an extended period while honoring notions of relative priority.
The Capacity Scheduler uses multiple job queues (to which access control can be applied) to which jobs are submitted, each of which is allocated a portion of the cluster resources. You could, for example, have a queue for large long-running jobs that is allocated 90 percent of the cluster and one for smaller high-priority jobs allocated the remaining 10 percent. If both queues have jobs submitted, the cluster resources will be allocated in this proportion.
If, however, one queue is empty and the other has jobs to execute, the Capacity Scheduler will temporarily allocate the capacity of the empty queue to the busy one. Once a job is submitted to the empty queue, it will regain its capacity as the currently running tasks complete execution. This approach gives a reasonable balance between the desired resource allocation and preventing long periods of unused capacity.
Though disabled by default, the Capacity Scheduler supports job priorities within each queue. If a high priority job is submitted after a low priority one, its tasks will be scheduled in preference to the other jobs as capacity becomes available.
The Fair Scheduler segments the cluster into pools into which jobs are submitted; there is often a correlation between the user and the pool. Though by default each pool gets an equal share of the cluster, this can be modified.
Within each pool, the default model is to share the pool across all jobs submitted to that pool. Therefore, if the cluster is split into pools for Alice and Bob, each of whom submit three jobs, the cluster will execute all six jobs in parallel. It is possible to place total limits on the number of concurrent jobs running in a pool, as too many running at once will potentially produce a large amount of temporary data and provide overall inefficient processing.
As with the Capacity Scheduler, the Fair Scheduler will over-allocate cluster capacity to other pools if one is empty, and then reclaim it as the pool receives jobs. It also supports job priorities within a pool to preferentially schedule tasks of high priority jobs over those with a lower priority.
Each of the alternative schedulers is provided as a JAR file in capacityScheduler
and fairScheduler
directories within the contrib directory in the Hadoop installation. To enable a scheduler, either add its JAR to the hadoop/lib directory or explicitly place it on the classpath. Note that each scheduler requires its own set of properties to configure its usage. Refer to the documentation for each for more details.
The alternative schedulers are very effective, but are not really needed on small clusters or those with no need to ensure multiple job concurrency or execution of late-arriving but high-priority jobs. Each has multiple configuration parameters and requires tuning to get optimal cluster utilization. But for any large cluster with multiple users and varying job priorities, they can be essential.