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 c7b2f1a865cc [SPARK-58314][PYTHON] Remove ArrowStreamUDFSerializer
c7b2f1a865cc is described below
commit c7b2f1a865cce621d2c702bdfdff5132e251ca74
Author: Yicong Huang <[email protected]>
AuthorDate: Sun Jul 26 14:36:05 2026 +0000
[SPARK-58314][PYTHON] Remove ArrowStreamUDFSerializer
### What changes were proposed in this pull request?
Removes the now-unused `ArrowStreamUDFSerializer` class from
`python/pyspark/sql/pandas/serializers.py`, and updates two stale docstring
references to it in `conversion.py` (`ArrowBatchTransformer.flatten_struct` /
`wrap_struct`).
After the PythonEvalType refactor moved serializer construction into
`worker.py` and its last subclass `TransformWithStateInPySparkRowSerializer`
was removed (SPARK-58297), `ArrowStreamUDFSerializer` has no remaining runtime
users. The `flatten_struct` / `wrap_struct` transformer methods it used to call
are retained -- they are still used directly by the Arrow UDF mappers in
`worker.py`.
### Why are the changes needed?
Dead-code cleanup. Part of SPARK-55388.
### 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 #57507 from Yicong-Huang/SPARK-58314.
Authored-by: Yicong Huang <[email protected]>
Signed-off-by: Yicong-Huang <[email protected]>
---
python/pyspark/sql/conversion.py | 4 ++--
python/pyspark/sql/pandas/serializers.py | 24 ------------------------
2 files changed, 2 insertions(+), 26 deletions(-)
diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py
index e660ddc18e12..e472c7304e2e 100644
--- a/python/pyspark/sql/conversion.py
+++ b/python/pyspark/sql/conversion.py
@@ -83,7 +83,6 @@ class ArrowBatchTransformer:
Flatten a struct column at given index into a RecordBatch.
Used by:
- - ArrowStreamUDFSerializer.load_stream
- SQL_GROUPED_MAP_ARROW_UDF mapper
- SQL_GROUPED_MAP_ARROW_ITER_UDF mapper
"""
@@ -111,7 +110,8 @@ class ArrowBatchTransformer:
"""
Wrap a RecordBatch's columns into a single struct column.
- Used by: ArrowStreamUDFSerializer.dump_stream
+ Used by: Arrow UDF mappers in worker.py to re-wrap flattened batches
+ before serialization.
"""
import pyarrow as pa
diff --git a/python/pyspark/sql/pandas/serializers.py
b/python/pyspark/sql/pandas/serializers.py
index e62d688c6a25..1ac0baad15e5 100644
--- a/python/pyspark/sql/pandas/serializers.py
+++ b/python/pyspark/sql/pandas/serializers.py
@@ -214,30 +214,6 @@ class ArrowStreamCoGroupSerializer(ArrowStreamSerializer):
)
-class ArrowStreamUDFSerializer(ArrowStreamSerializer):
- """
- Same as :class:`ArrowStreamSerializer` but it flattens the struct to Arrow
record batch
- for applying each function with the raw record arrow batch. See also
`DataFrame.mapInArrow`.
- """
-
- def load_stream(self, stream):
- """
- Flatten the struct into Arrow's record batches.
- """
- batches = super().load_stream(stream)
- flattened = map(ArrowBatchTransformer.flatten_struct, batches)
- return map(lambda b: [b], flattened)
-
- def dump_stream(self, iterator, stream):
- """
- Override because Pandas UDFs require a START_ARROW_STREAM before the
Arrow stream is sent.
- """
- batches = self._write_stream_start(
- (ArrowBatchTransformer.wrap_struct(x[0]) for x in iterator), stream
- )
- return super().dump_stream(batches, stream)
-
-
class ArrowStreamPandasSerializer(ArrowStreamSerializer):
"""
Serializes pandas.Series as Arrow data with Arrow streaming format.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
