Best Practices for Scheduling with Node Labels - E-MapReduce

Background information

In YARN, queues and Node Labels are two key strategies for resource management and scheduling, each serving a distinct purpose.

A queue is a logical method for partitioning resources, allowing you to allocate a cluster's computing resources to different user groups or applications based on predefined ratios. This ensures that different teams or workloads receive a fair share of resources, even in a shared cluster environment.
Node Labels allow you to tag specific nodes in a cluster and make more granular resource allocation decisions based on these tags. This is especially important for applications that require specific hardware features, such as GPUs or high-memory nodes.

You can combine queues and Node Labels for more flexible resource management and job scheduling.

Common partitioning strategies include categorizing by workload type (such as data synchronization, batch, or real-time jobs) or by instance type (such as CPU-optimized, memory-optimized, or GPU-accelerated nodes).

Prerequisites

You have created a cluster with the YARN service enabled. For more information, see Create a cluster.
The YARN scheduler for the cluster must be the Capacity Scheduler. Currently, only the Capacity Scheduler supports partition-based scheduling with Node Labels.

Example

Scenario

An organization divides its cluster queues into a data warehouse queue (warehouse) and a data analysis queue (analysis) based on team resource needs. It also partitions cluster resources into two partitions: batch and streaming. The data warehouse team will use 60% of the batch resources and 70% of the streaming resources. The data analysis team will use 30% of the batch resources and 20% of the streaming resources. For unlabeled resources, the warehouse queue is allocated 60% and the analysis queue is allocated 30%, with 10% reserved for other needs. The following diagram illustrates this partitioning.

This example shows how to combine queues and labels to physically isolate offline jobs from other workloads. The following steps use an EMR-5.16.0 Data Lake cluster. The procedure is similar for other cluster types.

Step 1: Edit resource queues

Log on to the EMR on ECS console. In the Actions column of your target cluster, click Services. Then, click YARN.
On the YARN service page, click the Edit Resource Queue tab, and then click Manage Queues.
On the Manage Child Queues page, add the warehouse queue and the analysis queue.
- The warehouse queue's Queue Capacity is 60%, and the analysis queue's Queue Capacity is 30%. The remaining 10% is allocated to the default queue. The Status for all queues is set to Enabled.
- The page now appears as follows.
  
  The Edit Resource Queue page shows that the current scheduler is capacity-scheduler. Under the root queue, there are three sub-queues: analysis (Queue Capacity 30%, Max. Capacity 100%), warehouse (Queue Capacity 60%, Max. Capacity 100%), and default (Queue Capacity 10%, Max. Capacity 100%). The status of all three queues is Running. A Not Effective Yet label appears next to the YARN service at the top of the page, indicating that the configuration changes have not been deployed.
Click Not Effective Yet. In the Configurations to Take Effect dialog box, click refresh_queues to apply the queue configuration.

In the Custom Effective Configuration table, find the row corresponding to the ResourceManager component and the capacity-scheduler.xml configuration file.

Step 2: Create node partitions

Click the Partitions tab, and then click Add Partition to add the streaming partition.

In the Add Partition dialog box, specify the Partition name, select a Partition Type, and select the Associated Node Groups.

Parameter	Description
Partition Name	In this example, the partition name is `streaming`.
Partition Type	In this example, select `exclusive`. This ensures that only containers requesting a matching partition are scheduled on these nodes.
Associated Node Groups	In this example, the `streaming` partition is associated with `emr-task-3`. This node group consists of memory-optimized instances suitable for real-time jobs that store large amounts of intermediate data with low latency. Assign node groups based on your business characteristics and node instance types.

Repeat the steps to add the batch partition. The final result is as follows.

After you create the partitions, the Partitions page lists three partitions: DEFAULT_PARTITION (Accessible Queues: all, Node Group: emr-core, Partition Type: Exclusive), streaming (Accessible Queues: root and root.default, Node Group: emr-task-3, Partition Type: Exclusive), and batch (Accessible Queues: root and root.default, Node Groups: emr-task-1 and emr-task-2, Partition Type: Exclusive). An orange Not Effective Yet status indicator appears in the top navigation bar. You need to click Deploy and Take Effect to apply the configuration.
After you add the partitions, click Not Effective Yet. In the Configurations to Take Effect dialog box, click refresh_labels to complete the node partition configuration.

In the Custom Effective Configuration section, the configuration file for the ResourceManager component is node-labels.xml.
After the configuration takes effect, you can view the Total Resources for each partition.

For example, the total resources for the streaming partition are displayed as <memory:4096, vCores:8>, and the total resources for the batch partition are displayed as <memory:8192, vCores:16>.

Step 3: Enable partition-queue association management

On the Edit Resource Queue page, select the batch and streaming partitions in sequence, and then click Enable Partition-queue Association Management.
In the Enable Partition-queue Association Management dialog box, configure the Queue Capacity for each queue within the partitions.
- For the batch partition, set the analysis Queue Capacity to 30%, the warehouse Queue Capacity to 60%, and the default Queue Capacity to 10%.
- For the streaming partition, set the analysis Queue Capacity to 20%, the warehouse Queue Capacity to 70%, and the default Queue Capacity to 10%.
Note
In this example, all queues have permission to run in all partitions. In a real-world configuration, you can set the capacity of a specific queue in a partition to 0 to ensure that only jobs from designated queues run in the partition.
After configuring the partition-queue association, click Not Effective Yet. In the Configurations to Take Effect dialog box, click refresh_queues to apply the changes.
On the YARN web UI, verify that the node partitions are active.
- Node Labels page
  
  The table on this page includes the columns Label Name, Label Type, Num Of Active NMs, and Total Resources. It displays three records: <DEFAULT_PARTITION> (Exclusive Partition, 2 active NMs, resources: memory:8192, vCores:16), batch (Exclusive Partition, 2 active NMs, resources: memory:8192, vCores:16), and streaming (Exclusive Partition, 1 active NM, resources: memory:4096, vCores:8).
- Scheduler page
  
  In the Scheduler page's Application Queues section, three partitions are configured: DEFAULT_PARTITION (<memory:8192, vCores:16>), Partition: batch (<memory:8192, vCores:16>), and Partition: streaming (<memory:4096, vCores:8>). Both the batch and streaming node label partitions contain the root queue with the sub-queues default, warehouse, and analysis. The Scheduler type is Capacity Scheduler, with a Minimum Allocation of <memory:32, vCores:1> and a Maximum Allocation of <memory:4096, vCores:8>.

Step 4: Submit a job to a partition

After you configure the cluster queues and labels, you can submit jobs to run on specific queues and nodes.

Log on to the master node of the cluster and navigate to the Spark installation directory, for example, /opt/apps/SPARK3/spark-3.3.1-hadoop3.2-1.1.1.
```
cd /opt/apps/SPARK3/spark-3.3.1-hadoop3.2-1.1.1
```

In a production environment, you must upload your job's JAR file to submit a Spark job. This example uses spark-submit to submit a job.

./bin/spark-submit --class org.apache.spark.examples.SparkLR --master yarn --deploy-mode cluster --driver-memory 1g --executor-memory 2g --conf spark.yarn.am.nodeLabelExpression=batch --conf spark.yarn.executor.nodeLabelExpression=batch --queue=warehouse examples/jars/spark-examples_2.12-3.3.1.jar

Parameter	Description
class	The main class of the application: `org.apache.spark.examples.SparkLR`.
master	The execution environment for the job: `yarn`.
deploy-mode	The location where the driver is launched: `cluster`.
driver-memory	The amount of memory for the driver: `1g`.
executor-memory	The amount of memory per executor: `2g`.
spark.yarn.am.nodeLabelExpression	The label expression for nodes where the ApplicationMaster can run: `batch`.
spark.yarn.executor.nodeLabelExpression	The label expression for nodes where executors can run: `batch`.
queue	The specific YARN queue to which the job is submitted: `warehouse`.

Note

If you specify only a queue and do not specify a partition when submitting a job, the job is submitted to the queue's default partition. You can change a queue's default partition on the Edit Resource Queue page by clicking Edit for that queue.

On the YARN cluster's web UI, you can verify that the SparkLR job submitted to the warehouse queue is running on the batch partition.

YARN queues and node partitions together provide logical and physical resource isolation, preventing interference between jobs and ensuring the stability of critical applications.

FAQ

When I submit a Spark job by using spark-submit, why does the job not run on the specified partition?
- Check whether the spark.yarn.executor.nodeLabelExpression parameter is included in your submission command.
- Confirm that your settings on the Partitions and Edit Resource Queue pages are active.
When I submit a Spark job by using spark-submit, why does the job status remain ACCEPTED?
1. Confirm that the queue to which the job was submitted has available resources. If not, wait for other jobs to release resources before resubmitting the job.
2. If the queue has available resources, click the job ID on the YARN web UI to view the job details and check the Diagnostics description. You might see a message: "Queue's AM resource limit exceeded". This indicates that although the queue has resources, the Application Master (AM) resources are insufficient. A queue's resources are allocated to both the AM and the executors. AM resources are used to launch the main application, while executor resources are used for task processing. Therefore, you need to adjust the queue's resource allocation ratio and increase the proportion of AM resources to ensure Spark jobs can be submitted successfully.
```
Diagnostics:   [Wed Dec 25 16:24:10 +0800 2024] Application is added to the scheduler and is not yet activated. Queue's AM resource limit exceeded. Details : AM
               Partition = &lt;DEFAULT_PARTITION&gt;; AM Resource Request = &lt;memory:2048, vCores:1&gt;; Queue Resource Limit for AM = &lt;memory:2048, vCores:4&gt;; User
               AM Resource Limit of the queue = &lt;memory:224, vCores:1&gt;; Queue AM Resource Usage = &lt;memory:896, vCores:1&gt;;
```
3. You can adjust the AM resource ratio on the YARN service configuration page. Find the capacity_scheduler.xml file and modify the value of the yarn.scheduler.capacity.maximum-am-resource-percent configuration item. For example, change it from 0.25 to 0.5.
Note
Adjust the default AM resource percentage defined by yarn.scheduler.capacity.maximum-am-resource-percent based on your specific use case. For example, if your cluster frequently processes a large number of jobs with small data volumes, you can increase the AM resource percentage.

E-MapReduce:Partition nodes using Node Labels