Skip to content

Commit e42a561

Browse files
Yicong-HuangHyukjinKwon
authored andcommitted
[SPARK-56359][PYTHON] Remove unused ArrowStreamArrowUDFSerializer
### What changes were proposed in this pull request? Remove `ArrowStreamArrowUDFSerializer` from `serializers.py`. This class is no longer used after: - SPARK-55577 refactored `SQL_SCALAR_ARROW_ITER_UDF` to use `ArrowStreamSerializer` directly - SPARK-56348 deleted `ArrowBatchUDFSerializer` (child) - SPARK-56349 deleted `ArrowStreamAggArrowUDFSerializer` (child) ### Why are the changes needed? Dead code cleanup. Part of SPARK-55384 (Refactor PySpark Serializers). ### 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 #55212 from Yicong-Huang/SPARK-56359/cleanup/arrow-stream-arrow-udf-serializer. Authored-by: Yicong-Huang <17627829+Yicong-Huang@users.noreply.github.com> Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>
1 parent 5beaa5b commit e42a561

1 file changed

Lines changed: 0 additions & 48 deletions

File tree

python/pyspark/sql/pandas/serializers.py

Lines changed: 0 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -549,54 +549,6 @@ def __repr__(self):
549549
return "ArrowStreamPandasUDFSerializer"
550550

551551

552-
class ArrowStreamArrowUDFSerializer(ArrowStreamSerializer):
553-
"""
554-
Serializer used by Python worker to evaluate Arrow UDFs
555-
"""
556-
557-
def __init__(
558-
self,
559-
*,
560-
safecheck,
561-
arrow_cast,
562-
):
563-
super().__init__()
564-
self._safecheck = safecheck
565-
self._arrow_cast = arrow_cast
566-
567-
def dump_stream(self, iterator, stream):
568-
"""
569-
Override because Arrow UDFs require a START_ARROW_STREAM before the Arrow stream is sent.
570-
This should be sent after creating the first record batch so in case of an error, it can
571-
be sent back to the JVM before the Arrow stream starts.
572-
"""
573-
import pyarrow as pa
574-
575-
def create_batch(
576-
arr_tuples: List[Tuple["pa.Array", "pa.DataType"]],
577-
) -> "pa.RecordBatch":
578-
names = ["_%d" % i for i in range(len(arr_tuples))]
579-
arrs = [arr for arr, _ in arr_tuples]
580-
batch = pa.RecordBatch.from_arrays(arrs, names)
581-
target_schema = pa.schema([pa.field(n, t) for n, (_, t) in zip(names, arr_tuples)])
582-
return ArrowBatchTransformer.enforce_schema(
583-
batch, target_schema, safecheck=self._safecheck, arrow_cast=self._arrow_cast
584-
)
585-
586-
def normalize(packed):
587-
if len(packed) == 2 and isinstance(packed[1], pa.DataType):
588-
return [packed]
589-
return list(packed)
590-
591-
batches = self._write_stream_start(
592-
(create_batch(normalize(packed)) for packed in iterator), stream
593-
)
594-
return ArrowStreamSerializer.dump_stream(self, batches, stream)
595-
596-
def __repr__(self):
597-
return "ArrowStreamArrowUDFSerializer"
598-
599-
600552
class ArrowStreamPandasUDTFSerializer(ArrowStreamPandasUDFSerializer):
601553
"""
602554
Serializer used by Python worker to evaluate Arrow-optimized Python UDTFs.

0 commit comments

Comments
 (0)