I'd like to thank @Jean-Philippe Player, @bpreachuk, @ghagleitner, @gopal, @ndembla and @Prasanth Jayachandran for providing input and content for this article.
Introduction
This document describes LLAP setup for reasonable performance with a typical workload. It is intended as a starting point, not as the definitive answer to all tuning questions. However, recommendations in this document have been applied both to our own instances as well as instances we help other teams to run.
The context for the examples
Each step below will be accompanied by an example based on allocating ~50% of a cluster to LLAP, out of 12 nodes with 16 cores and 96Gb dedicated to YARN NodeManager each.
Basic cluster setup
Pick the fraction of the cluster to be used by LLAP
This is determined by the workload (amount of data, concurrency, types of queries, desired latencies); typical values can be in 15-50% range on an existing cluster, or it is possible to give entire cluster to LLAP, depending on customer use case.
LLAP runs 3 types of YARN containers in the cluster:
a) Execution daemons which process the data
b) query coordinators (Tez AMs) orchestrating the execution of the queriues
and c) a single Slider AM starting and monitoring the daemons.
The bulk of the capacity allotted for LLAP will be used by the Execution daemons. For best results, whole YARN nodes should be allocated for these daemons; therefore, you should pick the number of whole nodes to use based on the fraction of the cluster to be dedicated to LLAP.
Example: we want to use 50% of a 12-node the cluster, so we will have 6 LLAP daemons.
Pick the query parallelism and number of coordinators
This will depend heavily on the application; one thing to be aware of is that with BI tools, the number of running queries at any moment in time is typically much smaller than the number of sessions -users don’t run queries all the time because they take time to look at the results and some tools avoid queries using caching.
The number of queries determines the number of query coordinators (Tez AMs) - one per query + some constant slack in case sessions are temporarily unavailable due to maintenance or cluster issues.
Example: we have determined we need to run 9 queries in parallel; use 10 AMs.
Determine LLAP daemon size
As described above, we will give whole NM nodes to LLAP daemons; however, we also need to leave some space on the cluster for query coordinators and the slider AM.
There are two options. First, it is possible to run these on non-LLAP nodes. In this case, each LLAP daemon can take the entire memory of a NM (see YARN config tab in Ambari, as "Memory allocated for all YARN containers on a node". 😞
If that is acceptable, it's the optimal approach (it maximizes daemon memory and avoids noisy neighbors). However,
- this will use more combined resources than just the (number of LLAP nodes)/(total number of nodes), because AMs will take resources on other nodes.
- it's impossible if all NMs are running LLAP daemons (~100% LLAP cluster), and risky even when there are 1-2 non-LLAP NMs (because all the query coordinators would be concentrated on one node).
The alternative approach is to run AMs on the nodes with LLAP daemons. In this case, LLAP daemon container size needs to be adjusted down to leave space:
There are (N_Tez_AMs + 1)/N_LLAP_nodes extra processes on each node; we recommend 2Gb per AM process. So, for this case, LLAP daemon will use (NM_memory – 2Gb * ceiling((N_Tez_AMs + 1)/N_LLAP_nodes)) .
Example: We will use the 2nd approach; we need ceiling((10+1)/6) = 2 AM processes per LLAP node, reserving 2 * 2Gb per node. Therefore, LLAP daemon size will be 96Gb – 4Gb = 92Gb.
Set up YARN and the LLAP queue
YARN settings for LLAP, as well as the LLAP queue are usually set up by Ambari; if setting up manually or encountering problems, there are a few things to ensure:
- YARN preemption should be enabled. Even if there is enough memory in the cluster for execution daemons, it might be too fragmented for Yarn to allocate whole nodes to LLAP. Preemption ensures that Yarn can free up memory and start LLAP, without it startup might not succeed.
- Minimum YARN container size should be 1024Mb and maximum should be the total memory per NM. LLAP daemon is a container so it will not be able to use more memory than that.
When setting up a custom LLAP queue:
- Set the queue size >= (2Gb * N_TEZ_AMs) + (1Gb [slider AM]) + (daemon_size * N_LLAP_nodes) .
- Ensure the queue has higher priority that other YARN queues (for preemption to work; see the above YARN setup section).
- Make sure queue min and max capacity is the same (full capacity), and there are no user limits on the queue.
Determine the LLAP daemon CPU & memory configuration
Overview
There are three numbers that determine the memory configuration for LLAP daemons.
- Total memory: The container size that was determined above
- Xmx: Memory used by executors, and
- Off-heap cache memory.
Total memory = off-heap cache memory + Xmx (used by executors and daemon-specific overhead) + JVM overhead . The Xmx overhead (shared Java objects and internal structures) is accounted for by LLAP internally, so it can be ignored in the calculations.
In most cases, you can just follow the standard steps below ; however, sometimes the process needs to be tweaked. The following conflicting criteria apply:
- It's good to maximize task parallelism (up to one executor per core).
- It's good to have at least 4 Gb RAM per executor. More memory can provide some diminishing returns; less can sometimes be acceptable but can lead to problems (OOMs, etc.).
- It's good to have IO layer enabled, which requires some cache space; more cache may be useful depending on workload, esp. on a cloud FS for BI queries.
So, number of executors, cache size, and then memory per executor are balance of these 3:
Pick the number of executors per node
Usually the same as the number of cores.
Example: there are 16 cores, so we will use 16 executors.
Determine the Xmx and minimum cache memory needed by executors
Daemon uses almost all of its heap memory (Xmx) for processing query work; the shared overheads are accounted for by LLAP. Thus, we determine the daemon Xmx based on executor memory; you should aim to give a single executor around 4Gb. So, Xmx = n_executors * 4Gb. LLAP executors use memory for things like group by tables, sort buffers, shuffle buffers, map join hashtables, PTF, etc.; generally, the more memory, the less likely they are to spill, or to go with a less efficient plan, e.g. switching from a map to shuffle join - for some background, see https://community.hortonworks.com/articles/149998/map-join-memory-sizing-for-llap.html
If there isn't enough memory on the machine, by default, adjust the number of executors down. It is possible to give executors less memory (e.g. 3Gb) to keep their number higher, however, that must always be backed by testing with the target workload. Insufficient memory can cause much bigger perf problems than reduced task parallelism.
Additionally, if IO layer is enabled, each executor would need about ~200Mb of cache (cache buffers are also used as IO buffer pool) for reasonable operation. The cache size will be covered in the next section.
Example: Xmx = 16
executors * 4 Gb = 64 Gb
Additionally, based on 16 executors, we know that it's good to have (0.2 Gb *
16) = 3.2 Gb cache if we are going to enable it.
Determine headroom and cache size
We need to leave some memory in the daemon for Java VM overhead (GC structures, etc.) as well as to give the container a little safety margin to prevent it being killed by YARN. This headroom should be about 6% of Xmx, but no more than 6 Gb . This is not set directly and is instead used to calculate the container memory remaining for the use by off-heap cache.
Take out Xmx and the headroom from the total container size to determine cache size. If the cache size is below the viable size above, reduce Xmx to allow for more memory here (typically by reducing the number of executors) or disable LLAP IO (hive.llap.io.enabled = false).
Example:
Xmx is 64 Gb, so headroom = 0.06 * 64 Gb = ~4 Gb (which is lower than 6 Gb).
Cache size = 92 Gb total – 64 Gb Xmx – 4 Gb headroom = 24 Gb. This is above the minimum of 3.2 Gb.
Configure interactive query
You can specify the above settings on the main interactive query panel, as well as in advanced config (or validate the values predetermined by Ambari).
Some other settings from the advanced config that may be of interest:
- AM size can be modified in custom Hive config by changing tez.am.resource.memory.mb.
