apache / apache/beam

Add alternate Sorting transform as an implementation of CombineFn

Open
#20,684 0 comments 0 reactions 0 assignees View on GitHub
extensions improvement java P3 sorter
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

My team has been using the [SortValues](https://github.com/apache/beam/blob/master/sdks/java/extensions/sorter/src/main/java/org/apache/beam/sdk/extensions/sorter/SortValues.java) transform in `extensions-java-sorter` to sort pre-grouped values by a secondary sorter key. However, for large key groups, we've run into many OOM issues and have to increase disk size quite a bit to accommodate the larger key groups spilling to disk, even if there are only a few large key groups and most fit in memory.

I drafted a new iteration of a Sorter that's a distributed merge-sort implemented as a `CombineFn`: each Accumulator maintains an always-sorted list of elements, and those Accumulators can be merged simply by zipping their lists together. This has the extra advantage that `extractOutput` can be lazily evaluated as a merging Iterator rather than as a fully materialized list. I also observed that this implementation is able to scale more effectively than the old SortValues, and for several use cases where `SortValues` ran OOM, the CombineFn-based implementation was able to complete using only the default Dataflow disk specs.

Finally, from an API perspective, I think it's a little easier to use, because the user doesn't have to extract the sortKey out into the PCollection itself, but instead provide a function mapping each element type T to its sort key K, which will be evaluated inside the combiner. So I think in that sense it's more intuitive and similar to a Comparator-style sort.

Imported from Jira [BEAM-11048](https://issues.apache.org/jira/browse/BEAM-11048). Original Jira may contain additional context.
Reported by: clairemcginty.

Contributor guide

Open the contributing guide

Research direction

Start with extensions-java-sorter and the existing SortValues.java transform, then review the CombineFn API described in the issue. Compare the proposed alternate sorter against the existing behavior; done means it supports a sort-key mapping, merges accumulators, lazily produces output, and avoids the reported large-group OOM cases.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
data-engineering
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.