[
https://issues.apache.org/jira/browse/SPARK-58399?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18099621#comment-18099621
]
ZequnLin commented on SPARK-58399:
----------------------------------
Note on why {{collect_set}} is not a substitute for
{{array_distinct(flatten(collect_list(...)))}}:
The distinct-union of an array column can be written two ways today, but both
have a real cost:
* {{array_distinct(flatten(collect_list(col)))}} keeps one GROUP BY and needs
no join, but the aggregation buffer holds every row's whole array before
de-duplicating, so it can OOM on skewed grouping keys.
* {{collect_set(element)}} over {{explode(col)}} bounds the buffer to the
distinct elements, but it is no longer a plain aggregate: each array column
must be exploded and grouped separately, then the per-column results joined
back on the grouping keys. For a query that needs the distinct union of N array
columns, that is N explodes + N joins -- the join cost is paid purely to work
around the lack of an array-input aggregate.
{{collect_union(col)}} gives the bounded-buffer behavior of the
explode+{{collect_set}} form while remaining an ordinary aggregate, so multiple
array columns can be aggregated side by side in a single GROUP BY with no
explode and no join.
> Add `collect_union` aggregate function
> --------------------------------------
>
> Key: SPARK-58399
> URL: https://issues.apache.org/jira/browse/SPARK-58399
> Project: Spark
> Issue Type: Improvement
> Components: PySpark, SQL
> Affects Versions: 4.3.0
> Reporter: ZequnLin
> Priority: Major
>
> Add a new aggregate function {{collect_union}} that takes an array-typed
> column and returns the distinct union of the elements of the arrays across
> rows.
> {code}
> collect_union(col: array<T>) : array<T>
> {code}
> It is equivalent to {{array_distinct(flatten(collect_list(col)))}}, but the
> aggregation buffer holds only the distinct elements (a set), so its size is
> bounded by the element universe rather than by the number of input rows. The
> {{array_distinct(flatten(collect_list(...)))}} workaround buffers every row's
> whole array before de-duplicating, which can OOM on skewed grouping keys;
> {{collect_union}} de-duplicates during aggregation, keeping the buffer
> bounded.
> There is currently no built-in aggregate that unions the elements of an array
> column across rows into a single distinct array.
> Semantics:
> * NULL input arrays are skipped; NULL elements inside a non-null array are
> skipped (following collect_set semantics).
> * Result element type is the input array's element type.
> * Element order in the result is unspecified (as with collect_set /
> collect_list).
> The function is exposed in SQL, the Scala DataFrame API, and PySpark (classic
> + Spark Connect). Spark Connect requires no protocol change.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
