Elasticsearch Aggregations at Scale: Accuracy, Memory, Limits

Aggregations break at scale in two directions. They become wrong, because a terms aggregation over many shards is an approximation that returns a confident-looking top ten which may not be the true top ten. And they become expensive, because every bucket, every cardinality sketch and every nested sub-aggregation lives on the heap until the response is built, so one dashboard panel can trip a circuit breaker on the coordinating node. The answers are specific: understand the shard-level reduce and use shard_size and the error fields to control and measure terms accuracy; use composite to page through all buckets instead of asking for a huge size; size cardinality memory from precision_threshold; count buckets before the cluster does it for you with search.max_buckets; and move repeated, heavy aggregations out of query time into transforms. This article covers each with the arithmetic.

The mental model: map on shards, reduce on the coordinator

An aggregation request executes in two stages.

1. Shard stage. Each shard computes the aggregation over its own documents, using doc values (the columnar on-disk structure) for keyword, numeric, date and similar fields. For terms on a keyword field, it typically works on global ordinals: a per-shard mapping from each unique term to an integer, built on first use after a refresh and reused until the next refresh changes the segments.
2. Reduce stage. The coordinating node receives a partial result from every shard and merges them. For large requests the coordinator reduces in batches as results arrive (batched_reduce_size, default 512 shard results), but the merged result for the whole request is held in memory on that node until it is returned.

Every accuracy problem in this article comes from stage 1 discarding information before stage 2 can see it. Every memory problem comes from either stage holding too many buckets.

Two consequences to keep in mind:

• Single-shard aggregations are exact for terms counts, because there is nothing to merge. If an index has one shard, or your query uses routing that hits one shard, the error discussion below does not apply.
• The coordinating node is a memory hotspot. If clients send heavy aggregations to data nodes directly, those data nodes pay the reduce cost on top of their shard work. Dedicated coordinating nodes exist largely for this workload. See cluster topology.

Terms accuracy: why the top ten can be wrong

Ask for the top 10 hosts by document count across 5 shards. Each shard does not return all its terms; it returns its own top shard_size terms. The default is:

shard_size = size x 1.5 + 10

For size: 10, that is 10 x 1.5 + 10 = 25. Each shard returns its top 25; the coordinator sums them and keeps the top 10.

A term that is moderately common everywhere but never in any shard's top 25 is undercounted or missing. A term that is in the top 25 on three shards but ranks 26th on the other two gets a count that excludes those two shards' contributions.

Elasticsearch tells you how bad it could be. The response includes:

{
  "aggregations": {
    "hosts": {
      "doc_count_error_upper_bound": 0,
      "sum_other_doc_count": 1802211,
      "buckets": [
        { "key": "web-14", "doc_count": 912003 },
        { "key": "web-02", "doc_count": 899120 }
      ]
    }
  }
}

• sum_other_doc_count: documents in terms that did not make the returned list. A large value is normal; it just means the long tail exists.
• doc_count_error_upper_bound: the maximum number of documents any term not in the list could have had and still been missed. If this is 0, the returned list is correct. If it is large compared with the counts of your bottom buckets, a term you did not see might belong in the top 10.

For per-bucket error, request it:

GET logs-*/_search
{
  "size": 0,
  "aggs": {
    "hosts": {
      "terms": {
        "field": "host.name",
        "size": 10,
        "shard_size": 100,
        "show_term_doc_count_error": true
      }
    }
  }
}

Each bucket then carries doc_count_error_upper_bound, the worst-case undercount for that specific term.

Worked example: how large should shard_size be

Assume size: 10, 20 shards, and a long-tailed distribution where the 10th-ranked term has about 50,000 documents across the index and the 25th about 18,000. Suppose that with the default shard_size of 25 the reported error bound comes back at around 15,000, comparable to the gap between rank 10 and rank 11. The ranking at the bottom of the list is unreliable.

Raise shard_size to 100. Each shard now returns its top 100; the error bound falls (in a long-tailed distribution, the count at rank 100 per shard is much smaller), and the bottom of the top 10 stabilizes. The cost is that each shard sends 100 buckets instead of 25, and the coordinator merges 20 x 100 = 2,000 partial buckets instead of 500. That is trivial. Raising shard_size to 10,000 on 200 shards is not: 2,000,000 partial buckets on one coordinator.

A practical rule of thumb: start with shard_size around 5 to 10 times size for important rankings, read doc_count_error_upper_bound, and stop increasing when it is small compared with the counts you care about.

Ordering pitfalls

• Ordering by _count descending (the default) is the case the error bounds are designed for.
• Ordering by _count ascending ("rarest terms") has unbounded error, because the rarest term on one shard says nothing about the rarest overall. Use the rare_terms aggregation instead, which is built for that question.
• Ordering by a sub-aggregation (for example, top hosts by average latency) has error that Elasticsearch cannot bound in the same way. Treat such rankings as approximate across many shards, or compute them with composite paging and sort the full result client-side.

Composite: paging through every bucket

The wrong way to get all unique users is "size": 1000000 on a terms aggregation. It builds a million buckets on every shard and the coordinator, and it collides with search.max_buckets.

The right way is composite, which returns buckets in a stable sort order, a page at a time, with a cursor:

GET orders/_search
{
  "size": 0,
  "aggs": {
    "by_customer_day": {
      "composite": {
        "size": 1000,
        "sources": [
          { "customer": { "terms": { "field": "customer.id" } } },
          { "day": { "date_histogram": { "field": "@timestamp", "calendar_interval": "1d" } } }
        ]
      },
      "aggs": {
        "revenue": { "sum": { "field": "order.total" } }
      }
    }
  }
}

The response carries after_key:

{
  "aggregations": {
    "by_customer_day": {
      "after_key": { "customer": "C-004127", "day": 1760918400000 },
      "buckets": [
        { "key": { "customer": "C-000001", "day": 1760832000000 }, "doc_count": 3, "revenue": { "value": 121.4 } }
      ]
    }
  }
}

Send the next request with "after": { "customer": "C-004127", "day": 1760918400000 } inside composite, and repeat until a page comes back with no buckets.

What makes composite different:

• It is exact. Because it walks the key space in sorted order rather than picking a top N per shard, each page has correct counts for its buckets.
• It is bounded. Each request holds only size buckets (default 10; you will usually set hundreds to a few thousand).
• It can be cheap with index sorting. If the index is sorted on the same fields in the same order as the composite sources, Elasticsearch can terminate collection early on each shard. For a large reporting index that is paged this way every night, configuring the index sort to match is worth considering at creation time (index sort cannot be changed afterwards).
• It does not rank. Composite pages in key order, not by count. If you want the top N by count, use terms. If you want all buckets, use composite.

Sources can be terms, histogram, date_histogram and geotile_grid. Pipeline aggregations do not work with composite, so derivative-style calculations happen on the client side when you page.

Cardinality: precision_threshold and its memory cost

cardinality counts distinct values with HyperLogLog++. It is approximate by design, and its accuracy and memory are governed by precision_threshold:

• Default 3,000. Maximum 40,000. Values above 40,000 are treated as 40,000.
• Below the threshold, counts are expected to be close to exact. Above it, the error grows slowly as cardinality rises; Elastic's documentation charts the error against the threshold, and it is worth reading before you choose one.
• Memory is about precision_threshold x 8 bytes per aggregation instance at most, which is the figure to plan with.

That last line is where at-scale problems come from. A single cardinality at the default costs:

3,000 x 8 bytes = 24,000 bytes, about 24 KB

Harmless. Now nest it:

GET logs-*/_search
{
  "size": 0,
  "aggs": {
    "per_service": {
      "terms": { "field": "service.name", "size": 500 },
      "aggs": {
        "per_hour": {
          "date_histogram": { "field": "@timestamp", "fixed_interval": "1h" },
          "aggs": {
            "unique_users": { "cardinality": { "field": "user.id", "precision_threshold": 40000 } }
          }
        }
      }
    }
  }
}

Over 7 days: 500 services x 168 hours = 84,000 buckets. Each has a cardinality sketch of up to 40,000 x 8 = 320,000 bytes, about 320 KB.

84,000 x 320 KB = 26,880,000 KB, about 26 GB

The request circuit breaker (indices.breaker.request.limit, default 60% of heap) trips long before that, and the request fails with a circuit_breaking_exception. Even with the default threshold, 84,000 x 24 KB is about 2 GB per shard-level computation. The lesson: a sketch's precision should be set for the bucket it lives in, and sketches nested inside high-cardinality bucket aggregations should keep low thresholds or move to a transform.

Also worth knowing: cardinality on a field with very high distinct counts benefits from pre-computed hashes. The murmur3 mapper plugin can store a hash at index time to make cardinality cheaper. That is a plugin, and adds index size; consider it only for fields you count distinct values on constantly.

search.max_buckets and counting before you query

Elasticsearch limits the number of buckets one search response can build with search.max_buckets, a dynamic cluster setting that defaults to 65,536. Exceeding it fails the request with a too_many_buckets_exception.

Buckets multiply through nesting, so count them before you build a dashboard panel:

The fix is rarely to raise the limit. It exists to protect the coordinating node's heap; raising it moves the failure from a clean exception to a breaker trip or a long GC pause. Fix the request instead:

• Coarsen the interval to match the panel's pixel width. A chart 1,000 pixels wide gains nothing from 2,016 bars.
• Use auto_date_histogram with a target buckets count to let Elasticsearch pick an interval that fits.
• Reduce the inner size, or move the split to a filter so each panel asks a narrower question.
• Page with composite for exports.
• Pre-aggregate with a transform for panels that are viewed constantly.

If you do raise it, do it knowingly and modestly, and watch the request breaker statistics afterwards:

GET _nodes/stats/breaker?filter_path=nodes.*.name,nodes.*.breakers.request

estimated_size versus limit_size and the tripped counter tell you how close aggregations come to the limit. See JVM heap and circuit breakers.

Date histograms done right

date_histogram is the most common aggregation in observability, and three details matter.

Calendar vs fixed intervals. Use calendar_interval for intervals that follow the calendar (1d, 1w, 1M, 1q, 1y, and minute or hour units): a day is midnight to midnight in the given time zone, so a daylight saving day can be 23 or 25 hours. Calendar intervals accept only a single unit, so 2d is not a calendar interval. Use fixed_interval for fixed-length durations (30s, 5m, 12h, 90m). The old combined interval parameter was deprecated in 7.x in favor of these two; current documentation describes only calendar_interval and fixed_interval, so do not write new code with it.

Time zones. Bucket boundaries are computed in UTC unless you pass time_zone (for example "Asia/Kolkata"). Daily reports for a region should always pass its time zone; otherwise "Monday" starts at 05:30 local time in India.

Empty buckets and bounds. By default buckets with zero documents are omitted at the edges of the data. min_doc_count: 0 with extended_bounds returns empty buckets across the full requested range, which charts need to draw gaps correctly. hard_bounds does the opposite: it limits buckets to a range, which is useful to prevent a stray far-future timestamp from creating thousands of empty buckets.

Caching. The shard request cache caches results of size: 0 requests per shard, and aggregation dashboards are the main beneficiary. It is invalidated when a shard refreshes with changes, so it helps most on indices that are no longer written to, such as yesterday's backing indices of a data stream. Requests that use now in date math are generally not cached by the request cache. Kibana resolves its time picker to absolute timestamps before sending, so its panels can use this cache while hand-written queries with now-15m usually cannot. See search latency for the caches in detail.

High cardinality: memory, global ordinals and alternatives

Aggregating over a field with millions of unique values (user IDs, session IDs, trace IDs) has three costs.

Global ordinals construction. For keyword fields, terms builds global ordinals per shard after each refresh, on first use. On a field with tens of millions of unique values in a shard that refreshes every second, that rebuild cost lands on whichever search arrives first after a refresh. Options:

• Set eager_global_ordinals: true in the mapping, which moves the cost to refresh time. This helps a field aggregated constantly on a search-heavy index, and hurts indexing throughput on a write-heavy one.
• Increase index.refresh_interval (default 1s) on indices where near-real-time visibility does not matter, so ordinals rebuild less often.
• Use the map execution_hint for a terms aggregation whose query matches very few documents; it avoids building ordinals for the whole shard. For most cases, leave the default.

Bucket memory. A terms aggregation over a high-cardinality field with sub-aggregations holds a bucket per term per shard until it can prune. Elasticsearch chooses breadth_first collection mode by default when the field's cardinality is larger than the requested size or unknown, which delays computing sub-aggregations until the top buckets are known. That is usually the right choice; leave it alone unless profiling says otherwise.

Partitioning. When you genuinely need to process every term of a huge field with terms (for example, to find users with an attribute computed by a sub-aggregation), split the key space with partitions:

GET events/_search
{
  "size": 0,
  "aggs": {
    "users": {
      "terms": {
        "field": "user.id",
        "include": { "partition": 0, "num_partitions": 50 },
        "size": 10000
      }
    }
  }
}

Run partitions 0 to 49. Each request handles roughly a 50th of the terms. Composite is usually simpler for the same task; partitioning is useful when you need terms-specific features per partition.

Sampling. When an approximate answer is enough, sample. The sampler and diversified_sampler aggregations restrict sub-aggregations to the top-scoring documents per shard. Recent 8.x releases add random_sampler, which aggregates a random fraction of documents and is designed for fast approximate dashboards over very large data. For exploration, aggregating 1% of a billion events usually gives the same shape as aggregating all of them.

Never aggregate on text. It requires enabling fielddata on the field, which loads an in-heap structure built from the inverted index. Use a keyword multi-field. See mapping design.

Pre-aggregation with transforms

If the same heavy aggregation runs every time someone opens a dashboard, you are recomputing yesterday's answer thousands of times. A transform computes it once and keeps it up to date in a destination index.

A pivot transform groups by fields and aggregates:

PUT _transform/orders-daily-by-customer
{
  "source": {
    "index": ["orders-*"],
    "query": { "range": { "@timestamp": { "gte": "now-400d/d" } } }
  },
  "dest": { "index": "orders-daily-by-customer" },
  "pivot": {
    "group_by": {
      "customer": { "terms": { "field": "customer.id" } },
      "day": { "date_histogram": { "field": "@timestamp", "calendar_interval": "1d" } }
    },
    "aggregations": {
      "revenue": { "sum": { "field": "order.total" } },
      "orders": { "value_count": { "field": "order.id" } }
    }
  },
  "frequency": "5m",
  "sync": {
    "time": { "field": "@timestamp", "delay": "120s" }
  }
}

Preview before creating it, then start it:

POST _transform/_preview
{ "...": "same body as above" }

POST _transform/orders-daily-by-customer/_start

How it works: a transform pages through the source with a composite aggregation built from group_by, and writes one document per bucket into the destination index. With sync.time, it is continuous: on each check (at frequency), it finds which buckets have new data since the last checkpoint and recomputes only those. The delay accounts for ingest lag; documents that arrive with a timestamp older than the last checkpoint minus the delay are missed, so set it longer than your worst normal ingest delay.

There is also a latest transform type, which keeps the most recent document per key (for example, current state per host), useful for "entity" indices.

The arithmetic usually makes the case. Suppose a dashboard panel aggregates 2 billion order documents into daily totals per customer, takes 8 seconds and runs 3,000 times a day. That is 24,000 seconds, about 6.7 hours, of cluster time a day for one panel. A transform recomputes only changed buckets every 5 minutes, and the dashboard queries a destination index orders of magnitude smaller. Measure it on your own data, but the shape of the saving is common.

Monitor transforms with:

GET _transform/orders-daily-by-customer/_stats

Watch state, checkpointing.last.checkpoint and the health field. A transform that fails stops updating silently from the dashboard's point of view.

Downsampling applies to time series data streams and replaces raw metrics in older backing indices with pre-aggregated ones. The older rollup jobs feature is deprecated in favor of downsampling in recent 8.x releases; do not start new designs on rollups. See storage cost optimization.

Failure modes

A top 10 that changes when you refresh the page with no new data. Cause: ties near the boundary combined with terms approximation across shards, or partial results from shard failures. Confirm: check doc_count_error_upper_bound and _shards.failed. Fix: raise shard_size, and treat any failed shards as a separate incident. See thread pools and rejections.

too_many_buckets_exception on a dashboard after the time range widens. Cause: fixed interval times split size exceeds 65,536. Confirm with the multiplication in the table above. Fix: auto interval or coarser interval, smaller split, or a transform.

circuit_breaking_exception with [request] in the message. Cause: nested high-cardinality buckets or cardinality sketches with high thresholds. Confirm with _nodes/stats/breaker on the coordinating node and by removing sub-aggregations one at a time. Fix: lower thresholds, fewer nested buckets, page with composite, or pre-aggregate.

The first query after every refresh is slow. Cause: global ordinals rebuilt lazily on a high-cardinality keyword field. Confirm with the profile API (time in building ordinals) or by comparing the first and second run. Fix: eager_global_ordinals on search-heavy indices, longer refresh interval, or avoid terms on that field.

Exports that time out. Cause: a huge terms size or deep from paging. Fix: composite with after.

A transform's numbers lag or miss data. Cause: delay shorter than real ingest delay, or the transform stopped on an error. Confirm with _transform/<id>/_stats. Fix: increase delay, fix the error, and if data was missed, reset and restart the transform for the affected range.

Runbook: an aggregation is slow or failing

1. Get the exact request. From Kibana's Inspect panel, or from the slow log. Note size: 0, the time range and the nesting.
2. Count buckets. Multiply interval count by every nested size. If it is near 65,536, the request shape is the problem.
3. Check accuracy fields. For each terms, read doc_count_error_upper_bound and sum_other_doc_count. Check _shards.failed is 0.
4. Profile it. Add "profile": true and look for aggregation time per shard and time in building ordinals.
5. Check the breaker. GET _nodes/stats/breaker on the node that coordinated it. Rising request estimates or trips identify memory pressure.
6. Simplify in order. Coarsen the interval, lower inner sizes, lower precision_threshold in nested cardinality, remove sub-aggregations that are not displayed.
7. Change the approach. Exports go to composite paging. Repeated panels over large history go to a transform or, for metrics, downsampling.
8. Confirm. Re-run with profiling, compare time and breaker usage, and check the error bound is acceptable.

Key takeaways

• terms is approximate across shards. Default shard_size is size x 1.5 + 10. Read doc_count_error_upper_bound before trusting a ranking.
• Composite is exact and bounded; use it to page through all buckets, never a huge terms size.
• Cardinality memory is about precision_threshold x 8 bytes per instance, default 3,000, maximum 40,000. Nested sketches multiply that by the bucket count.
• search.max_buckets defaults to 65,536 and protects the coordinator. Fix the request, not the limit.
• Use calendar_interval or fixed_interval, always pass time_zone for human reports, and use hard_bounds to block runaway ranges.
• Heavy aggregations that repeat belong in a continuous transform, and old metrics belong in downsampled indices.

How Sparced helps

Aggregation trouble usually shows up in the cluster as heap pressure and breaker trips, and that is what Sparced reads. Its one read-only pass includes the circuit breaker statistics and node heap for every node, so a request breaker that is tripping, or a coordinating node past 85% heap, is reported with the node named and the breaker's own counters as evidence. It also reports search thread pool rejections, which explain partial aggregation results. Sparced does not read your queries or dashboards, so it will not tell you which panel is responsible; it tells you which node is under pressure and from which breaker, which is where the runbook above starts.