Apache Druid Cluster Tuning & Resource Management

Source Discussion: Based on detailed tuning guides and community discussions.

Answer: Correctly configuring threads and buffers is a balancing act to maximize CPU utilization without causing contention.

On Historical Nodes (The Workhorses):

• druid.processing.numThreads: This is the most critical performance parameter on a data node. It defines the maximum number of segments that can be processed concurrently. The standard recommendation is (Number of CPU Cores) - 1. This aims to fully saturate the CPU while leaving one core for the OS and background tasks.
• druid.processing.buffer.sizeBytes: These are off-heap buffers for intermediate query results. A starting value of 512MiB to 1GiB is common. If a query’s intermediate results on a single segment exceed this, it will spill to disk, causing a massive performance drop.
• druid.processing.numMergeBuffers: A dedicated pool of buffers for GroupBy queries. A reasonable starting point is a quarter of druid.processing.numThreads. If the mergeBuffer/pendingRequests metric is consistently greater than zero, this value should be increased.

On Broker Nodes (The Coordinators):

• Brokers primarily merge final results and don’t do heavy segment processing. The key thread pool is druid.server.http.numThreads, which should be sized based on expected client concurrency. They also require numMergeBuffers for the final merge step of GroupBy queries.

Answer: Memory sizing is crucial for stability and performance. Druid uses both JVM heap and off-heap direct memory.

Historical Nodes:

• Heap (-Xmx): The heap is used for partial query results and lookups. A good starting point is (0.5GiB * number of CPU cores). If you use lookups, you must add (2 * total size of all loaded lookups) to this value.
• Direct Memory (-XX:MaxDirectMemorySize): This must be large enough to hold all processing and merge buffers. Calculate it with the formula: (druid.processing.numThreads + druid.processing.numMergeBuffers + 1) * druid.processing.buffer.sizeBytes.
• Page Cache: The remaining system memory is used by the OS as a page cache to memory-map segments. For best performance, you want enough free system memory to cache your most frequently accessed (“hot”) segments.

Broker Nodes:

• Heap (-Xmx): The Broker heap scales with the number of segments in the cluster, as it holds segment timeline information. For small to medium clusters, 4-8GiB is a reasonable start. Very large clusters might need 30-60GiB.
• Direct Memory (-XX:MaxDirectMemorySize): Brokers only need direct memory for merge buffers. The formula is (druid.processing.numMergeBuffers + 1) * druid.processing.buffer.sizeBytes.

Source Discussion: Addresses the common “noisy neighbor” problem of resource contention.

Answer: To manage mixed workloads, you must implement workload management strategies. Druid provides two primary mechanisms: Query Laning for logical isolation and Service Tiering for physical isolation.

• Query Laning (Soft Isolation): This feature creates logical partitions within the Broker’s query scheduler. You can define different “lanes” (e.g., ‘high_priority’, ‘low_priority’) and assign a percentage of the Broker’s query threads to each. This ensures that even if the low-priority lane is saturated with heavy analytical jobs, there are always dedicated resources for high-priority dashboard queries to execute immediately.
• Service Tiering (Hard Isolation): This provides complete physical resource isolation. It involves creating different tiers of Historical nodes (e.g., a ‘hot’ tier on fast SSDs for recent data, a ‘cold’ tier on cheaper hardware for old data). You can then deploy separate sets of Brokers configured to query only specific tiers. This guarantees that a heavy query on historical data can never consume resources from the nodes serving your real-time dashboards.

Answer: Druid supports two main types of query caching to improve performance for frequently accessed data. Caching is most effective for increasing concurrency, not for speeding up a single, slow query.

• Per-Segment Caching (Default): This is the primary form of caching, enabled by default on Historicals. It stores partial query results on a per-segment basis. When a query comes in, if the results for a specific segment are in the cache, they are returned immediately without hitting the processing threads. This is highly effective for segments that no longer change.
• Whole-Query Caching: This cache lives on the Broker and stores the final, complete result of a query. It is useful when there is little risk of underlying data changing, such as in a pure batch environment.

Configuration and Best Practices:

• Caching can be enabled or disabled on a per-query basis using context flags like useCache and populateCache.
• The largest performance gains from caching are seen with TopN and Timeseries queries. For GroupBy queries, if the bottleneck is the final merge step on the Broker, caching provides less benefit.
• By default, Druid uses an in-heap Caffeine cache. For larger clusters, an external distributed cache like Memcached can be used.
• Don’t rely on caching to fix a slow query. Your queries should meet performance goals even with a cold cache. Caching is a tool for handling high-concurrency workloads.