[
https://issues.apache.org/jira/browse/SPARK-60650?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-60650:
-----------------------------------
Labels: pull-request-available (was: )
> pandas-on-Spark named size aggregation is mapped to SQL size and fails on an
> integer column
> -------------------------------------------------------------------------------------------
>
> Key: SPARK-60650
> URL: https://issues.apache.org/jira/browse/SPARK-60650
> Project: Spark
> Issue Type: Bug
> Components: Pandas API on Spark, PySpark
> Affects Versions: 4.2.0
> Environment: Platform: Ubuntu 24.04 Linux-6.17.0-29-generic x86_64
> Python: 3.11.17
> pyspark 4.2.0
> findspark 2.0.1
> pandas 3.0.6
> numpy 2.4.6
> Reporter: asddfl
> Priority: Critical
> Labels: pull-request-available
>
> When using the pandas API on Spark (pyspark.pandas), the named aggregation
> agg(c1=('c0', 'size')) fails for an integer column. Pandas returns the group
> size, but pandas-on-Spark generates a SQL size(c0) expression. Spark SQL
> interprets size as the array/map length function and rejects the INT input.
> The example uses a single row and an explicitly int32 column. ANSI mode is
> disabled.
> {code:python}
> import pandas as pd
> from pyspark.sql import SparkSession
> import pyspark.pandas as ps
> t0 = pd.DataFrame({'c0': [1]}, dtype='int32')
> print("Pandas:", flush=True)
> result = t0.groupby('c0', as_index=False).agg(c1=('c0', 'size'))
> print(result, flush=True)
> spark = (
> SparkSession.builder
> .master("local[2]")
> .config("spark.sql.ansi.enabled", "false")
> .config("spark.sql.shuffle.partitions", "2")
> .getOrCreate()
> )
> t0 = ps.DataFrame({'c0': [1]}, dtype='int32')
> print("PySpark Pandas:", flush=True)
> result = t0.groupby('c0', as_index=False).agg(c1=('c0', 'size'))
> print(result, flush=True)
> spark.stop()
> {code}
> {code:bash}
> Pandas:
> c0 c1
> 0 1 1
> PySpark Pandas:
> Traceback (most recent call last):
> File "/test.py", line 24, in <module>
> result = t0.groupby('c0', as_index=False).agg(c1=('c0', 'size'))
> ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
> File
> "/opt/conda/envs/lib/python3.11/site-packages/pyspark/pandas/groupby.py",
> line 314, in aggregate
> GroupBy._spark_groupby(self._psdf, func_or_funcs, self._groupkeys)
> File
> "/opt/conda/envs/lib/python3.11/site-packages/pyspark/pandas/groupby.py",
> line 406, in _spark_groupby
> sdf = sdf.groupby(*groupkey_names).agg(*reordered)
> ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
> File "/opt/conda/envs/lib/python3.11/site-packages/pyspark/sql/group.py",
> line 188, in agg
> jdf = self._jgd.agg(exprs[0]._jc, _to_seq(self.session._sc, [c._jc for c
> in exprs[1:]]))
>
> ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
> File "/opt/conda/envs/lib/python3.11/site-packages/py4j/java_gateway.py",
> line 1362, in __call__
> return_value = get_return_value(
> ^^^^^^^^^^^^^^^^^
> File
> "/opt/conda/envs/lib/python3.11/site-packages/pyspark/errors/exceptions/captured.py",
> line 257, in deco
> raise converted from None
> pyspark.errors.exceptions.captured.AnalysisException:
> [DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE] Cannot resolve "size(c0)" due to
> data type mismatch: The first parameter requires the ("ARRAY" or "MAP") type,
> however "c0" has the type "INT". SQLSTATE: 42K09; line 1 pos 0;
> 'Aggregate [__index_level_0__#3], [__index_level_0__#3, size(c0#1, true) AS
> (c0, size)#4]
> +- Project [c0#1 AS __index_level_0__#3, c0#1]
> +- Project [__index_level_0__#0L, c0#1, monotonically_increasing_id() AS
> __natural_order__#2L]
> +- LocalRelation [__index_level_0__#0L, c0#1]
> {code}
> *Expected behavior*
> GroupBy size should count rows in each group. For c0 = 1, the result should
> contain one row with c0 = 1 and c1 = 1, matching Pandas.
> *Actual behavior*
> The pandas-on-Spark aggregation raises AnalysisException with condition
> DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE and SQLSTATE 42K09. The message
> states that size(c0) requires an ARRAY or MAP, but c0 has type INT. The
> captured logical plan contains size(c0, true), rather than a row-count
> aggregation.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
