asddfl created SPARK-60650:
------------------------------
Summary: 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
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]
