E-MapReduce:YARN schedulers

YARN schedulers are pluggable plug-ins. YARN provides the following types of schedulers: First-In-First-Out (FIFO) Scheduler, Fair Scheduler and Capacity Scheduler.

• FIFO Scheduler is the simplest scheduler. It does not allow applications to be submitted to multiple tenants for scheduling. This indicates that all applications are submitted to the default queue for scheduling without cluster resource allocation. In this case, applications are scheduled based on the FIFO policy without other control and scheduling configurations. FIFO Scheduler is suitable only for simple scenarios and is rarely used in production environments.

• Fair Scheduler is the default scheduler that is used by Cloudera Distributed Hadoop (CDH). After CDH is merged with Hortonworks Data Platform (HDP), Cloudera Data Platform (CDP) comes into being and uses Capacity Scheduler instead of Fair Scheduler. If you use Apache Hadoop, Capacity Scheduler is also recommended. Fair Scheduler provides the multi-tenant management and resource scheduling capabilities. The capabilities include multi-level queue management, quota management, access control list (ACL)-based control, resource sharing, fair scheduling among tenants, application scheduling within tenants, resource reservation and preemption, and asynchronous scheduling. However, the development of Fair Scheduler is not as up-to-date as the development of Capacity Scheduler in Apache Hadoop. Fair Scheduler does not consider the resource layout of clusters and does not support scheduling features such as node labels, node attributes, and placement constraints.

• Capacity Scheduler is the default scheduler that is used by Apache Hadoop, HDP, and CDP. Capacity Scheduler provides complete multi-tenant management and resource scheduling capabilities, which include all capabilities of Fair Scheduler. Capacity Scheduler can optimize the resource layout of clusters based on global scheduling, reduce the risk of hotspot issues, and maximize the use of cluster resources. Capacity Scheduler also supports scheduling features such as node labels, node attributes, and placement constraints.

MainScheduler of Capacity Scheduler can be triggered by using one of the following methods:

• Node heartbeat-driven: When a capacity scheduler receives the heartbeat of a node, MainScheduler is triggered to select an application that can be scheduled for the node. This type of scheduling is local scheduling of nodes. MainScheduler is subject to heartbeat intervals and is similar to random scheduling. As a result, a large number of nodes may not be hit because of insufficient resources and unmet scheduling requirements. This causes low scheduling efficiency. This method is suitable for clusters that do not have high requirements on scheduling performance and scheduling features.

• Asynchronous scheduling: MainScheduler is triggered by an asynchronous thread or multiple parallel threads. A random node is selected from the node list for scheduling. This improves the scheduling performance. This method is suitable for clusters that have high requirements on scheduling performance but low requirements on scheduling features.

• Global scheduling: MainScheduler is triggered by a global thread. This type of scheduling is global scheduling of applications. Applications are selected for scheduling based on factors such as multi-tenant fairness and application priority. Then, nodes are selected for scheduling based on factors such as the resource size, scheduling requirements, and resource distribution of the applications. This way, optimal scheduling decisions can be made. This method is suitable for clusters that have high requirements on scheduling performance and scheduling features.

The following figure shows the architecture of global scheduling based on YARN 3.2 or later.

• MainScheduler is an asynchronous multithreading processing framework based on resource requests. MainScheduler contains allocation and submission threads. One or more allocation threads are used to locate the resource request that has the highest priority, select appropriate candidate nodes based on the resource size and placement constraints, generate allocation proposals, and arrange the allocation proposals in the intermediate queue. A submission thread is used to consume the allocation proposals, recheck various placement constraints, and then submit or reject the allocation proposals. The submission thread can also be used to update the status of a capacity scheduler.

• ReScheduler is a dynamic resource monitoring framework that periodically runs. ReScheduler contains multiple resource monitoring policies, such as policies that are related to inter-queue preemption, intra-queue preemption, and reserved resource preemption.

• Node Sorting Manager and Placement Constraint Manager are global scheduling plug-ins of MainScheduler. The plug-ins are used for load balancing and management of complex placement constraints.

The following figure shows the global scheduling process of Capacity Scheduler.

• Container allocation process of MainScheduler:

  1. Select partitions (node labels). A cluster may have one or more partitions. MainScheduler allocates containers to partitions in turn.

  2. Select leaf queues. MainScheduler traverses queues from the root queue. Child queues at the same level are traversed in ascending order of the percentage of guaranteed resources that are allocated to queues until the leaf queues are found. In the preceding figure, a red queue indicates that the resource usage is high, and a green queue indicates that the resource usage is low. Resources are preferentially allocated to green queues.

  3. Select applications. MainScheduler selects applications within a queue based on the fair or FIFO policies. If the fair policy is used, applications are selected in the ascending order of the amount of memory resources that are allocated to the applications. If the FIFO policy is used, applications are selected in the descending order of priority and ascending order of application IDs.

  4. Select requests. MainScheduler selects requests based on the priority.

  5. Select sorted candidate nodes. MainScheduler searches for the candidate nodes that meet specific resource requirements from all sorted nodes based on the request placement constraints.

  6. Traverse the candidate nodes and allocate a container to each candidate node. During the allocation process, MainScheduler checks the allocated, used, or unconfirmed resources of a queue or a node. After the check passes, an allocation proposal is generated and put into the proposal queue.

• Container submission process of MainScheduler: A submission thread checks the allocation proposal in the proposal queue. In this case, the submission thread checks whether the requirements on applications, nodes, and placement constraints are met. If requirements are not met, the allocation proposal is discarded. If the requirements are met, the allocated container takes effect, and the resources of applications or nodes are updated.

• Resource preemption process of ReScheduler: If the total amount of available resources of a cluster is less than a specific resource threshold and the resources that are allocated to specific applications are insufficient, ReScheduler may trigger resource preemption by using different methods.

  • Inter-queue preemption: YARN allows unused guaranteed resources of a queue to be shared by other queues. When an application in the queue whose guaranteed resources are not sufficient requires more resources, but the cluster has no idle resources, inter-queue preemption is triggered. The resources within the queue capacity are guaranteed resources, and the resources that exceed the queue capacity but are less than the maximum queue capacity are shared resources.

  • Intra-queue preemption: ReScheduler monitors and adjusts the resources of applications based on the FIFO or fair policy of queues. When applications that have a high priority require resources but the resources that are allocated to the queue are exhausted, intra-queue preemption is triggered.

  • Reserved resource preemption: When a task meets specific conditions, the task and the resources that are reserved for the task are released. For example, if resources are not allocated to a task within the timeout period, the task and resources that are reserved for the task are released.

Capacity Scheduler provides the following features:

• Multi-level queue management: Capacity Scheduler allows you to manage multiple levels of queues in multiple tenants. The resource quota of a parent queue limits the resource usage of all its child queues, and the resource quota of a single child queue cannot exceed the resource quota of the parent queue. This ensures controllable multi-tenant management and meets the requirements of various complex application scenarios.

• Resource quota: Capacity Scheduler allows you to configure the resources that are guaranteed for a queue and the maximum amount of resources that can be used by a queue. You can also configure the maximum number of applications that can run on a queue, the maximum number of resources that ApplicationMaster can use, and the ratio of resources that a user uses to the configured resources. This way, you can control queues from multiple dimensions, such as applications, task types, and users.

• Elastic resource sharing: The amount of elastic resources in a queue can be calculated by using the following formula: Maximum queue capacity - Queue capacity. When a cluster and the parent queue have idle resources, child queues can use the unused guaranteed resources of other queues. This ensures schedule-based reuse of resources in different queues and improves the resource utilization of the cluster.

• ACL-based control: Capacity Scheduler allows you to manage permissions on queues. You can specify multiple users to submit and manage tasks, or specify the same user to manage multiple queues.

• Inter-tenant queue scheduling: Queues at the same level are scheduled in the ascending order of the percentage of guaranteed resources of queues. This ensures that queues that require fewer guaranteed resources are scheduled first. If you configure the priorities of queues, queues at the same level are divided into two groups: queues whose used resources are less than or equal to the guaranteed resources and queues whose used resources are greater than the guaranteed resources. Resources are first allocated to the queues whose used resources are less than or equal to the guaranteed resources. Then, all queues in the two groups are sorted in the descending order of priority and in the ascending order of the percentage of guaranteed resources.

• Intra-tenant application scheduling: Applications in a tenant are scheduled based on the FIFO or fair policy. If applications are scheduled based on the FIFO policy, all applications in a tenant are scheduled in the descending order of priority and the ascending order of submission time. If applications are scheduled based on the fair policy, all applications in a tenant are scheduled in the ascending order of the percentage of resources that are used by the applications and the ascending order of submission time.

• Preemption: The resources of clusters are constantly changing. To meet the requirements of elastic resource sharing, fair intra-tenant queue scheduling, and inter-tenant application scheduling, the preemption process is used to balance the resource usage of queues and applications.

Basic queue configurations include the configurations of the following items: hierarchical relationships of queues, capacity of queues, and the maximum capacity of queues. If a YARN cluster is shared by multiple departments and teams of an enterprise , you can organize YARN queues based on a specific organization mode and expected percentage of resources. The parent queue root has four child queues, as shown in the following figure: dev, test, support, and default. The percentages of guaranteed resources for the queues are 50%, 30%, 10%, and 10%, and the percentages of the maximum amount of resources for the queues are 100%, 50%, 30%, and 100% . The dev queue has two child queues: training and services. The percentages of guaranteed resources for the child queues are 40% and 60%, and the percentages of the maximum amount of resources for the child queues are both 100%.

The following table describes the basic configurations of queues.

The list on the YARN Scheduler page displays the hierarchy of queues, the percentage of guaranteed resources, and the percentage of the maximum amount of resources for the queues. In the following figure, the solid gray box represents the percentage of the maximum amount of resources of each queue, and the dashed box represents the percentage of guaranteed resources of each queue to the cluster resources. You can expand a queue to view the details of the queue, such as the queue status, resource information, application information, and other configurations, as shown in the following figure.

The following table describes the ACL configurations of queues. In most cases, you can disable ACL. You can enable ACL based on your business requirements.

Preemption is used to ensure the fairness of queue scheduling among multiple tenants and the priority of applications. For clusters that have high scheduling requirements, you can enable preemption. YARN V2.8.0 or later supports the preemption feature. The following table describes the related configurations.

To enable Capacity Scheduler, configure the configuration items that are described in the following table.

The resource limits for a single task or container are determined by the configuration items that are described in the following table.