Haotian Sun created SPARK-58625:
-----------------------------------
Summary: Support pa.ChunkedArray columns in PandasToArrowConversion
Key: SPARK-58625
URL: https://issues.apache.org/jira/browse/SPARK-58625
Project: Spark
Issue Type: Bug
Components: PySpark
Affects Versions: 5.0.0
Reporter: Haotian Sun
{{PandasToArrowConversion.convert}} assembles its result with
{{pa.RecordBatch.from_arrays}}, which rejects a {{pa.ChunkedArray}}. The
per-column conversion it feeds from calls {{pa.Array.from_pandas}}, which
returns a {{ChunkedArray}} when the input pandas Series is backed by a chunked
Arrow array (e.g. a pyarrow-backed extension dtype). A pandas UDF returning
such a Series fails with a raw pyarrow error:
{code}
TypeError: Cannot convert pyarrow.lib.ChunkedArray to pyarrow.lib.Array
{code}
Repro (no Spark session needed):
{code:python}
import pandas as pd, pyarrow as pa
from pyspark.sql.conversion import PandasToArrowConversion
from pyspark.sql.types import StructType, StructField, StringType
chunked = pa.chunked_array([pa.array(["a", "b"]), pa.array(["c", "d", "e"])])
series = pd.Series(chunked, dtype="string[pyarrow]")
schema = StructType([StructField("s", StringType())])
PandasToArrowConversion.convert([series], schema, arrow_cast=True)
{code}
In a UDF, a chunked-backed Series arises from ordinary pandas operations --
{{pd.concat}} of two pyarrow-backed Series produces two chunks, and {{.copy()}}
/ {{.reset_index()}} preserve the chunking, so whether the UDF crashes depends
on which operation it happened to end with.
The sibling function {{create_arrow_table_from_pandas}}
(python/pyspark/sql/pandas/conversion.py) already handles this and documents
why: it assembles with {{pa.Table.from_arrays}}, which accepts both {{Array}}
and {{ChunkedArray}}. That was SPARK-46776, which fixed the {{createDataFrame}}
path only. {{PandasToArrowConversion}} -- the pandas UDF return path -- never
got the same treatment, and its {{convert_column}} is annotated {{->
"pa.Array"}}, so the assembly trusts an annotation that {{from_pandas}} can
violate.
There is also no error classification here: the assembly call sits outside the
{{try/except}} that wraps the per-column conversion, so the pyarrow
{{TypeError}} escapes without becoming a {{PySparkTypeError}}. And unlike
{{createDataFrame}} -- where
{{spark.sql.execution.arrow.pyspark.fallback.enabled}} (default true)
downgrades the failure to a warning plus a non-Arrow slow path -- a UDF has no
such fallback, so this is a hard task failure.
Note on scope: string data over 2 GB also makes {{from_pandas}} return a
{{ChunkedArray}}, but that case cannot be fixed while returning a single
{{pa.RecordBatch}}, since 2 GiB of data does not fit one int32-offset buffer
(max 2 GiB - 1). It needs {{spark.sql.execution.arrow.useLargeVarTypes=true}}
(int64 offsets), which avoids the chunking entirely.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
