[
https://issues.apache.org/jira/browse/SPARK-39931?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Enrico Minack updated SPARK-39931:
----------------------------------
Description:
Calling `DataFrame.groupby(...).applyInPandas(...)` for very small groups in
PySpark is very slow. The reason is that for each group, PySpark creates a
Pandas DataFrame and calls into the Python code. For very small groups, the
overhead is huge, for large groups, it is smaller.
Here is a benchmarks (seconds to groupBy(...).applyInPandas(...) 10m rows):
||groupSize||Scala||pyspark.sql||pyspark.pandas||
|1024|8.9|20.9|7.8|
|512|9.4|31.8|9.8|
|256|9.3|47.0|20.2|
|128|9.5|83.3|48.8|
|64|9.5|137.8|91.9|
|32|9.6|263.6|207.3|
|16|9.6|525.9|261.5|
|8|9.5|1,043|663.0|
|4|9.8|2,073|1,168|
|2|10.4|4,132|2,456|
|1|11.3|8,162|4,642|
*Idea to overcome this* is to call into Python side with a Pandas DataFrame
that contains potentially multiple groups, then perform a Pandas
DataFrame.groupBy(...).apply(...). With large groups, that Panadas DataFrame
has all rows of single group, with small groups it contains many groups. This
should improve efficiency.
I have prepared a PoC to benchmark that idea but am struggling to massage the
internal rows before sending them to Python. I have the key and the grouped
values of each row as {{InternalRow}}s and want to turn them into a single
{{InternalRow}} in such a way that in Python I can easily group by the key and
get the same grouped values. For this, I think adding the key as a single
column (struct) while leaving the grouped values as is should work best. But I
have not found a way to get this working.
was:
Calling `DataFrame.groupby(...).applyInPandas(...)` for very small groups in
PySpark is very slow. The reason is that for each group, PySpark creates a
Pandas DataFrame and calls into the Python code. For very small groups, the
overhead is huge, for large groups, it is smaller.
Here is a benchmarks (seconds to groupBy(...).applyInPandas(...) 10m rows):
||groupSize||Scala||pyspark.sql||pyspark.pandas||
|1024|8.9|20.9|7.8|
|512|9.4|31.8|9.8|
|256|9.3|47.0|20.2|
|128|9.5|83.3|48.8|
|64|9.5|137.8|91.9|
|32|9.6|263.6|207.3|
|16|9.6|525.9|261.5|
|8|9.5|1,043|663.0|
|4|9.8|2,073|1,168|
|2|10.4|4,132|2,456|
|1|11.3|8,162|4,642|
*Idea to overcome this* is to call into Python side with a Pandas DataFrame
that contains potentially multiple groups, then perform a Pandas
DataFrame.groupBy(...).apply(...). With large groups, that Panadas DataFrame
has all rows of single group, with small groups it contains many groups. This
should improve efficiency.
I have prepared a PoC to benchmark that idea but am struggling to massage the
internal Rows before sending them to Python. The idea is to turn the
{{Dataset[T]}} into {{{}Dataset[(K, T)]{}}}, create a bespoke
{{GroupedIterator}} that splits the Dataset into chunks that contain complete
groups, sends them to Python, which the groups by key and apply the udf on the
values. The first bit seems to be the hard bit:
{{{}sql/core/src/main/scala/org/apache/spark/sql/execution/python/FlatMapGroupsInPandasExec.scala{}}}:
{code:scala}
val inputRDD = ProjectExec(Seq(
Alias(CreateNamedStruct(groupingAttributes.flatMap(a => Literal(a.name)
:: a :: Nil)), "key")(),
Alias(CreateNamedStruct(dedupAttributes.flatMap(a => Literal(a.name) ::
a :: Nil)), "val")()
), child).execute()
{code}
> Improve performance of applyInPandas for very small groups
> ----------------------------------------------------------
>
> Key: SPARK-39931
> URL: https://issues.apache.org/jira/browse/SPARK-39931
> Project: Spark
> Issue Type: Improvement
> Components: PySpark
> Affects Versions: 3.4.0
> Reporter: Enrico Minack
> Priority: Major
>
> Calling `DataFrame.groupby(...).applyInPandas(...)` for very small groups in
> PySpark is very slow. The reason is that for each group, PySpark creates a
> Pandas DataFrame and calls into the Python code. For very small groups, the
> overhead is huge, for large groups, it is smaller.
> Here is a benchmarks (seconds to groupBy(...).applyInPandas(...) 10m rows):
> ||groupSize||Scala||pyspark.sql||pyspark.pandas||
> |1024|8.9|20.9|7.8|
> |512|9.4|31.8|9.8|
> |256|9.3|47.0|20.2|
> |128|9.5|83.3|48.8|
> |64|9.5|137.8|91.9|
> |32|9.6|263.6|207.3|
> |16|9.6|525.9|261.5|
> |8|9.5|1,043|663.0|
> |4|9.8|2,073|1,168|
> |2|10.4|4,132|2,456|
> |1|11.3|8,162|4,642|
> *Idea to overcome this* is to call into Python side with a Pandas DataFrame
> that contains potentially multiple groups, then perform a Pandas
> DataFrame.groupBy(...).apply(...). With large groups, that Panadas DataFrame
> has all rows of single group, with small groups it contains many groups. This
> should improve efficiency.
> I have prepared a PoC to benchmark that idea but am struggling to massage the
> internal rows before sending them to Python. I have the key and the grouped
> values of each row as {{InternalRow}}s and want to turn them into a single
> {{InternalRow}} in such a way that in Python I can easily group by the key
> and get the same grouped values. For this, I think adding the key as a single
> column (struct) while leaving the grouped values as is should work best. But
> I have not found a way to get this working.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
