You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by yi...@apache.org on 2026/07/23 18:23:49 UTC
(spark) branch master updated: [SPARK-58262][PYTHON] Remove ArrowStreamPandasUDFSerializer
This is an automated email from the ASF dual-hosted git repository.
Yicong-Huang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new d44c1eab083e [SPARK-58262][PYTHON] Remove ArrowStreamPandasUDFSerializer
d44c1eab083e is described below
commit d44c1eab083e2993377b8c0c18fbe4947bf1e2a1
Author: Yicong Huang <17...@users.noreply.github.com>
AuthorDate: Thu Jul 23 18:23:29 2026 +0000
[SPARK-58262][PYTHON] Remove ArrowStreamPandasUDFSerializer
### What changes were proposed in this pull request?
Remove the `ArrowStreamPandasUDFSerializer` class from `python/pyspark/sql/pandas/serializers.py`.
### Why are the changes needed?
All non-grouped Pandas eval types (`SQL_SCALAR_PANDAS_UDF`, `SQL_SCALAR_PANDAS_ITER_UDF`, `SQL_MAP_PANDAS_ITER_UDF`) were refactored to use `ArrowStreamSerializer` with `ArrowBatchTransformer`, and the last remaining subclass `ApplyInPandasWithStateSerializer` was removed in SPARK-58243. The class no longer has any instantiation sites, so it can be removed.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing tests. No behavior change.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #57433 from Yicong-Huang/SPARK-58262.
Authored-by: Yicong Huang <17...@users.noreply.github.com>
Signed-off-by: Yicong-Huang <17...@users.noreply.github.com>
---
python/pyspark/sql/pandas/serializers.py | 88 --------------------------------
1 file changed, 88 deletions(-)
diff --git a/python/pyspark/sql/pandas/serializers.py b/python/pyspark/sql/pandas/serializers.py
index 2823cefd5821..a84debfab32e 100644
--- a/python/pyspark/sql/pandas/serializers.py
+++ b/python/pyspark/sql/pandas/serializers.py
@@ -341,94 +341,6 @@ class ArrowStreamPandasSerializer(ArrowStreamSerializer):
return "ArrowStreamPandasSerializer"
-class ArrowStreamPandasUDFSerializer(ArrowStreamPandasSerializer):
- """
- Serializer used by Python worker to evaluate Pandas UDFs
- """
-
- def __init__(
- self,
- *,
- timezone,
- safecheck,
- assign_cols_by_name,
- df_for_struct: bool = False,
- struct_in_pandas: str = "dict",
- ndarray_as_list: bool = False,
- prefer_int_ext_dtype: bool = False,
- arrow_cast: bool = False,
- input_type: Optional[StructType] = None,
- int_to_decimal_coercion_enabled: bool = False,
- prefers_large_types: bool = False,
- ignore_unexpected_complex_type_values: bool = False,
- is_legacy: bool = False,
- ):
- super().__init__(
- timezone=timezone,
- safecheck=safecheck,
- int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
- prefers_large_types=prefers_large_types,
- struct_in_pandas=struct_in_pandas,
- ndarray_as_list=ndarray_as_list,
- prefer_int_ext_dtype=prefer_int_ext_dtype,
- df_for_struct=df_for_struct,
- input_type=input_type,
- arrow_cast=arrow_cast,
- )
- self._assign_cols_by_name = assign_cols_by_name
- self._ignore_unexpected_complex_type_values = ignore_unexpected_complex_type_values
- self._is_legacy = is_legacy
-
- def dump_stream(self, iterator, stream):
- """
- Override because Pandas UDFs require a START_ARROW_STREAM before the Arrow stream is sent.
- This should be sent after creating the first record batch so in case of an error, it can
- be sent back to the JVM before the Arrow stream starts.
-
- Each element in iterator is:
- - For batched UDFs: tuple of (series, spark_type) tuples: ((s1, t1), (s2, t2), ...)
- - For iterator UDFs: single (series, spark_type) tuple directly
- """
- import pandas as pd
-
- def create_batch(
- series_tuples: Tuple[Tuple["pd.Series", DataType], ...],
- ) -> "pa.RecordBatch":
- # When struct_in_pandas="dict", UDF must return DataFrame for struct types
- if self._struct_in_pandas == "dict":
- for s, spark_type in series_tuples:
- if isinstance(spark_type, StructType) and not isinstance(s, pd.DataFrame):
- raise PySparkValueError(
- "Invalid return type. Please make sure that the UDF returns a "
- "pandas.DataFrame when the specified return type is StructType."
- )
-
- series_data = [s for s, _ in series_tuples]
- types = [t for _, t in series_tuples]
- schema = StructType([StructField(f"_{i}", t) for i, t in enumerate(types)])
- return PandasToArrowConversion.convert(
- series_data,
- schema,
- timezone=self._timezone,
- safecheck=self._safecheck,
- arrow_cast=self._arrow_cast,
- prefers_large_types=self._prefers_large_types,
- assign_cols_by_name=self._assign_cols_by_name,
- int_to_decimal_coercion_enabled=self._int_to_decimal_coercion_enabled,
- ignore_unexpected_complex_type_values=self._ignore_unexpected_complex_type_values,
- is_legacy=self._is_legacy,
- )
-
- batches = self._write_stream_start(
- (create_batch(_normalize_packed(packed)) for packed in iterator),
- stream,
- )
- return ArrowStreamSerializer.dump_stream(self, batches, stream)
-
- def __repr__(self):
- return "ArrowStreamPandasUDFSerializer"
-
-
class TransformWithStateInPySparkRowSerializer(ArrowStreamUDFSerializer):
"""
Serializer used by Python worker to evaluate UDF for
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org