[
https://issues.apache.org/jira/browse/SPARK-58399?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18099622#comment-18099622
]
ZequnLin commented on SPARK-58399:
----------------------------------
Industry precedent: BigQuery (GoogleSQL) already supports an array-input
aggregate of this style -- {{ARRAY_CONCAT_AGG}} concatenates array-typed values
across rows (distinct is then applied on top). PostgreSQL has no dedicated
function for this and users fall back to {{array_agg(DISTINCT v)}} over
{{unnest(arr)}}, which is the SQL analogue of the explode + {{collect_set}}
workaround. So there is prior art for a first-class array-input aggregate, and
{{collect_union}} gives Spark the bounded-buffer, distinct form of it.
> 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]
