Add partial reduce nodes for reducing intermediate aggregation results

[hackerwin7]

Currently Elasticsearch in aggregation search is to scatter the requests and then gather these sub-requests into coordinator. then coordinator reduce these requests in heap.

However, if all the sub-request seems huge in memory, coordinator may have a horrible huge cost of heap. then may cause to high latency GC or OOM.

Can we import new style for this: a new role for coordinator-agg node: data node do not return a complete sub-request result but a series of blocks of the sub-request continuously and coordinator agg node receive many blocks in parallel from multiple data-node. and coordinator agg node can pre-aggregate these blocks early and in parallel way and continuously.

When the coordinator agg node receive every data-node of last series of blocks, coordinator agg node can return this final aggregate result to user. by aggregate the smaller blocks in coordinator agg node continuously and in parallel, the memory cost and control is more smoothly compare to current style.

In the end, we may have two kinds of coordinator node:

1. original common coordinator node is good at common query search cases with high QPS
2. stream aggregation coordinator node is good at specific super huge memory cost of aggregation search with low QPS due to cost many threads to continuously aggregate blocks in parallel

[elasticmachine]

Pinging @elastic/es-analytics-geo (:Analytics/Aggregations)

[imotov]

@hackerwin7 I am not totally sure I understand your proposal. Are you suggesting to introduce a new role for nodes. So we could have something like coordinating-node -> partial-reduce-node -> data-node, and a coordinating node would distribute a request across several randomly choose partial-reduce-node, which will in turn talk to data-nodes? Is this the proposal? I see how this might reduce the maximum memory required on a single node, but I don't really see overall cost on the cluster. Could you clarify?

[hackerwin7]

@hackerwin7 I am not totally sure I understand your proposal. Are you suggesting to introduce a new role for nodes. So we could have something like coordinating-node -> partial-reduce-node -> data-node, and a coordinating node would distribute a request across several randomly choose partial-reduce-node, which will in turn talk to data-nodes? Is this the proposal? I see how this might reduce the maximum memory required on a single node, but I don't really see overall cost on the cluster. Could you clarify?

@imotov
Yes, the partial-reduce-node is a role to share the responsibility to receive the partial merge to compute and prune the results, the compute flow of search may be like this:

coordinate-node -> data-node -> reduce node (shuffle or reroute the shard results into reduce node to merge partially and in parallel, this is phase-1 merge, it is distributed and in parallel) -> coordinate-node (final reduce from the reduce node, phase-2 reduce).

on the other hand, this feature is expected to support the search aggregations in streaming which allow the partial merge aggregations results can return to User early.

I think this proposal is good at this case: the search shard count is huge, such as search many time-based indices with a huge amount of shards; the search buckets count is extremely huge.

so, I want to add a intermediate phase for aggregation (shuffle phase) to implement this:

• distribute merge compute to multiple nodes; (1) this can merge huge results in parallel to speed up merge compute (2) reduce the memory cost in single ES node
• all of this style of aggregation compute is streaming, it can return partial aggregations early to User and User can call like search after to return the next streaming of aggregation result

[nik9000]

Traditionally we've suggested folks that are concerned about running out of memory on the coordinating node to use dedicated coordinating nodes so at least in that case the destruction caused by the OOM is contained. We've started towards a solution to this with the batched_reduce request parameter. I believe our current plan is to trigger partial reductions based on memory usage. Mostly because that is a less invasive change than what you are talking about. We do have the ccs_minimize_roundtrips which sort of plays here in that it makes the remote clusters to partialy reduce the aggregations before forwarding them back to the coordinating node.

[maosuhan]

@nik9000 I think the proposal raised from @hackerwin7 is very useful in our case. In our case, user would query large range of date from time-based indices and it requires to scan huge amount of shards and sometimes a single shard response is very large. In that scenario the query would be very slow and OOM sometimes. Though we have many improvement in circuit breaker(#62223, #67431) and partial reduce speed (#53216) and some open issues(#67474, #67478), I think they are not enough. They just try to early drop the request or try to improve single node computation which is still slow in big query path.

What we need is to run big query fast and stable.
My proposal would be

1. Early detect the heavy query in the first place. I think it is already doable after many PRs related to it.
2. If a big request is detected, the request fall back to distributed computation.
3. The query path would be initial coordinate-node -> distributed coordinate-node(parallelize mapper, each visit subset of shards) -> data-node -> distributed coordinate-node(parallelize mapper, local aggregation and send the same bucket key to the same reducer) -> distributed coordinate-node(parallelize reducer, do partial reduce) -> initial coordinate-node(lightweight final reduce, because it do not need to merge buckets)

If a new role of node is more expansive implementation, we could still use existing client node to cooperate this. If we want to speed up the big query performance, we just increase the client nodes.