Turbocharge Your Data Pipeline: YARN Optimization Techniques#

ResourceManager#

The ResourceManager (RM) is the master service in YARN, responsible for tracking available cluster resources and allocating them to different applications. It has two main modules:

1. Scheduler: Decides how to allocate resources to various running applications.
2. ApplicationManager: Manages job submissions, negotiations, and application states.

NodeManager#

The NodeManager (NM) runs on each node in the cluster. It manages containers, checks resource usage, and reports node health. By doing so, it ensures that each worker node is functioning correctly and that container usage is within set limits (CPU, memory, etc.).

ApplicationMaster#

Each application submits an ApplicationMaster (AM) to the cluster. The AM negotiates resource requirements with the ResourceManager and interacts with NodeManagers to launch and monitor containers. Each application has its specific AM, allowing for custom optimization and management of that application’s tasks.

Container#

A container is a collection of physical resources (memory, CPU cores, etc.) assigned to a particular application or job. Containers isolate workloads from one another, and YARN draws upon these containers to provide fine-grained resource management.

HDFS vs. YARN#

While HDFS is the storage layer in Hadoop for large-scale data, YARN is focused on resource management and execution. They complement each other in a typical Hadoop environment but serve different primary objectives.

Prerequisites#

• Java installed on all cluster nodes
• SSH password-less login set up
• Hadoop installed and configured (including HDFS)

Basic Configuration Steps#

1. Edit yarn-site.xml: Configure basic properties like yarn.resourcemanager.hostname, yarn.nodemanager.resource.memory-mb, and the path for logs.

2. Edit mapred-site.xml: Specify the framework to use YARN by setting mapreduce.framework.name to yarn.

3. Enable NodeManagers: Start yarn daemons on each worker node with the command:

   1
   $ start-yarn.sh
   

   Ensure each NodeManager successfully registers with the ResourceManager.

4. Verify the Setup: Use the resource manager’s web UI (default at http://:8088) to confirm active nodes, resource allocation, and running applications.

Below is a sample minimal yarn-site.xml configuration snippet to demonstrate essential properties:

1
<configuration>
2
    <property>
3
        <name>yarn.resourcemanager.hostname</name>
4
        <value>resourcemanager.example.com</value>
5
    </property>
6
    <property>
7
        <name>yarn.nodemanager.resource.memory-mb</name>
8
        <value>4096</value>
9
    </property>
10
    <property>
11
        <name>yarn.nodemanager.resource.cpu-vcores</name>
12
        <value>2</value>
13
    </property>
14
    <property>
15
        <name>yarn.scheduler.minimum-allocation-mb</name>
16
        <value>512</value>
17
    </property>
18
    <property>
19
        <name>yarn.scheduler.maximum-allocation-mb</name>
20
        <value>4096</value>
21
    </property>
22
</configuration>

Memory and CPU Cores#

• Memory: YARN primarily focuses on memory-based container allocations. Jobs requiring more memory will have larger containers, which can affect the ability to run multiple containers in parallel.
• CPU: Specified as virtual cores (vcores). Allocating more vcores per container can speed up computation, but it also increases the chance of CPU contention.

Containers and Queuing#

• Container Sizing: Choosing the right container size ensures you don’t underutilize or overload your nodes. A common approach is to match the container memory size to a fraction of the total node memory, leaving room for system processes.
• Queue Configuration: YARN uses queues to group resources for different user groups or applications. Well-planned queues reduce conflicts and maximize resource utilization.

Best Practices for Resource Allocation#

1. Measure, Don’t Guess: Use cluster-wide metrics like CPU utilization, memory usage, and job completion times to guide container sizes.
2. Separate Batch and Interactive Work: If you have both Spark streaming (interactive) workloads and batch MapReduce jobs, place them in separate queues. This avoids having interactive queries starve while large batch jobs hog resources.
3. Use Resource-Based Quotas: Set per-queue resource quotas (e.g., memory, CPU) to ensure no single user or app monopolizes the cluster.

FIFO (First-In-First-Out)#

• How it works: Jobs are processed in the order they arrive.
• Pros: Easy to understand and configure.
• Cons: Not suitable for multi-tenant situations where users or teams need fair resource sharing.

Capacity Scheduler#

• How it works: Divides cluster resources among different queues; each queue has a capacity defined in percentages. Any idle capacity can be borrowed by other queues.
• Pros: Ideal for multi-tenant environments with a requirement to guarantee minimum capacity for each team.
• Cons: More complex to configure than FIFO.

Fair Scheduler#

• How it works: Distributes resources among all running jobs in a fair manner. Over time, each job gets an equal share of resources.
• Pros: Ensures all jobs make progress at roughly the same rate, preventing any single job from taking over the cluster.
• Cons: Advanced configurations can get complicated, especially with hierarchical fair schedules.

Yarn-Site.xml Optimization#

Below are a few critical properties to focus on:

1. yarn.scheduler.minimum-allocation-mb
2. yarn.scheduler.maximum-allocation-mb
3. yarn.nodemanager.resource.memory-mb
4. yarn.nodemanager.resource.cpu-vcores

Set these properties to reflect the actual capacity of your worker nodes. If you have nodes with 64GB of RAM and 16 cores, you might set yarn.nodemanager.resource.memory-mb to slightly less than 64GB (e.g., 60GB) to leave overhead for the operating system and Hadoop daemons.

Capacity-Scheduler.xml Optimization#

When using the Capacity Scheduler, you can create multiple queues. For instance, a queue for the data science team, one for the ETL team, and another for ad-hoc queries. Each queue can have a percentage of total cluster resources.

1
<configuration>
2
    <property>
3
        <name>yarn.scheduler.capacity.root.queues</name>
4
        <value>etl,analytics,adhoc</value>
5
    </property>
6

7
    <!-- ETL Queue Configuration -->
8
    <property>
9
        <name>yarn.scheduler.capacity.root.etl.capacity</name>
10
        <value>50</value>
11
    </property>
12

13
    <!-- Analytics Queue Configuration -->
14
    <property>
15
        <name>yarn.scheduler.capacity.root.analytics.capacity</name>
16
        <value>30</value>
17
    </property>
18

19
    <!-- Adhoc Queue Configuration -->
20
    <property>
21
        <name>yarn.scheduler.capacity.root.adhoc.capacity</name>
22
        <value>20</value>
23
    </property>
24
</configuration>

Fair-Scheduler.xml Optimization#

If you use the Fair Scheduler, a sample configuration might look like this:

1
<allocations>
2
    <pool name="etl">
3
        <minResources>10g,10vcores</minResources>
4
        <maxResources>50g,50vcores</maxResources>
5
    </pool>
6
    <pool name="analytics">
7
        <minResources>5g,5vcores</minResources>
8
        <weight>1.5</weight>
9
        <minSharePreemptionTimeout>60</minSharePreemptionTimeout>
10
    </pool>
11
</allocations>

Hierarchical Queue Structures#

In both Capacity and Fair schedulers, you can create nested queues to match organizational hierarchies. This offers a way to distribute resources among departments and teams in a manner that aligns with internal structures:

• root
  • etl
    • teamA
    • teamB
  • analytics
    • teamC
    • teamD
  • adhoc

This arrangement ensures that each sub-queue has guaranteed minimum resources but can still share idle resources from the parent queue.

Container Reuse#

When running iterative or incremental applications (like Spark jobs), enabling container reuse may help reduce overhead. This feature keeps containers alive providing an application can reuse them for subsequent tasks without re-initializing.

To enable container reuse for MapReduce, add:

1
<property>
2
    <name>mapreduce.job.reduce.slowstart.completedmaps</name>
3
    <value>0.7</value>
4
</property>
5
<property>
6
    <name>mapreduce.job.jvm.numtasks</name>
7
    <value>-1</value>
8
</property>

This ensures reduce tasks only start when a certain fraction of map tasks are complete (reducing idle time) and reuses JVM containers across tasks.

Speculative Execution#

Speculative execution helps when tasks are running slowly on a particular node due to hardware issues or data skew. YARN can launch duplicate tasks on different nodes to complete earlier.

1
<property>
2
    <name>mapreduce.map.speculative</name>
3
    <value>true</value>
4
</property>
5
<property>
6
    <name>mapreduce.reduce.speculative</name>
7
    <value>true</value>
8
</property>

Be careful with this setting for jobs that are sensitive to out-of-order updates, such as certain incremental data pipelines (where duplication might cause consistency issues).

ApplicationMaster Configuration#

For large clusters or resource-intensive jobs, you may need to increase the memory allocated to the ApplicationMaster. This is usually done by editing mapreduce.am.memory.mb in mapred-site.xml for MapReduce, or by Spark’s configuration for Spark on YARN.

Common Monitoring Tools#

1. ResourceManager Web UI (default port 8088) - Provides quick health checks and resource usage stats.
2. NodeManager Web UI (default port 8042) - Shows node-level container usage and logs.
3. Hadoop Metrics2 - Exposes JMX counters for deeper monitoring.
4. Grafana + Prometheus or Ambari - Popular third-party solutions for real-time cluster-wide dashboards.

Memory Overhead and Container Settings#

Each container has allocated memory plus overhead for the JVM and system processes. Tune the following property for advanced overhead control:

1
<property>
2
    <name>yarn.nodemanager.vmem-check-enabled</name>
3
    <value>false</value>
4
</property>

When enabled (true), containers that exceed their virtual memory limits will be killed. Disabling it might prevent false positives but make sure to keep an eye on available system memory.

YARN ResourceManager HA#

To avoid a single point of failure (SPOF), enable ResourceManager High Availability by running backup ResourceManagers in a Standby mode, which can take over automatically if the Active ResourceManager fails.

Key steps include:

1. Configuring ZooKeeper for leader election.
2. Syncing ResourceManager state using the ResourceManager state store (FileSystem-based or ZooKeeper-based).
3. Ensuring the failover controller is properly set up.

Multi-Tenancy Considerations#

In multi-tenant environments:

• Allocate separate queues for different teams, each with minimum resource capacity.
• Implement access control lists (ACLs) to restrict which users can submit jobs to specific queues.
• Integrate with enterprise authentication systems (e.g., Kerberos, LDAP).

Kerberos Integration#

Kerberos is a key security mechanism for authenticating users and services. In YARN:

1. Each YARN component (ResourceManager, NodeManager) must have a valid Kerberos principal and keytab.
2. End-users submit jobs using authenticated credentials, ensuring only authorized persons can access cluster resources.

YARN ACLs#

YARN supports Access Control Lists (ACLs) for queues and job submissions. For instance:

1
<property>
2
    <name>yarn.scheduler.capacity.root.analytics.acl_submit_applications</name>
3
    <value>alice,bob,group analytics-team</value>
4
</property>

This ensures only specified users/groups can submit applications to the analytics queue.

Implementation#

• Use the Capacity Scheduler with three top-level queues: root.etl at 40%, root.analytics at 40%, and root.adhoc at 20%.
• Configure yarn.nodemanager.resource.memory-mb to 60GB on each of the 64GB worker nodes.
• Enable Speculative Execution to handle variable run times.
• Use Yarn ACLs to restrict queue usage to authorized staff.

Outcome#

• ETL jobs no longer starve analytics workflows, and vice versa.
• The Data Science group can leverage additional capacity if the ETL jobs finish early.
• Overall cluster utilization improved from 60% to 85% since idle resources can be “borrowed” across queues.

Symptom 1: Containers Keep Getting Killed#

Possible causes:

• Insufficient container memory.
• The NodeManager’s vmem ratio setting is too strict.

Solution:

• Increase container memory configurations, or disable vmem checks.
• Review logs in the NodeManager Web UI to identify memory limit breaches.

Symptom 2: Slow Job Completion Times#

Likely reasons:

• Over-allocation of CPU.
• Network or disk bottlenecks.
• Highly skewed data leading to large tasks.

Solution:

• Fine-tune container CPU allocations.
• Evaluate the I/O throughput of the nodes, and possibly add more disks or use SSD for critical tasks.
• Enable data skew handling features in your application logic (e.g., Spark’s adaptive execution).

Symptom 3: Queues Starving Other Queues#

Caused by:

• Improper capacity configuration, allowing one queue to over-consume resources.

Fix:

• Restrict maximum capacity of each queue.
• Implement preemption in the Capacity Scheduler, which allows YARN to reclaim resources from low-priority jobs.