How to Configure Impala with Dedicated Coordinators

Each host that runs the Impala Daemon acts as both a coordinator and as an executor, by default, managing metadata caching, query compilation, and query execution. In this configuration, Impala clients can connect to any Impala daemon and send query requests.

During highly concurrent workloads for large-scale queries, the dual roles can cause scalability issues because:

  • The extra work required for a host to act as the coordinator could interfere with its capacity to perform other work for the later phases of the query. For example, coordinators can experience significant network and CPU overhead with queries containing a large number of query fragments. Each coordinator caches metadata for all table partitions and data files, which requires coordinators to be configured with a large JVM heap. Executor-only Impala daemons should be configured with the default JVM heaps, which leaves more memory available to process joins, aggregations, and other operations performed by query executors.

  • Having a large number of hosts act as coordinators can cause unnecessary network overhead, or even timeout errors, as each of those hosts communicates with the Statestored daemon for metadata updates.

  • The "soft limits" imposed by the admission control feature are more likely to be exceeded when there are a large number of heavily loaded hosts acting as coordinators. Check IMPALA-3649 and IMPALA-6437 to see the status of the enhancements to mitigate this issue.

The following factors can further exacerbate the above issues:

  • High number of concurrent query fragments due to query concurrency and/or query complexity

  • Large metadata topic size related to the number of partitions/files/blocks

  • High number of coordinator nodes

  • High number of coordinators used in the same resource pool

If such scalability bottlenecks occur, in CDH 5.12 / Impala 2.9 and higher, you can assign one dedicated role to each Impala daemon host, either as a coordinator or as an executor, to address the issues.

  • All explicit or load-balanced client connections must go to the coordinator hosts. These hosts perform the network communication to keep metadata up-to-date and route query results to the appropriate clients. The dedicated coordinator hosts do not participate in I/O-intensive operations such as scans, and CPU-intensive operations such as aggregations.

  • The executor hosts perform the intensive I/O, CPU, and memory operations that make up the bulk of the work for each query. The executors do communicate with the Statestored daemon for membership status, but the dedicated executor hosts do not process the final result sets for queries.

Using dedicated coordinators offers the following benefits:

  • Reduces memory usage by limiting the number of Impala nodes that need to cache metadata.

  • Provides better concurrency by avoiding coordinator bottleneck.

  • Eliminates query over-admission.

  • Reduces resource, especially network, utilization on the Statestored daemon by limiting metadata broadcast to a subset of nodes.

  • Improves reliability and performance for highly concurrent workloads by reducing workload stress on coordinators. Dedicated coordinators require 50% or fewer connections and threads.

  • Reduces the number of explicit metadata refreshes required.

  • Improves diagnosability if a bottleneck or other performance issue arises on a specific host, you can narrow down the cause more easily because each host is dedicated to specific operations within the overall Impala workload.

In this configuration with dedicated coordinators / executors, you cannot connect to the dedicated executor hosts through clients such as impala-shell or business intelligence tools as only the coordinator nodes support client connections.

Determining the Optimal Number of Dedicated Coordinators

You should have the smallest number of coordinators that will still satisfy your workload requirements in a cluster. A rough estimation is 1 coordinator for every 50 executors.

To maintain a healthy state and optimal performance, it is recommended that you keep the peak utilization of all resources used by Impala, including CPU, the number of threads, the number of connections, and RPCs, under 80%.

Consider the following factors to determine the right number of coordinators in your cluster:

  • What is the number of concurrent queries?

  • What percentage of the workload is DDL?

  • What is the average query resource usage at the various stages (merge, runtime filter, result set size, etc.)?

  • How many Impala Daemons (impalad) is in the cluster?

  • Is there a high availability requirement?

  • Compute/storage capacity reduction factor

Start with the below set of steps to determine the initial number of coordinators:

  1. If your cluster has less than 10 nodes, we recommend that you configure one dedicated coordinator. Deploy the dedicated coordinator on a DataNode to avoid losing storage capacity. In most of the cases, one dedicated coordinator is enough to support all workloads on a cluster.
  2. Add more coordinators if the dedicated coordinator CPU or network peak utilization is 80% or higher. You might need 1 coordinator for every 50 executors.
  3. If the Impala service is shared by multiple workgroups with a dynamic resource pool assigned, use one coordinator per pool to avoid admission control over admission.
  4. If high availability is required, double the number of coordinators. One set as an active set and the other as a backup set.

Advanced Tuning

Use the following guidelines to further tune the throughput and stability.
  1. The concurrency of DML statements does not typically depend on the number of coordinators or size of the cluster. Queries that return large result sets (10,000+ rows) consume more CPU and memory resources on the coordinator. Add one or two coordinators if the workload has many such queries.
  2. DDL queries, excluding COMPUTE STATS and CREATE TABLE AS SELECT, are executed only on coordinators. If your workload contains many DDL queries running concurrently, you could add one coordinator.
  3. The CPU contention on coordinators can slow down query executions when concurrency is high, especially for very short queries (<10s). Add more coordinators to avoid CPU contention.
  4. On a large cluster with 50+ nodes, the number of network connections from a coordinator to executors can grow quickly as query complexity increases. The growth is much greater on coordinators than executors. Add a few more coordinators if workloads are complex, i.e. (an average number of fragments * number of Impalad) > 500, but with the low memory/CPU usage to share the load. Watch IMPALA-4603 and IMPALA-7213 to track the progress on fixing this issue.
  5. When using multiple coordinators for DML statements, divide queries to different groups (number of groups = number of coordinators). Configure a separate dynamic resource pool for each group and direct each group of query requests to a specific coordinator. This is to avoid query over admission.
  6. The front-end connection requirement is not a factor in determining the number of dedicated coordinators. Consider setting up a connection pool at the client side instead of adding coordinators. For a short-term solution, you could increase the value of fe_service_threads on coordinators to allow more client connections.
  7. In general, you should have a very small number of coordinators so storage capacity reduction is not a concern. On a very small cluster (less than 10 nodes), deploy a dedicated coordinator on a DataNode to avoid storage capacity reduction.

Estimating Coordinator Resource Usage

Resource Safe range Notes / CM tsquery to monitor
Memory

(Max JVM heap setting +

query concurrency *

query mem_limit)

<=

80% of Impala process memory allocation

Memory usage:

SELECT mem_rss WHERE entityName = "Coordinator Instance ID" AND category = ROLE

JVM heap usage (metadata cache):

SELECT impala_jvm_heap_current_usage_bytes WHERE entityName = "Coordinator Instance ID" AND category = ROLE (only in release 5.15 and above)
TCP Connection Incoming + outgoing < 16K

Incoming connection usage:

SELECT thrift_server_backend_connections_in_use WHERE entityName = "Coordinator Instance ID" AND category = ROLE

Outgoing connection usage:

SELECT backends_client_cache_clients_in_use WHERE entityName = "Coordinator Instance ID" AND category = ROLE
Threads < 32K SELECT thread_manager_running_threads WHERE entityName = "Coordinator Instance ID" AND category = ROLE
CPU

Concurrency =

non-DDL query concurrency <=

number of virtual cores allocated to Impala per node

CPU usage estimation should be based on how many cores are allocated to Impala per node, not a sum of all cores of the cluster.

It is recommended that concurrency should not be more than the number of virtual cores allocated to Impala per node.

Query concurrency:

SELECT total_impala_num_queries_registered_across_impalads WHERE entityName = "IMPALA-1" AND category = SERVICE

If usage of any of the above resources exceeds the safe range, add one more coordinator.

Monitoring Coordinator Resource Usage

Using Cloudera Manager, monitor the coordinator resource usage to understand your workload and adjust the number of coordinators according to the guidelines above. The available options are:
  • Impala Queries tab: Monitor such attributes as DDL queries and Rows produced. See Monitoring Impala Queries for detail information.
  • Custom charts: Monitor activities, such as query complexity which is an average fragment count per query (total fragments / total queries).
  • tsquery: Build the custom charts to monitor and estimate the amount of resource the coordinator needs. See tsquery Language for more information.

The following are sample queries for common resource usage monitoring. Replace entityName values with your coordinator instance id.

Per coordinator tsquery
Resource Usage Tsquery
Memory usage SELECT impala_memory_total_used, mem_tracker_process_limit WHERE entityName = "Coordinator Instance ID" AND category = ROLE
JVM heap usage (metadata cache) SELECT impala_jvm_heap_current_usage_bytes WHERE entityName = "Coordinator Instance ID" AND category = ROLE (only in release 5.15 and above)
CPU usage SELECT cpu_user_rate / getHostFact(numCores, 1) * 100, cpu_system_rate / getHostFact(numCores, 1) * 100 WHERE entityName="Coordinator Instance ID"
Network usage (host level) SELECT total_bytes_receive_rate_across_network_interfaces, total_bytes_transmit_rate_across_network_interfaces WHERE entityName="Coordinator Instance ID"
Incoming connection usage SELECT thrift_server_backend_connections_in_use WHERE entityName = "Coordinator Instance ID" AND category = ROLE
Outgoing connection usage SELECT backends_client_cache_clients_in_use WHERE entityName = "Coordinator Instance ID" AND category = ROLE
Thread usage SELECT thread_manager_running_threads WHERE entityName = "Coordinator Instance ID" AND category = ROLE
Cluster wide tsquery
Resource usage Tsquery
Front-end connection usage SELECT total_thrift_server_beeswax_frontend_connections_in_use_across_impalads, total_thrift_server_hiveserver2_frontend_connections_in_use_across_impalads
Query concurrency SELECT total_impala_num_queries_registered_across_impalads WHERE entityName = "IMPALA-1" AND category = SERVICE

Deploying Dedicated Coordinators and Executors in Cloudera Manager

This section describes the process to configure a dedicated coordinator and a dedicated executor roles for Impala.

  • Dedicated coordinator:
    • Should be on an edge node with no other services running on it.
    • Does not need large local disks but still needs some that can be used for Spilling.
    • Require at least the same or even larger memory allocation than executors.
  • (Dedicated)Executors:
    • Should be colocated with DataNodes.
    • The number of hosts with dedicated executors typically increases as the cluster grows larger and handles more table partitions, data files, and concurrent queries.
To configuring dedicated coordinators/executors:
  1. Navigate to Clusters > Impala > Configuration > Role Groups.
  2. Click Create to create two role groups with the following values.
    1. Group for Coordinators
      1. Group Name: Coordinators
      2. Role Type: Impala Daemon
      3. Copy from:
        • Select Impala Daemon Default Group if you want the existing configuration gets carried over to the Coordinators.
        • Select None if you want to start with a blank configuration.
    2. Group for Executors
      1. Group Name: Executors
      2. Role Type: Impala Daemon
      3. Copy from:
        • Select Impala Daemon Default Group if you want the existing configuration gets carried over to the Executors.
        • Select None if you want to start with a blank configuration.
  3. In the Role Groups page, click Impala Daemon Default Group.
    1. Select the set of nodes intended to be coordinators.
      1. Click Action for Selected and select Move To Different Role Group….
      2. Select the Coordinators.
    2. Select the set of nodes intended to be Executors.
      1. Click Action for Selected and select Move To Different Role Group….
      2. Select Executors.
  4. Click Configuration. In the search field, type Impala Daemon Specialization.
  5. Click Edit Individual Values.
  6. For Coordinators role group, select COORDINATOR_ONLY.
  7. For Executors role group, select EXECUTOR_ONLY.
  8. Click Save Changes and then restart the Impala service.

Deploying Dedicated Coordinators and Executors from Command Line

To configuring dedicated coordinators/executors, you specify one of the following startup flags for the impalad daemon on each host:

  • ‑‑is_executor=false for each host that does not act as an executor for Impala queries. These hosts act exclusively as query coordinators. This setting typically applies to a relatively small number of hosts, because the most common topology is to have nearly all DataNodes doing work for query execution.

  • ‑‑is_coordinator=false for each host that does not act as a coordinator for Impala queries. These hosts act exclusively as executors. The number of hosts with this setting typically increases as the cluster grows larger and handles more table partitions, data files, and concurrent queries. As the overhead for query coordination increases, it becomes more important to centralize that work on dedicated hosts.