[jira] [Work logged] (BEAM-11048) Add alternate Sorting transform as an implementation of CombineFn

["ASF GitHub Bot (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/BEAM-11048?focusedWorklogId=511510&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-511510 ]

ASF GitHub Bot logged work on BEAM-11048:
-----------------------------------------

                Author: ASF GitHub Bot
            Created on: 13/Nov/20 19:32
            Start Date: 13/Nov/20 19:32
    Worklog Time Spent: 10m 
      Work Description: aaltay commented on pull request #13059:
URL: https://github.com/apache/beam/pull/13059#issuecomment-726990914


   R: @tysonjh 


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Issue Time Tracking
-------------------

    Worklog Id:     (was: 511510)
    Time Spent: 20m  (was: 10m)

> Add alternate Sorting transform as an implementation of CombineFn
> -----------------------------------------------------------------
>
>                 Key: BEAM-11048
>                 URL: https://issues.apache.org/jira/browse/BEAM-11048
>             Project: Beam
>          Issue Type: Improvement
>          Components: extensions-java-sorter
>            Reporter: Claire McGinty
>            Priority: P2
>              Labels: Clarified
>          Time Spent: 20m
>  Remaining Estimate: 0h
>
> 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.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)