Boost Your Cluster’s IQ: YARN Intelligent Scheduling Explained#

What is YARN?#

YARN stands for Yet Another Resource Negotiator. It was introduced in Hadoop 2.x to separate the responsibilities of resource management (like CPU, memory, and I/O tracking) from job scheduling and monitoring. Before YARN, Hadoop used a monolithic system called MapReduce v1, where the JobTracker was responsible for both scheduling tasks and allocating resources. YARN addressed the scalability concerns of that approach by introducing a global ResourceManager and per-application ApplicationMasters.

Here’s a simple breakdown of the Hadoop ecosystem with YARN:

• Storage layer (HDFS)
• YARN for cluster resource management
• Multiple execution engines (MapReduce v2, Spark, Tez, etc.)

This separation of responsibilities allows Hadoop-based infrastructures to run various analytical engines and ensures more efficiency.

Key Components of YARN#

Before diving into scheduling, let’s clarify the essential components of YARN:

1. ResourceManager (RM):
   The central authority that manages resources among all applications running in the cluster. It has two main components:

   • Scheduler: Responsible for allocating resources (containers) to running applications. It does not track or monitor the status of application processing.
   • ApplicationManager (ApplicationsManager): Accepts job submissions, negotiates the first container for an application’s ApplicationMaster, and restarts the ApplicationMaster in case it fails.

2. NodeManager (NM):
   Runs on each data node within the cluster. It is responsible for launching and monitoring application containers, reporting resource usage back to the ResourceManager, and managing local disk and log files.

3. ApplicationMaster (AM):
   A per-application process that negotiates resources from the ResourceManager and works directly with the NodeManager(s) to execute tasks and coordinate the overall job lifecycle.

4. Containers:
   Logical bundles of resources (CPU, memory, disk, etc.) allocated by the ResourceManager to tasks. A container is a running process (or a set of processes) on a node with specified resources.

At a high level, the flow of YARN scheduling is as follows:

1. An application is submitted to the ResourceManager.
2. The ResourceManager looks at scheduling constraints and decides where the ApplicationMaster will be launched.
3. The ApplicationMaster then requests more containers for individual tasks of the application.
4. The ResourceManager allocates containers based on in-flight demands and scheduling policies.

FIFO Scheduler#

FIFO (First In, First Out) is the simplest of the three schedulers.

• All applications go into a single queue and get scheduled in time order.
• The entire cluster capacity is used to run the jobs in the order they arrived.
• Once a job is finished, resources free up for the subsequent job in the queue.

Usually, FIFO is not optimal in modern, busy clusters that serve multiple teams or workloads. However, it is straightforward to set up and might suffice when the cluster usage pattern is simple.

Capacity Scheduler#

The Capacity Scheduler was primarily designed for multitenant environments, where multiple teams or departments within an organization share a common cluster. Its key concepts include:

1. Queues: Hierarchical structures that define how resources are shared among different organizational groups. Each queue can further be sub-divided into sub-queues (parent-child relationship), allowing for more granular control.
2. Capacity Guarantees: Each queue enforces a minimum capacity (in percentage of cluster resources) that it can use. If a queue is underutilizing its capacity, other queues can borrow resources up to a certain limit.
3. ACLs (Access Control Lists): Provides a security model, restricting which users or groups can submit applications to specific queues.
4. Priority: Supports priority for applications inside a queue, where higher priority apps get containers first.

Fair Scheduler#

The Fair Scheduler tries to balance resource allocation so that all running applications get, on average, an equal share of cluster resources over time. Its main advantages include:

1. Fair Allocation: Each application and user can be assigned a fair share of cluster resources, preventing any single job from monopolizing resources.
2. Preemption: Automatically reclaims resources from lower-priority tasks and redistributes them to higher-priority tasks.
3. Pools: Instead of queues, the Fair Scheduler uses “pools.” Each pool can have a minimum and maximum capacity. Individual applications (or groups of users) can belong to specific pools for better resource governance.

Although the Fair Scheduler is conceptually simpler than the Capacity Scheduler’s hierarchical queue design, both aim to improve cluster utilization and fairness compared to a plain FIFO approach.

Resource Allocation Strategies#

In YARN, resource allocation is typically measured in terms of CPU cores and memory. However, more recent versions can encompass additional resources such as disks or GPUs. Intelligent scheduling might:

• Evaluate job characteristics (e.g., short-lived vs. long-lived)
• Dynamically adjust queue capacities based on real-time usage
• Employ container-level resource negotiations (like cgroups for CPU throttling)

The overarching goal is to ensure cluster resources are used fully (high utilization) and that critical applications can still meet Service-Level Agreements (SLAs) or latency requirements.

Priority-Based Scheduling#

Priority-based scheduling ensures that certain applications or tasks get precedence when requesting resources. For example:

• Marketing analytics or critical dashboards might need immediate resource allocation for timely results.
• Less critical batch tasks or testing workloads might run at lower priority, filling up idle resources as needed.

In the Capacity Scheduler, you can configure priorities at the queue level and within each queue. In the Fair Scheduler, you can set job priorities or rely on preemption to reassign resources temporarily.

Dynamic Resource Sharing#

Dynamic resource sharing refers to the cluster’s ability to reclaim unused resources and allocate them to other waiting jobs. Both the Capacity and Fair Schedulers have mechanisms to redistribute resources if certain queues or pools are underutilized.

In more advanced environments, dynamic resource sharing can incorporate load-based predictions or advanced heuristics. For instance, a large Spark job that’s mostly idle between stages might temporarily reduce resource ownership, freeing up containers for other workloads. When the Spark stage is ready to run again, preemption or priority can shift resources back where needed.

Step-by-Step Configuration#

Here is a general process for configuring YARN for more advanced scheduling:

1. Choose the Scheduler: Decide which scheduler best fits your use case (Capacity or Fair).
2. Edit Yarn Configuration: Update yarn-site.xml with the appropriate properties.
3. Configure Scheduler-Specific Files:
   • Capacity Scheduler: capacity-scheduler.xml
   • Fair Scheduler: fair-scheduler.xml
4. Enable Priority/Preemption: Configure the scheduler to allow priority-based scheduling and resource preemption.
5. Activate HA (if necessary): For production clusters, enable High Availability (HA) for the ResourceManager.
6. Monitor and Adjust: Use YARN’s web UI, logs, and metrics to verify that scheduling is working as intended. Refine capacity or fairness settings as needed.

Configuration Examples#

Below are simplified examples showing how you might configure either the Capacity Scheduler or the Fair Scheduler for intelligent scheduling.

1
<property>
2
  <name>yarn.resourcemanager.scheduler.class</name>
3
  <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler</value>
4
</property>

1
<property>
2
  <name>yarn.scheduler.capacity.root.queues</name>
3
  <value>A,B</value>
4
</property>
5

6
<property>
7
  <name>yarn.scheduler.capacity.root.A.capacity</name>
8
  <value>60</value>
9
</property>
10

11
<property>
12
  <name>yarn.scheduler.capacity.root.B.capacity</name>
13
  <value>40</value>
14
</property>
15

16
<!-- Minimum & maximum capacities for Queue A -->
17
<property>
18
  <name>yarn.scheduler.capacity.root.A.minimum-user-limit-percent</name>
19
  <value>50</value>
20
</property>
21

22
<property>
23
  <name>yarn.scheduler.capacity.root.A.maximum-capacity</name>
24
  <value>90</value>
25
</property>
26

27
<!-- Enable Priority Preemption -->
28
<property>
29
  <name>yarn.scheduler.capacity.root.A.enable-user-limit</name>
30
  <value>true</value>
31
</property>
32

33
<property>
34
  <name>yarn.scheduler.capacity.root.A.user-limit-factor</name>
35
  <value>2</value>
36
</property>

1
<property>
2
  <name>yarn.resourcemanager.scheduler.class</name>
3
  <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler</value>
4
</property>

1
<allocations>
2
  <pool name="production">
3
    <minResources>1024 mb,1 vcores</minResources>
4
    <maxResources>8192 mb,8 vcores</maxResources>
5
    <weight>2.0</weight>
6
    <priority>1</priority>
7
  </pool>
8
  <pool name="development">
9
    <minResources>512 mb,1 vcores</minResources>
10
    <maxResources>4096 mb,4 vcores</maxResources>
11
    <weight>1.0</weight>
12
    <priority>2</priority>
13
  </pool>
14
</allocations>

Real-Time Streaming Applications#

Real-time streaming frameworks like Spark Streaming, Flink, or Storm constantly require low-latency processing. To ensure these jobs always have enough resources:

• Configure high-priority pools or queues dedicated to streaming tasks.
• Enable preemption so that critical streaming tasks can quickly reclaim resources.
• Use auto-scaling if possible, so your cluster can provision extra nodes under heavy streaming loads.

Ensuring streaming tasks don’t starve out other workloads is also key. Configure maximum capacity limits so that streaming workloads cannot saturate the entire cluster if they spike unexpectedly.

Batch Processing Workloads#

Batch jobs often have flexible deadlines, but can be large, resource-intensive, and long-running. Recommendations:

• Separate batch workloads into their own queue or pool with a guaranteed minimum capacity.
• Enable integration with workflow managers like Oozie or Airflow that can dynamically submit jobs to the cluster.
• Tune container sizes (CPU, memory) to balance I/O overhead with parallelism.

Batch workloads can often fill in any gaps left by interactive or streaming tasks, so consider configuring them at medium or lower priorities, allowing more critical workloads to preempt.

Mixed Workloads and Multitenancy#

Many organizations have clusters shared by different teams. Some best practices for mixed workloads:

• Establish queues or pools by department, function, or job type (e.g., data science vs. ETL).
• Use a combination of minimum, maximum capacity settings, and user-limit factors to ensure fairness.
• Monitor cluster usage data over time. Adjust queue capacities or fairness weights to match usage patterns.
• Encourage each team to use resource-based job submission guidelines (proper memory/cpu requests).

Fine-Tuning Scheduler Parameters#

Fine-tuning advanced scheduling parameters can make the difference between a well-run cluster and a chaotic environment. Examples include:

• Preemption Timeout: How long a high-priority task waits before forcibly preempting lower-priority containers.
• User-Limit Factor: How many resources a single user can consume within a queue.
• Maximum-AM-Resource-Percent: Limits the fraction of the queue (or pool) that ApplicationMasters can occupy, preventing a flood of small tasks from consuming too many containers.

Each parameter has to be tested in your own environment. Lab or staging clusters can help verify that your modifications produce the desired outcome without impacting production workloads.

Multi-Tenant Environment Optimization#

For multi-tenant clusters with hundreds of users and varied skill levels, you’ll need robust policies and possibly custom scripts:

1. Quota Enforcement: Setting up advanced access controls ensures each department or project only uses its allocated share.
2. Dynamic Quota Borrowing: Allowing certain queues to exceed their guaranteed capacity if others are underutilized.
3. User Education: Train users on how to declare resource requests responsibly (e.g., using Spark’s --executor-memory correctly) to avoid resource misallocations.

Enable clear logging, monitoring, and alerting. Tools like Grafana, Kibana, or Cloudera Manager can help visualize resource usage across queues, identifying bottlenecks or misuse.

YARN Federation#

YARN Federation is a high-level feature that allows multiple YARN clusters to cooperate, forming a single “federated” cluster. Schedulers can then allocate resources from multiple sub-clusters:

• Useful for geo-distributed data centers or extremely large clusters.
• Eliminates single-point bottlenecks by horizontally scaling YARN’s ResourceManagers.
• Federated scheduling policies can be complex but provide near-infinite scalability.

With Federation, you can have one application request containers from multiple sub-clusters. Intelligent scheduling is extended further, requiring an understanding of network latencies, data locality, and global resource availability.