Performance Tuning, Cost Optimization / Internals, Research by Dmitry Tolpeko
In cloud it is typical to run multiple compute clusters, so browsing the Web UI for every cluster to check the current resource consumption by applications is not always easy and convenient especially if YARN clusters are managed by different Hadoop distributions (Amazon EMR, Cloudera, Qubole etc.).
Let’s see how you can automate this process and find out how many applications are running and which resources they are consuming (containers, memory and CPU).
You can expect that the total number of vCores available to YARN limits the number of containers you can run concurrently, that’s not true in some cases.
Let’s consider one of them – Capacity Scheduler with DefaultResourceCalculator (Memory only).
Let’s review how EC2 vCPUs correspond to YARN vCores in Amazon EMR and Qubole Hadoop clusters. As an example, I will choose m4.4xlarge, r4.4xlarge and c4.4xlarge EC2 instance types.
EC2 vCPU is a thread of a CPU core (typically, there are two threads per core). Does it mean that YARN vCores should be equal to the number of EC2 vCPU? That’s not always the case.
On one of my clusters I got my favorite YARN error, although now it was in a Flink application:
Container is running beyond physical memory limits. Current usage: 99.5 GB of 99.5 GB physical memory used; 105.1 GB of 227.8 GB virtual memory used. Killing container.
Why did the container take so much physical memory and fail? Let’s investigate in detail.
On one of the clusters I noticed an increased rate of shuffle errors, and the restart of a job did not help, it still failed with the same error.
The error was as follows:
Error: Error while running task ( failure ) :
org.apache.tez.runtime.library.common.shuffle.orderedgrouped.Shuffle$ShuffleError:
error in shuffle in Fetcher
at org.apache.tez.runtime.library.common.shuffle.orderedgrouped.Shuffle$RunShuffleCallable.callInternal
(Shuffle.java:301)
Caused by: java.io.IOException:
Shuffle failed with too many fetch failures and insufficient progress!failureCounts=1,
pendingInputs=1, fetcherHealthy=false, reducerProgressedEnough=true, reducerStalled=true
In the previous article about YARN container memory (see, Tez Memory Tuning – Container is Running Beyond Physical Memory Limits) I wrote about the physical memory. Now I would like to pay attention to the virtual memory in YARN.
A typical YARN memory error may look like this:
Container is running beyond virtual memory limits. Current usage: 1.0 GB of 1.1 GB physical memory used; 2.9 GB of 2.4 GB virtual memory used. Killing container.
So what is the virtual memory, how to solve such errors and why is the virtual memory size often so large?
In the previous article Calculating Utilization of Cluster using Resource Manager Logs I showed how to estimate per-second utilization for a Hadoop cluster.
This information can be useful to calculate the idle time statistics for a cluster i.e. time when no any containers are running.
I was asked to tune a Hive query that ran more than 10 hours. It was running on a 100 node cluster with 16 GB available for YARN containers on each node.
Although the query processed about 2 TB of input data, it did a fairly simple aggregation on user_id column and did not look too complex. It had 1 Map stage with 1,500 tasks and 1 Reduce stage with 7,000 tasks.
All map tasks completed within 30 minutes, and the query stuck on the Reduce phase. So what was wrong?
You can use YARN REST API to collect various Hadoop cluster metrics such as available and allocated memory, CPU, containers and so on.
If you set up a process to extract data from this API once per minute e.g. you can very easily collect and analyze historical and current cluster utilization quite accurately. For more details, see Collecting Utilization Metrics from Multiple Clusters article.
But even if you query YARN REST API every second it still can only provide a snapshot of the used YARN resources. It does not show which application allocates or releases containers, their memory and CPU capacity, in which order these events occur, what is their exact timestamp and so on.
For this reason I prefer a different approach that is based on using the YARN Resource Manager logs to calculate the exact per second utilization metrics of a Hadoop cluster.
Memory allocation in Hadoop YARN clusters has some drawbacks that may lead to significant cluster under-utilization and at the same time (!) to large queues of pending applications.
So you have to pay for extra compute resources that you do not use and still have unsatisfied users. Let’s see how this can happen and how you can mitigate this.