Impala Scale, in depth: dedicated coordinators, metadata size, on-demand catalog, admission control across coordinators and sizing a large cluster

Four axes of growth, four different bottlenecks

It helps to stop thinking of Impala scale as a single number of nodes. Growth happens on four independent axes, and each stresses a different component.

The Impala documentation lists the same pressures when it explains dedicated coordinators: a high number of concurrent query fragments, a large metadata topic driven by partitions, files and blocks, many coordinator nodes, and many coordinators sharing one resource pool. Every fix in this article targets one of those four.

Why the control plane is the first thing to break

In the classic configuration, catalogd loads table metadata from the Hive Metastore and the file system, and publishes it as a catalog topic through the statestore. Every impalad that subscribes receives a copy and keeps the whole catalog in its own JVM heap. The cost therefore grows with the product of two numbers: how large the metadata is and how many daemons subscribe to it. A cluster of 300 combined-role daemons holds 300 copies of the catalog, and a large ALTER TABLE ... ADD PARTITION is serialised by catalogd and shipped to 300 subscribers.

Metadata size is driven far more by files and blocks than by tables. A table with 50,000 partitions and 40 small files per partition has two million file descriptors to track, each with block locations on HDFS. Nothing about that table looks large in a data catalog, yet it can dominate catalogd heap and topic size. The symptom is not slow queries at first; it is catalogd garbage-collection pauses, statestore update latency, coordinators whose heap never goes down, and DDL that takes seconds to become visible on other coordinators.

Dedicated coordinators and executors

The first and cheapest structural change is to stop giving every daemon both roles. Start coordinators with --is_executor=false and executors with --is_coordinator=false. Executors no longer plan queries, so they do not need table metadata in the planning sense, which removes most catalog broadcast traffic and heap from the bulk of the cluster. Coordinators no longer run scans and joins, so their CPU and memory are available for planning, admission, result handling and client sessions.

# coordinator hosts
impalad --is_coordinator=true --is_executor=false --use_local_catalog=true

# executor hosts
impalad --is_coordinator=false --is_executor=true

# catalogd
catalogd --catalog_topic_mode=minimal

How many coordinators? The Impala documentation gives a rough estimate of one coordinator for every 50 executors and suggests a single dedicated coordinator below about ten nodes. It recommends adding coordinators when coordinator CPU or network peaks at 80% or more, when queries are complex (its guide is average fragments multiplied by impalad count above 500), or when DDL concurrency is high, and doubling the count if you need high availability. Treat these as starting points to measure against, not as a formula.

Clients must spread across coordinators. Put a load balancer in front of them with session affinity for protocols that keep session state, and make sure no single BI tool is pinned to one coordinator by a hard-coded host name.

Shrinking metadata: on-demand catalog mode

Dedicated roles reduce the number of catalog copies; on-demand metadata reduces the size of each one. In this mode, catalogd runs with --catalog_topic_mode=minimal and coordinators run with --use_local_catalog=true. The catalog topic then carries only minimal invalidation information, and each coordinator fetches metadata from catalogd when a query needs it, at partition granularity, caching it locally. Cloudera's documentation describes the cache as defaulting to 60% of the coordinator's JVM heap (local_catalog_cache_mb=-1) with a one-hour expiry (local_catalog_cache_expiration_s=3600). A mixed mode, --catalog_topic_mode=mixed, lets old-style and on-demand coordinators coexist during migration.

The trade-off is a cold-cache penalty. The first query after a coordinator restart, or after an invalidation, waits for metadata to be fetched from catalogd; a dashboard that touches hundreds of tables on its first load will feel slower. In exchange, coordinator heap is bounded by what is actually queried, and adding a partition no longer means serialising and broadcasting a large update. For clusters with many tables, only a fraction of which are hot, this is almost always the right trade.

Refresh discipline and file counts

Metadata operations have very different costs at scale, and the wrong habit can undo everything above. INVALIDATE METADATA with no table name discards the whole catalog and forces every table to reload lazily; on a large cluster it is an outage-sized operation and should be restricted by policy. INVALIDATE METADATA db.table drops one table's metadata. REFRESH db.table reloads file metadata for a table incrementally, and REFRESH db.table PARTITION (...) touches a single partition, which is what ingestion jobs should run after adding data.

-- after an ingestion job writes one day's data
REFRESH sales.orders PARTITION (dt='2026-09-30');

-- after statistics-relevant changes, not after every load
COMPUTE INCREMENTAL STATS sales.orders PARTITION (dt='2026-09-30');

-- never in a scheduled job on a large cluster
-- INVALIDATE METADATA;

File counts are the other lever. Streaming ingestion that writes a small file every minute into hourly partitions produces thousands of files a day per table. Each file costs metadata, a scan range and an open call. Compact small files into larger Parquet files on a schedule, choose partition columns so that typical partitions hold at least a few large files, and avoid partitioning by a high-cardinality column just because queries filter on it; Parquet statistics and runtime filters often prune well enough without it.

Incremental statistics add per-partition state to metadata, so on tables with very many partitions consider periodic full COMPUTE STATS instead.

Admission control with many coordinators

Admission control decides whether a query can start based on pool limits and per-host memory. In a cluster with several coordinators, each coordinator admits queries independently, using pool usage information that the other coordinators publish through the statestore. That view is eventually consistent. If a burst of queries arrives at five coordinators at the same moment, each may see headroom that the others are about to consume, and together they over-admit. The executors then hit their memory limits and queries fail or spill, rather than waiting politely in a queue.

The mitigations are structural. Keep pool limits conservative relative to real executor memory so that brief over-admission does not exhaust it. Set a realistic per-query memory limit in each pool so that admission decisions are based on reserved memory rather than optimistic estimates. Split distinct workloads into distinct pools, and avoid funnelling every heavy query through one pool served by many coordinators, since that is exactly the pattern the documentation calls out. Watch queue lengths and queue timeouts per pool; a rising queue is the signal to add executors or reshape the workload, while rising failures with a short queue suggest the limits are too loose.

Executor groups and elastic capacity

Impala can partition executors into executor groups. Coordinators schedule each query onto a single group, which bounds fragment fan-out: a 200-executor cluster split into four groups of 50 creates at most 50 fragments per plan node instead of 200, and a failed or restarting group does not stall the others. Groups also make it possible to add or remove capacity in whole units, which is how elastic deployments grow and shrink.

Cloudera Data Warehouse builds workload-aware auto-scaling on this idea: it defines executor group sets of different sizes, maps each to an admission pool, and replans a query onto a larger group set if its memory and CPU estimates do not fit the smaller one. That behaviour is a Cloudera Data Warehouse feature, not something a plain Apache Impala deployment does on its own. On self-managed clusters, executor groups are still useful for isolation and fault containment, but sizing and scaling them is your job.

Worked example: sizing a 200-executor cluster

Assume a platform with 200 executor hosts, around 4,000 tables, of which perhaps 300 are queried daily, one very large fact table with 30,000 partitions, 60 concurrent BI queries at peak and nightly batch loads. The numbers below are illustrative planning inputs, not measurements.

1. Coordinators. The 1:50 guidance gives four. Typical BI plans here have about 15 fragments, and 15 multiplied by 200 hosts is 3,000, well above the documentation's threshold of 500 for complex workloads, so start with six and double for high availability only if a coordinator outage must not reduce capacity. Six to eight coordinators behind a load balancer is a reasonable first deployment.
2. Metadata. With 4,000 tables and a small hot set, enable on-demand metadata from day one. Coordinators then cache the 300 hot tables instead of 4,000, and the large fact table's partition list is fetched only for partitions that queries touch.
3. Files. Set a compaction target so that each daily partition of the fact table holds files of hundreds of megabytes, and make the ingestion job run a partition-level REFRESH.
4. Pools. Create separate pools for BI, batch and ad hoc work, with per-query memory limits and queue timeouts. Route batch jobs to a coordinator subset so their bursts do not over-admit alongside BI traffic.
5. Verification. Load-test with a replay of real queries while watching coordinator CPU, catalogd heap and GC, statestore topic sizes and pool queues. Add coordinators when CPU or network peaks at 80%.

# rough planning helper; replace inputs with your own measurements
def plan(executors, avg_fragments, peak_concurrency, ha=False):
    coords = max(1, round(executors / 50))
    if avg_fragments * executors > 500:
        coords += 2                      # complex plans: start with headroom
    if peak_concurrency > 40 * coords:   # assumption: tune from measured CPU
        coords = -(-peak_concurrency // 40)
    return coords * 2 if ha else coords

print(plan(executors=200, avg_fragments=15, peak_concurrency=60))  # 6

Operating a large cluster

Every daemon has a debug web UI: impalad on port 25000, statestored on 25010 and catalogd on 25020 by default. On the statestore, the topics page shows the size of each topic and the number of subscribers, which is the single most useful scale metric to trend. On catalogd, watch JVM heap after GC, the number of tables loaded and the duration of metadata operations. On coordinators, watch heap, planning time in query profiles, and admission queue state per pool. Query profiles show time spent waiting for metadata separately from execution, which tells you whether a slow query is a data problem or a catalog problem.

Roll coordinator restarts behind the load balancer; restarting all of them at once in on-demand mode makes every first query pay the cold-cache penalty simultaneously.

Failure modes

• Catalog topic explosion. A partition-heavy table grows; catalogd GC pauses and statestore updates lag. Enable on-demand metadata, compact files and reduce partition counts.
• Combined roles on every node. Hundreds of full catalog copies and heap pressure on executors. Move to dedicated coordinators.
• Scheduled global invalidate. Every table reloads and first queries stall cluster-wide. Replace with table or partition refreshes.
• Over-admission across coordinators. Memory-limit failures during bursts even though pools looked healthy. Tighten pool limits and per-query memory limits, and split workloads by pool.
• Coordinator hot spot. One coordinator saturates because clients connect by host name. Put coordinators behind a load balancer.
• Cold caches after restart. Slow first queries after a mass restart in on-demand mode. Roll restarts gradually and pre-warm hot tables with cheap queries.

Related reading

For the mechanics behind these levers, read the catalog service, the statestore, metadata management in depth, admission control, memory limits and the data cache for remote storage.

What to do next

1. Chart statestore topic sizes and subscriber counts, catalogd heap after GC, and coordinator CPU, and keep them on one dashboard.
2. Separate roles: start with the 1:50 coordinator guidance and adjust from measured CPU and network peaks.
3. Enable on-demand metadata on catalogd and coordinators, using mixed mode for a staged rollout.
4. Ban unscoped INVALIDATE METADATA in jobs and make ingestion run partition-level REFRESH.
5. Find the ten tables with the most files and partitions and schedule compaction or re-partitioning.
6. Review admission pools: separate workloads, set per-query memory limits and queue timeouts, and alert on queue growth and memory-limit failures.
7. Put coordinators behind a load balancer and rehearse a rolling restart.