Building a Resilient Data Ecosystem with YARN Resource Management#

Why YARN?#

YARN, or Yet Another Resource Negotiator, is designed to separate resource management from the processing engine. This modular approach empowers Hadoop so that diverse data applications (Spark, MapReduce, Flink, or custom apps) can run concurrently on the same cluster. By scheduling and orchestrating resources intelligently, YARN provides a dynamic and robust environment capable of handling mixed workloads with utmost efficiency.

Key benefits of adopting YARN:

• Multi-tenancy: Multiple applications can coexist in the same cluster, isolating them from each other while sharing underlying resources.
• Scalability: Seamlessly scale your cluster up or down to accommodate varying workloads.
• Fault tolerance: YARN detects and reacts to component failures, providing a resilient environment.
• Efficient resource utilization: Allocate CPU, memory, and other resources dynamically across applications.

1. ResourceManager (RM)#

• Manages the entire cluster’s resources.
• Receives requests for resources from the ApplicationMaster.
• Makes decisions based on scheduling policies.

2. NodeManager (NM)#

• Runs on each node in the cluster.
• Executes tasks on those nodes within allocated containers.
• Monitors container resource usage and health.
• Sends periodic heartbeat messages to the ResourceManager.

3. ApplicationMaster (AM)#

• A dedicated process created for each submitted application (e.g., Spark job, MapReduce job).
• Negotiates container allocation with the ResourceManager.
• Works with NodeManagers to launch containers.
• Tracks task execution, deals with task failures, and frees resources upon completion.

4. Containers#

• Encapsulate resources needed by a task.
• Unit of execution in a YARN cluster.
• Can be configured to request certain memory (MB), CPU (vCores), and other resources.

A simplified overview of data flow in YARN:

1. Client submits an application to the cluster.
2. YARN creates an ApplicationMaster for that application.
3. The ApplicationMaster negotiates resources and requests containers.
4. Tasks run inside containers on various nodes, orchestrated by the NodeManagers and the ResourceManager.
5. Once completed, resources are freed, and the application finishes.

1. Prerequisites#

• A set of machines (physical or virtual) with a compatible OS (e.g., Linux).
• Java installed on each node.
• A stable network setup for the cluster.

2. Hadoop Installation#

1. Download the binary distribution of Apache Hadoop (version 3.x or newer).
2. Extract the distribution to a directory (e.g., /usr/local/hadoop).
3. Configure environment variables for Hadoop and Java:

1
export HADOOP_HOME=/usr/local/hadoop
2
export JAVA_HOME=/usr/lib/jvm/java-8-openjdk
3
export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin

4. Ensure SSH access is configured for passwordless login between the nodes.

3. Configuring yarn-site.xml#

In your Hadoop configuration directory (e.g., $HADOOP_HOME/etc/hadoop), edit the yarn-site.xml:

1
<configuration>
2
    <property>
3
        <name>yarn.resourcemanager.hostname</name>
4
        <value>resourcemanager.example.com</value>
5
    </property>
6

7
    <property>
8
        <name>yarn.nodemanager.resource.memory-mb</name>
9
        <value>8192</value>
10
    </property>
11

12
    <property>
13
        <name>yarn.nodemanager.resource.cpu-vcores</name>
14
        <value>4</value>
15
    </property>
16

17
    <property>
18
        <name>yarn.scheduler.maximum-allocation-mb</name>
19
        <value>8192</value>
20
    </property>
21

22
    <property>
23
        <name>yarn.scheduler.minimum-allocation-mb</name>
24
        <value>1024</value>
25
    </property>
26
</configuration>

• yarn.resourcemanager.hostname: The hostname of your primary node where ResourceManager runs.
• yarn.nodemanager.resource.memory-mb: Total memory allocated for NodeManager’s containers.
• yarn.nodemanager.resource.cpu-vcores: CPU cores available for containers.
• yarn.scheduler.maximum-allocation-mb: Maximum memory for a single container.
• yarn.scheduler.minimum-allocation-mb: Minimum memory for a single container.

4. Starting and Verifying the Cluster#

1. Start the HDFS daemons (NameNode, DataNode):

   1
   start-dfs.sh
   

2. Start the YARN daemons (ResourceManager, NodeManager):

   1
   start-yarn.sh
   

3. Verify the ResourceManager UI by opening http://resourcemanager.example.com:8088 in a web browser. You should see an overview of the cluster and available resources.

Example: Capacity Scheduling in Action#

Suppose your organization has a data science team that runs heavy Spark jobs, and a marketing team that runs smaller ad-hoc queries. You create two queues:

• DataScience: Guaranteed 70% of the cluster
• Marketing: Guaranteed 30% of the cluster

If the marketing team is idle, data science workloads can temporarily use up to 100% of the cluster. As soon as marketing jobs start, the scheduler reclaims resources gradually until each queue is back to its configured capacity.

1. YARN ResourceManager UI#

• Easily accessible at http://<resourcemanager-host>:8088.
• Displays cluster metrics: number of applications running, allocated memory, allocated CPU, container statuses.
• Allows you to kill misbehaving or stuck applications and see their progress.

2. NodeManager UI#

• Typically available at http://<nodemanager-host>:8042.
• Shows the containers running on that node, memory usage, logs, and local node metrics.

3. YARN Logs#

• Logs are crucial when debugging.
• Use the yarn logs -applicationId <app_id> command to retrieve logs of completed or running applications.

Sample command:

1
yarn application -list
2
yarn logs -applicationId application_1630928765000_0003

• This retrieves container logs (stdout, stderr, syslog) for the targeted application.

1. ResourceManager High Availability (HA)#

In a production Hadoop setup, you can configure a standby ResourceManager that takes over if the active ResourceManager fails. This approach, called automatic failover, ensures that the cluster remains functional even when the primary RM goes down. Typically, ZooKeeper is used as a coordination mechanism.

Configuration Excerpt:

1
<property>
2
    <name>yarn.resourcemanager.ha.enabled</name>
3
    <value>true</value>
4
</property>
5

6
<property>
7
    <name>yarn.resourcemanager.ha.rm-ids</name>
8
    <value>rm1, rm2</value>
9
</property>

2. NodeManager Recovery: If a NodeManager crashes, it can re-register itself with the ResourceManager upon restart. Running containers on that node might be lost, but the rest of the cluster remains active.

3. Application Attempt and Retry: Each YARN application can have multiple attempts. The ResourceManager automatically negotiates new containers for tasks that fail, within configured retry limits.

1. Hierarchical Queues and Dynamic Resource Pools#

For large organizations, a single flat queue structure can become cumbersome. CapacityScheduler supports a hierarchical model for dividing cluster resources:

• Root Queue
  • Dept1 (sub-queue)
    • Analytics (sub-queue)
    • ETL (sub-queue)
  • Dept2 (sub-queue)

This hierarchy allows you to nest resource controls at multiple levels. Dynamic resource pools in some Hadoop distributions let you create or remove sub-queues on the fly, reacting to real-time workload changes.

2. Preemption#

When using the FairScheduler or capacity scheduling with preemption enabled, the ResourceManager can forcibly terminate containers in low-priority queues to honor resource guarantees for higher-priority queues. While potentially disruptive for running tasks, preemption ensures critical jobs have guaranteed throughput.

3. Resource Reservation#

YARN can “reserve” resources for specific scheduling requests to ensure that long-running or high-priority tasks obtain consistent performance. The ResourceManager reserves containers on specific nodes to avoid fragmentation or constant rebalancing.

4. Custom Resource Types#

By default, YARN handles CPU and memory as the principal resources. However, advanced deployments might require specialized hardware—GPUs, FPGAs, or new ephemeral storage devices. Modern YARN supports the concept of custom resource types so that you can assign GPU fractions or ephemeral storage in your scheduling policies.

1. Docker Container Executor#

In your NodeManager’s configuration (yarn-site.xml), you can specify that containers should be executed under Docker:

1
<property>
2
    <name>yarn.nodemanager.container-executor.class</name>
3
    <value>org.apache.hadoop.yarn.server.nodemanager.LinuxContainerExecutor</value>
4
</property>
5

6
<property>
7
    <name>yarn.nodemanager.linux-container-executor.cgroups.mount</name>
8
    <value>true</value>
9
</property>
10

11
<property>
12
    <name>yarn.nodemanager.runtime.linux.docker.enabled</name>
13
    <value>true</value>
14
</property>

2. Image Management: The NodeManager downloads or references the specified Docker image. The images must be accessible via Docker registry, or built locally on each node if you prefer an offline approach.

3. Resource Isolation: Docker containers provide an additional layer of isolation. YARN’s container resource controls (CPU, memory) integrate with cgroup-based resource management in Docker, preventing applications from over-consuming resources.

4. Example: Running a Spark job in Docker

When requesting resources, your Spark application might specify a Docker image name, e.g.:

1
spark-submit \
2
  --master yarn \
3
  --conf spark.yarn.containerLauncher=org.apache.spark.launcher.docker.DockerContainerLauncher \
4
  --conf spark.yarn.docker.image=organization/spark:latest \
5
  path/to/your/job.jar

This approach ensures a consistent Python or Java environment for your Spark jobs across the entire cluster.

1. Kerberos Authentication#

Hadoop’s default security mechanism is Kerberos. Each service (ResourceManager, NodeManager, HDFS NameNode, etc.) has a principal and keytab. When a user submits a job, they obtain a Kerberos ticket for authentication. This ensures only authorized users can launch or modify applications.

2. Access Control Lists (ACLs)#

Hadoop YARN allows you to define ACLs on queues, controlling who can submit or administer applications. For instance, only certain groups might be allowed to kill jobs in certain queues. In capacity-scheduler.xml, you can configure:

1
<property>
2
    <name>yarn.scheduler.capacity.root.analytics.acl_submit_applications</name>
3
    <value>analytics-team</value>
4
</property>
5

6
<property>
7
    <name>yarn.scheduler.capacity.root.analytics.acl_administer_queue</name>
8
    <value>admin-group</value>
9
</property>

3. Data Encryption#

In addition to Yarn’s built-in security, you should consider encrypting data in-transit and at rest. HDFS offers transparent data encryption zones, and TLS can secure NodeManager and ResourceManager communication.

1. Sizing NodeManager Resources#

• Avoid over-allocating memory to NodeManager containers.
• Reserve a small part of the system’s physical memory for OS overhead and the NodeManager itself.
• Balance the vCores (CPU) and memory to suit your workloads (e.g., memory-bound Spark queries vs. CPU-heavy analytics).

2. Handling Straggler Jobs#

• Use speculative execution for frameworks like MapReduce or Spark to mitigate stragglers.
• Configure the ApplicationMaster to detect tasks that are slow relative to others and launch backup tasks in parallel.

3. Log Aggregation and Archival#

• Configure log aggregation in YARN, so application logs are moved to a central data store (e.g., HDFS) upon job completion.
• Retain logs for a reasonable time to investigate errors post-flight, but also monitor disk usage.

4. Auto-Scaling YARN Clusters#

In cloud environments, you can dynamically scale YARN worker nodes based on metrics like CPU usage or queue wait times. Tools like Kubernetes or data pipeline-specific orchestrators can spin up additional EC2 instances (if using AWS) or VMs in other clouds on-demand, then add them to the Hadoop cluster to handle peak loads.

5. Rolling Upgrades#

Upgrading a production cluster can be disruptive. YARN supports rolling upgrades wherein you upgrade nodes in small batches without bringing down the entire cluster. This ensures continuous availability while you stay up to date with the latest patches and features.

6. Large-Scale Mixed Workloads#

Enterprises might run streaming analytics (Spark Streaming or Flink), batch ETL jobs (MapReduce or Hive), and interactive queries (Hive on Tez or Impala) on the same cluster. YARN-based resource scheduling makes it feasible to isolate or prioritize certain workloads without deploying multiple clusters.