# High-Cardinality Aggregation Alternate Approaches (2 million buckets)

**URL:** <https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814>\
**Category:** Elasticsearch\
**Created:** [March 9, 2020, 9:56pm UTC](https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814 "2020-03-09T21:56:37Z")\
**Posts on this page:** 12\
**Page:** 1

<div class="post-metadata">

**Author:** ![khickey](https://avatars.discourse-cdn.com/v4/letter/k/ecb155/32.png) [@khickey](https://discuss.elastic.co/u/khickey)\
**Post date:** [March 9, 2020, 9:56pm UTC](https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814/1 "2020-03-09T21:56:37Z")

</div>

**Requirement** : _calculate a sum aggregate on a high-cardinality field of 2M unique values, order the results by the sum, and return the top 10_ (or pages of 10 each).

We are moving some of our materialized-view based data summarization queries from Postgres to Elasticsearch, and came across a need to aggregate a sum of on a high-cardinality field; basically think email addresses. We have around 2 million unique values for this field.

In our version 7.1 cluster, we're hit by the default limit of 10000 buckets based on the [cluster setting](https://www.elastic.co/guide/en/elasticsearch/reference/current/cluster-update-settings.html) for `search.max_bucket`. We could of course update this to something higher, but getting to 2 million would probably put us in danger of bringing down the cluster. There is the possibility of many concurrent queries too.

We cannot do a `composite` aggregation and paginate the buckets here because our requirement is to return the results ordered by sum - highest values first. So we have a `bucket_sort` pipeline aggregation alongside the `composite` as a means to do this, but hit a limit when the `size` value in the `composite` is over 10k.

I've seen some posts that generally say: _"you can't do this in Elasticsearch. Instead pre-aggregate the data and dump it into a new entity-centric index"_.

_**Is that really the only approach when you come up against trying to post-process (e.g. sort) aggregated data that has greater than 10k (or some reasonable limit) of unique buckets?**_

---

<div class="post-metadata">

**Author:** ![khickey](https://avatars.discourse-cdn.com/v4/letter/k/ecb155/32.png) [@khickey](https://discuss.elastic.co/u/khickey)\
**Post date:** [March 13, 2020, 12:16pm UTC](https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814/2 "2020-03-13T12:16:20Z")

</div>

Bump + further info:

So we headed down the path of an entity-centric index. Basically to summarize metrics on a per entity (e.g. user) basis. Even found the new (since 7.2) [Transform API](https://www.elastic.co/guide/en/elasticsearch/reference/7.x/put-transform.html), which adds key missing features to the Reindex API and allows you to create/populate a target index as the aggregation of a source index.

BUT then ... we realized that our use case requires the flexibility of dynamically choosing dimensional filters on which the aggregation is to be based. And an entity-centric index effectively requires you to answer _one_ question (or small related problem space of questions) that assume a _ **fixed** _ set of dimensional filters. For example you could do any ONE of the below, but not have flexibility to do any arbitrary combination of them on the fly.

Fixing time:

- summarize users _only in the last month_

Fixing product category:

- summarize users _only buying from these product categories_

Fixing location:

- summarize users _only from these locations_

Fixing amount:

- summarize transactions _only above a certain dollar amount_

Dynamically querying for these aggregated values works on a small scale and uses query-time filter values in the `"query": { "bool": { "filter": {...} } }` section, before the aggregation is applied, so we get potentially different count/sum aggregation values for each user entity, depending on the dimensional filter selection (since the set of documents being aggregated is different for each filter applied).

However the above works only up to a certain amount of buckets -- **if** we need to provide a sort order to those buckets -- which should reasonably not go above 10k or 50k.

To **store** aggregated values (e.g. purchase count, total expenditure, etc.) in an entity-centric index that can be queried fast later on, we'd need to establish that fixed dimensional filter ahead of time which totally undermines the ability to query the data from different dimensions.

This appears to be an intractable use case for Elasticsearch. A hard limit of what it can do and a question it can't answer in one go. A dead end. Please let me know if that is NOT the case and there are ways to accommodate this use case.

---

<div class="post-metadata">

**Author:** ![Mark\_Harwood](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mark_harwood/32/10538_2.png) [@Mark\_Harwood](https://discuss.elastic.co/u/Mark_Harwood)\
**Post date:** [March 13, 2020, 12:22pm UTC](https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814/3 "2020-03-13T12:22:04Z")

</div>

> A hard limit of what it can do and a question it can't answer in one go. A dead end.

Distributed data makes this hard. But we have some different approaches. You might find [this wizard](https://plnkr.co/edit/eZr7r3KZW02AxNKAHCQa?p=preview&preview) helpful when choosing which is appropriate.

I think term partitioning might be useful in this case.

 ![Plunker_-_Simple_AngularJS_decision_tree_wizard](https://us1.discourse-cdn.com/elastic/original/3X/7/1/714a21862489c0dbc56ad2aa7083e54d8a34c0ec.png)

---

<div class="post-metadata">

**Author:** ![khickey](https://avatars.discourse-cdn.com/v4/letter/k/ecb155/32.png) [@khickey](https://discuss.elastic.co/u/khickey)\
**Post date:** [March 13, 2020, 3:30pm UTC](https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814/4 "2020-03-13T15:30:03Z")

</div>

Ha, it seems you get these types of questions a lot -- that's a nice little javascript decision tree!

Answering the decision tree, it does indeed lead to trying [partitioning](https://www.elastic.co/guide/en/elasticsearch/reference/7.1/search-aggregations-bucket-terms-aggregation.html#_filtering_values_with_partitions).

So I tried a sample query like the below. Did I get that right?

Problem: The sorts _within_ each partition of buckets is correct, but the global sort across all buckets is not (i.e. across subsequent partitions). Is this what you meant by "client-side stitching" (maybe you didn't say that, I thought I read that somewhere)? Is this just picking disjoint sets of 50 terms each, and then on a per-partition basis, asking the shards to provide a sum agg of each term in the given set, then ordering only among the 50?

i.e. will I have to iterate through _all_ partitions, and then resort the entire high-cardinality set of buckets on the client side to get an accurate sort order of buckets?

_Assume this is a subset of data only just under 3000 unique users in it. 60 partitions, at 50 buckets per partition = 3000._

```auto
GET my_data/_search
{
  "size": 0,
  "aggs": {
    "group_by_email": {
      "terms": {
        "field": "email_address",
        "size": 50,
        "order": { "sum_purchase_price": "desc" },
        "include": {
          "partition": 0,
          "num_partitions": 60
        }
      },
      "aggs": {
        "sum_purchase_price": {
          "sum": {
            "field": "purchase_price"
          }
        }
      }
    }
  }
}

```

If I have to run through all partitions to get all buckets in order to calculate a global sort and return the top-most buckets, how is this different from the below composite query, paging through all with `after` key, and then doing a sort on the client side?

Does partitioning support sending parallel requests for multiple partitions at once, where composite cannot (due to paging)? That is, will requests for the same partition always return the same buckets if the docs have not changed?

Still, either approach is going to be too slow for expected response times and ask the client software to manage and sort 2M datapoints in this case.

Similar `composite`

```auto
GET my_data/_search
{
  "size": 0, 
  "aggs": {
    "group_by_email": {
      "composite": {
        "size": 50,
        "sources": [
          {
            "email": {
              "terms": {
                "field": "email_address"
              }
            }
          }
        ]
      },
      "aggs": {
        "sum_purchase_price": {
          "sum": {
            "field": "purchase_price"
          }
        }
      }
    }
  }
}

```

---

<div class="post-metadata">

**Author:** ![Mark\_Harwood](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mark_harwood/32/10538_2.png) [@Mark\_Harwood](https://discuss.elastic.co/u/Mark_Harwood)\
**Post date:** [March 13, 2020, 3:44pm UTC](https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814/5 "2020-03-13T15:44:00Z")

</div>

> [@khickey](#):
>
> Ha, it seems you get these types of questions a lot

You guessed it. 🙂 I always say aggregations are a simple request like the "carry a chicken, fox and grain" problem but logistics of carrying it out are made much more complex by a constraint - like a small boat (or in our case limited memory for ferrying data across the network).

> [@khickey](#):
>
> how is this different from the below composite query, paging through all with `after` key, and then doing a sort on the client side?

Composite doesn't prune uninteresting results and returns _all_ values sorted by a grouping key whereas with partitioning you _can_ sort by some other order like a sum or max of a different field organised under a common grouping key. That way you can avoid dragging all values to the client.

> [@khickey](#):
>
> will requests for the same partition always return the same buckets if the docs have not changed?

Yes. it is deterministic.

> [@khickey](#):
>
> Assume this is a subset of data only just under 3000 unique users in it. 60 partitions,

With only 3k users that's manageable in one `terms` agg without partitioning, no? It would be below the default 10k buckets-per-agg-request limit? Only having 50 terms in a partition is needlessly low.

---

<div class="post-metadata">

**Author:** ![khickey](https://avatars.discourse-cdn.com/v4/letter/k/ecb155/32.png) [@khickey](https://discuss.elastic.co/u/khickey)\
**Post date:** [March 13, 2020, 4:54pm UTC](https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814/6 "2020-03-13T16:54:08Z")

</div>

> [@Mark\_Harwood](#):
>
> With only 3k users that's manageable in one `terms` agg without partitioning, no?

Oh yes, absolutely. I was just testing the **sort** behavior of this query, and didn't want to use a full dataset. So my query/test was just on a small subset, to see if sort was global across all partitions (which it doesn't seem to be).

The real query would need to look like:

```auto
GET my_data/_search
{
  "size": 0,
  "aggs": {
    "group_by_email": {
      "terms": {
        "field": "email_address",
        "size": 1000,
        "order": { "sum_purchase_price": "desc" },
        "include": {
          "partition": 0,
          "num_partitions": 2000
        }
      },
      "aggs": {
        "sum_purchase_price": {
          "sum": {
            "field": "purchase_price"
          }
        }
      }
    }
  }
}

```

... To traverse all 2,000,000 buckets to know I _definitely_ got the top 10 e.g. buckets out of 2M.

So please confirm this is how this would play out in this case:

In order to do a sum-agg on each bucket, and then pull out only the top 10, I'd need to:

1. consider the default `10000` limit for `search.max_buckets`
2. pick a reasonable bucket size that is under that for each request: `1000` here
3. do the cardinality agg to count unique buckets, and divide by `1000` to get number of partitions. Assume just under 2M unique yields about `2000` partitions
4. send 2000 requests to ES to gather all the bucket sums
5. run a client-side sort on those 2000 sum values
6. pick out the top 10

- ... then they want to see the next 10

1. So either keep all this data stored on the client-side
2. Or repeat the whole lengthy process again for the next 10

Is that how it would need to play out? I could see this maybe being okay for backoffice analytics... but not in an online dashboard or site that people could use to gather quick info.

---

<div class="post-metadata">

**Author:** ![Mark\_Harwood](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mark_harwood/32/10538_2.png) [@Mark\_Harwood](https://discuss.elastic.co/u/Mark_Harwood)\
**Post date:** [March 13, 2020, 5:06pm UTC](https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814/7 "2020-03-13T17:06:57Z")

</div>

Deep, accurate, fast pagination is not practical for data distributed in this way.  
We can't beat physics so you'd need to bring related data together - this could be either:

1. Same-document (using transforms to create various entity-centric indices) or
2. Same-shard (use the email address as a routing key to ensure all data for an email address ends up on the same shard or just store all data in one shard).

There's pros and cons to each of these. Option 2 means you can rely on accurate and fast aggregation but can't paginate deeply in sum order. Option 1 has all the data pre-summed so you can just sort by that field and paginate deeply

---

<div class="post-metadata">

**Author:** ![khickey](https://avatars.discourse-cdn.com/v4/letter/k/ecb155/32.png) [@khickey](https://discuss.elastic.co/u/khickey)\
**Post date:** [March 13, 2020, 8:57pm UTC](https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814/8 "2020-03-13T20:57:44Z")

</div>

> [@Mark\_Harwood](#):
>
> whereas with partitioning you _can_ sort by some other order like a sum or max of a different field organised under a common grouping key. **That way you can avoid dragging all values to the client.**

That last part. I'm not sure if I understood this correctly. My tests seem to indicate that getting a sort across all 2M buckets would require dragging all values to the client. Fetching the first partition of 50 buckets did sort, but _only among those 50 buckets_. The next partition of 50 may yield a bucket with a greater sum than the greatest in the first partition.

> [@Mark\_Harwood](#):
>
> Option 2 means you can rely on accurate and fast aggregation but can't paginate deeply in sum order.

Can you clarify a bit more how the routing key helps? I've come across a few docs that suggested terms on its own (without partitions and without composite) may yield inaccurate sums/counts if size is too small. But in the cases above, the sum aggs should sum all documents for each term, across all shards, correct? Would the routing key simply speed up the agg latency, since partial sums from each shard wouldn't have to be brought back to the coordinating node to complete the term sum?

---

<div class="post-metadata">

**Author:** ![Mark\_Harwood](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mark_harwood/32/10538_2.png) [@Mark\_Harwood](https://discuss.elastic.co/u/Mark_Harwood)\
**Post date:** [March 13, 2020, 10:54pm UTC](https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814/9 "2020-03-13T22:54:03Z")

</div>

Here’s the way to think of it. In the default scenario of randomly distributed docs and a request for the top N terms each shard will return more than N terms of what they individually consider to be the best answers. These are summed and then trimmed on the coordinating node for return to the client. The problem is one or more of the shards may not have returned some stats for a term that was low-scoring on that shard but was ultimately returned because it was high scoring on another shard.  
This can be solved (up to a point) by getting each shard to return significantly more than N results in the hope that this helps fill in some of the missing data for each term. Selecting terms on most-popular or highest spend is typically accurate but long-tail items like least popularity or lowest spend can be wildly inaccurate.

By using custom data routing each shard is guaranteed to have complete visibility of all stats for terms held on that shard. Accuracy is guaranteed.

---

<div class="post-metadata">

**Author:** ![khickey](https://avatars.discourse-cdn.com/v4/letter/k/ecb155/32.png) [@khickey](https://discuss.elastic.co/u/khickey)\
**Post date:** [March 14, 2020, 9:57pm UTC](https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814/10 "2020-03-14T21:57:10Z")

</div>

What I takeaway from your explanation and a [re-read of the relevant doc warnings](https://www.elastic.co/guide/en/elasticsearch/reference/7.1/search-aggregations-bucket-terms-aggregation.html#search-aggregations-bucket-terms-aggregation-approximate-counts) is that a "top N" terms query is dangerous due to potentially missing/underrepresented term aggs sent back by the shards, leading to inaccurate final sums/counts/aggs by the coordinator, which in turn could lead to incorrect sorting to choose the _top_ N.

I understand inaccuracies due to approximation to be the case for `A)` below, but NOT for `B)` or `C)`. _ **Can you confirm?** _

- `A)` A `terms` agg where `shard_size < term cardinality`, and _no routing key for the term_ (... `shard_size defaluts to floor(1.5*size)+10`)
- `B)` A `composite` agg that uses a `terms` source
- `C)` A `terms` agg that uses `include.partition`, `include.num_partitions`

Some sample tests across a small 3-node, 3-shard cluster seemed to show `B)` and `C)` did not suffer from the same inaccuracy problem with sum that `A)` did.

Sorting to get "top N" with `B)` or `C)` is a different challenge. To get an accurate "top N" buckets, it seems I'd need to pull _all_ terms' buckets to the client side and compare them to get a global sort. To do those:

- `B)` _sequentially_ paginate all buckets one page at a time using the `after` key, pulling every one back to the client-side, and doing a global sort of the buckets' sums to get "top N"
- `C)` requesting _every_ partition from the cluster, each being a disjoint set of terms by way of a hashing the term against the partition numbers, accumulating those on the client-side, and doing a global sort of the buckets' sums to get "top N"

**So to solve this use case with `B)` (`composite`) or `C)` (`partitions`), I'll need to siphon all buckets to the client first to do the final sorting pass, yes?**

Now back to `A)`. If documents with the same term are all sent to the same shard by way of a routing key, it does seem to solve the potentially inaccurate sums. _**Does a term-based routing key at index time also allow for "top N" queries to be answered without looking at all the buckets (server-side or client-side)?**_

Seems like it would because every shard could confidently send its top `N` candidates forward to the coordinator and let it sort out the final `N`. **Am I missing some subtle corner case where a top `N` candidate wouldn't make it to the coordinator?**

I'd just be left with a deep-pagination problem. e.g. if someone wanted page 100,000, to look at items 1,000,001-1,000,010 ... I'd need the server to process a terms agg with `"size": 1000010` (ouch). Or fall back to `B)` or `C)` to make the client suffer instead of the server. Maybe a reasonable compromise here is to limit deep-pagination, and have that pagination stop at the 1000th page (10,000th item for pages of 10) to keep the 10k `search.max_buckets` limit.

Sorry to ramble through this. Any inaccuracies in these assessments?

---

<div class="post-metadata">

**Author:** ![Mark\_Harwood](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mark_harwood/32/10538_2.png) [@Mark\_Harwood](https://discuss.elastic.co/u/Mark_Harwood)\
**Post date:** [March 15, 2020, 9:11am UTC](https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814/11 "2020-03-15T09:11:49Z")

</div>

Good write up of the options.  
The thing to consider with B vs C are the different distributed counting approaches.

When there are too many unique terms the B and C strategies are used to try get each shard to consider the same subset of the overall term space.  
With B (composite) each shard is focusing in on gathering stats for the subset of terms that occur immediately after the ‘after’ value provided when they are sorted by the term value. This works because each shard can independently agree that the term value X comes before Y and have the same ranking logic for term selection. What they can’t do is sort by a sum eg all of X’s sales so the poor client has to page through every term alphabetically.  
Option C (partitioning) makes each shard consider the same subset of terms because they can all independently agree that hash of term X, modulo N gives a partition number of 3 and that might be the partition we are considering. This is just like option A but we found a way to cut the index into arbitrarily small subdivisions and can afford to do global sorts by sums of another field knowing that we are more likely to be accurate. The sorting by sum means you may be able to avoid returning a large percentage of the terms to the client for each partition.  
Bear in mind option A with a suitable shard\_size frequently works ok - you just need to pay attention if there are non-zero error margin values returned in results.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [April 12, 2020, 9:12am UTC](https://discuss.elastic.co/t/high-cardinality-aggregation-alternate-approaches-2-million-buckets/222814/12 "2020-04-12T09:12:02Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.