This is an automated email from the ASF dual-hosted git repository.
Yicong-Huang pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 9f69eff77928 [SPARK-57680][PYTHON] Remove unused
ArrowStreamAggPandasUDFSerializer
9f69eff77928 is described below
commit 9f69eff77928c68853e9d013ee572c52871f00ec
Author: Yicong Huang <[email protected]>
AuthorDate: Thu Jul 2 18:50:21 2026 +0000
[SPARK-57680][PYTHON] Remove unused ArrowStreamAggPandasUDFSerializer
### What changes were proposed in this pull request?
Remove `ArrowStreamAggPandasUDFSerializer` from `serializers.py`. This
class is no longer used after SPARK-57676 refactored
`SQL_GROUPED_AGG_PANDAS_ITER_UDF` (the last consumer) to use
`ArrowStreamGroupSerializer` directly, following the earlier refactors of
`SQL_GROUPED_AGG_PANDAS_UDF` (SPARK-56781) and `SQL_WINDOW_AGG_PANDAS_UDF`
(SPARK-57381).
### Why are the changes needed?
Dead code cleanup. Part of SPARK-55384 (Refactor PySpark Serializers).
Mirrors the equivalent Arrow-side cleanup in SPARK-56349.
### 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 #56948 from Yicong-Huang/cleanup-agg-pandas-ser.
Authored-by: Yicong Huang <[email protected]>
Signed-off-by: Yicong-Huang <[email protected]>
(cherry picked from commit c5a16fee9a3b955a5cce53ce1aac9d3758c4b209)
Signed-off-by: Yicong-Huang <[email protected]>
---
python/pyspark/sql/pandas/serializers.py | 56 --------------------------------
python/pyspark/worker.py | 5 ++-
2 files changed, 2 insertions(+), 59 deletions(-)
diff --git a/python/pyspark/sql/pandas/serializers.py
b/python/pyspark/sql/pandas/serializers.py
index 20e1f80a99b4..1bbd4ed296cd 100644
--- a/python/pyspark/sql/pandas/serializers.py
+++ b/python/pyspark/sql/pandas/serializers.py
@@ -492,62 +492,6 @@ class
ArrowStreamPandasUDTFSerializer(ArrowStreamPandasUDFSerializer):
return "ArrowStreamPandasUDTFSerializer"
-# Serializer for SQL_GROUPED_AGG_PANDAS_ITER_UDF
-class ArrowStreamAggPandasUDFSerializer(ArrowStreamPandasUDFSerializer):
- def __init__(
- self,
- *,
- timezone,
- safecheck,
- assign_cols_by_name,
- prefer_int_ext_dtype,
- int_to_decimal_coercion_enabled,
- ):
- super().__init__(
- timezone=timezone,
- safecheck=safecheck,
- assign_cols_by_name=assign_cols_by_name,
- df_for_struct=False,
- struct_in_pandas="dict",
- ndarray_as_list=False,
- prefer_int_ext_dtype=prefer_int_ext_dtype,
- arrow_cast=True,
- input_type=None,
- int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
- )
-
- def load_stream(self, stream):
- """
- Yield an iterator that produces one tuple of pandas.Series per batch.
- Each group yields Iterator[Tuple[pd.Series, ...]], allowing UDF to
- process batches one by one without consuming all batches upfront.
- """
- for batches in ArrowStreamGroupSerializer.load_stream(self, stream):
- # Lazily read and convert Arrow batches to pandas Series one at a
time
- # from the stream. This avoids loading all batches into memory for
the group
- series_iter = map(
- lambda batch: tuple(
- ArrowBatchTransformer.to_pandas(
- batch,
- timezone=self._timezone,
- schema=self._input_type,
- struct_in_pandas=self._struct_in_pandas,
- ndarray_as_list=self._ndarray_as_list,
- prefer_int_ext_dtype=self._prefer_int_ext_dtype,
- df_for_struct=self._df_for_struct,
- )
- ),
- batches,
- )
- yield series_iter
- # Make sure the batches are fully iterated before getting the next
group
- for _ in series_iter:
- pass
-
- def __repr__(self):
- return "ArrowStreamAggPandasUDFSerializer"
-
-
class ApplyInPandasWithStateSerializer(ArrowStreamPandasUDFSerializer):
"""
Serializer used by Python worker to evaluate UDF for
applyInPandasWithState.
diff --git a/python/pyspark/worker.py b/python/pyspark/worker.py
index 1b19409ec562..18cfc15f8f89 100644
--- a/python/pyspark/worker.py
+++ b/python/pyspark/worker.py
@@ -3746,9 +3746,8 @@ def invoke_udf(message_receiver: SparkMessageReceiver,
outfile: BinaryIO):
def _reader_thread():
try:
for batch in deserializer.load_stream(input_data_stream):
- # Some serializers (e.g., ArrowStreamGroupSerializer,
- # ArrowStreamAggPandasUDFSerializer) yield lazy
iterators
- # that still read from the input stream. Materialize
them here so
+ # Some serializers (e.g., ArrowStreamGroupSerializer)
yield lazy
+ # iterators that still read from the input stream.
Materialize them here so
# the main thread can consume them without touching
the stream.
if hasattr(batch, "__next__"):
batch = list(batch)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
