[
https://issues.apache.org/jira/browse/SPARK-58625?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ruifeng Zheng resolved SPARK-58625.
-----------------------------------
Fix Version/s: 4.4.0
Resolution: Fixed
Issue resolved by pull request 57829
[https://github.com/apache/spark/pull/57829]
> 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: 4.4.0
> Reporter: Haotian Sun
> Assignee: Haotian Sun
> Priority: Major
> Labels: pull-request-available
> Fix For: 4.4.0
>
>
> {{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]
