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 c7d354abcd65 [SPARK-58190][PYTHON] Delete unused
TransformWithStateInPandasInitStateSerializer
c7d354abcd65 is described below
commit c7d354abcd65ede10df56b0c9797677b8eed9185
Author: Yicong Huang <[email protected]>
AuthorDate: Mon Jul 20 17:07:19 2026 +0000
[SPARK-58190][PYTHON] Delete unused
TransformWithStateInPandasInitStateSerializer
### What changes were proposed in this pull request?
Delete the now-unused `TransformWithStateInPandasInitStateSerializer` class
from `serializers.py`.
[SPARK-58128](https://issues.apache.org/jira/browse/SPARK-58128) moved its
per-eval-type logic into `read_udfs()` in `worker.py` and switched
`SQL_TRANSFORM_WITH_STATE_PANDAS_INIT_STATE_UDF` to the plain
`ArrowStreamSerializer`, leaving this class with no references. This is the
follow-up cleanup that PR promised.
### Why are the changes needed?
Part of [SPARK-55388](https://issues.apache.org/jira/browse/SPARK-55388).
Removing the dead serializer keeps serializers as pure Arrow stream I/O with
eval-type-specific logic concentrated in `worker.py`.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing tests. No behavior change: the class had no remaining references
(verified repo-wide), and the `SQL_TRANSFORM_WITH_STATE_PANDAS_INIT_STATE_UDF`
path already runs through `worker.py` after SPARK-58128.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #57331 from Yicong-Huang/SPARK-58190-cleanup-tws-pandas-init-ser.
Authored-by: Yicong Huang <[email protected]>
Signed-off-by: Yicong-Huang <[email protected]>
(cherry picked from commit 2695dcde4e4f7d0133c1c798c74aff39de0a858e)
Signed-off-by: Yicong-Huang <[email protected]>
---
python/benchmarks/bench_eval_type.py | 7 +-
python/pyspark/sql/pandas/serializers.py | 160 -------------------------------
2 files changed, 3 insertions(+), 164 deletions(-)
diff --git a/python/benchmarks/bench_eval_type.py
b/python/benchmarks/bench_eval_type.py
index 8e0f6a312754..da59c664df56 100644
--- a/python/benchmarks/bench_eval_type.py
+++ b/python/benchmarks/bench_eval_type.py
@@ -2006,10 +2006,9 @@ class TransformWithStatePandasUDFPeakmemBench(
# either inputData or initState rows -- never both -- with the inactive column
# written as an all-null struct. Matching the JVM ``initData ++ data``
ordering,
# all initial-state batches are emitted first (initState populated), then all
-# data batches (inputData populated).
``TransformWithStateInPandasInitStateSerializer``
-# regroups rows by the leading key column, so each key surfaces as an init-only
-# call followed by a data-only call; the empty side of each call is filtered
out
-# before the UDF sees it.
+# data batches (inputData populated). The worker regroups rows by the leading
+# key column, so each key surfaces as an init-only call followed by a data-only
+# call; the empty side of each call is filtered out before the UDF sees it.
class
_TransformWithStatePandasInitStateBenchMixin(_TransformWithStatePandasBenchMixin):
diff --git a/python/pyspark/sql/pandas/serializers.py
b/python/pyspark/sql/pandas/serializers.py
index 890ce0d773e8..48516f47be54 100644
--- a/python/pyspark/sql/pandas/serializers.py
+++ b/python/pyspark/sql/pandas/serializers.py
@@ -980,166 +980,6 @@ class
TransformWithStateInPandasSerializer(ArrowStreamPandasUDFSerializer):
super().dump_stream(flatten_iterator(), stream)
-class
TransformWithStateInPandasInitStateSerializer(TransformWithStateInPandasSerializer):
- """
- Serializer used by Python worker to evaluate UDF for
-
:meth:`pyspark.sql.GroupedData.transformWithStateInPandasInitStateSerializer`.
- Parameters
- ----------
- Same as input parameters in TransformWithStateInPandasSerializer.
- """
-
- def __init__(
- self,
- *,
- timezone,
- safecheck,
- assign_cols_by_name,
- prefer_int_ext_dtype,
- arrow_max_records_per_batch,
- arrow_max_bytes_per_batch,
- int_to_decimal_coercion_enabled,
- ):
- super().__init__(
- timezone=timezone,
- safecheck=safecheck,
- assign_cols_by_name=assign_cols_by_name,
- prefer_int_ext_dtype=prefer_int_ext_dtype,
- arrow_max_records_per_batch=arrow_max_records_per_batch,
- arrow_max_bytes_per_batch=arrow_max_bytes_per_batch,
- int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
- )
- self.init_key_offsets = None
-
- def load_stream(self, stream):
- import pyarrow as pa
- import pandas as pd
- from pyspark.sql.streaming.stateful_processor_util import (
- TransformWithStateInPandasFuncMode,
- )
-
- def generate_data_batches(batches):
- """
- Deserialize ArrowRecordBatches and return a generator of
pandas.Series list.
-
- The deserialization logic assumes that Arrow RecordBatches contain
the data with the
- ordering that data chunks for same grouping key will appear
sequentially.
- See `TransformWithStateInPandasPythonInitialStateRunner` for arrow
batch schema sent
- from JVM.
- This function flatten the columns of input rows and initial state
rows and feed them
- into the data generator.
- """
-
- def flatten_columns(cur_batch, col_name):
- state_column =
cur_batch.column(cur_batch.schema.get_field_index(col_name))
-
- # Check if the entire column is null
- if state_column.null_count == len(state_column):
- # Return empty table with no columns
- return pa.Table.from_arrays([], names=[])
-
- state_field_names = [
- state_column.type[i].name for i in
range(state_column.type.num_fields)
- ]
- state_field_arrays = [
- state_column.field(i) for i in
range(state_column.type.num_fields)
- ]
- table_from_fields = pa.Table.from_arrays(
- state_field_arrays, names=state_field_names
- )
- return table_from_fields
-
- """
- The arrow batch is written in the schema:
- schema: StructType = new StructType()
- .add("inputData", dataSchema)
- .add("initState", initStateSchema)
- We'll parse batch into Tuples of (key, inputData, initState) and
pass into the Python
- data generator. Rows in the same batch may have different
grouping keys,
- but each batch will have either init_data or input_data, not mix.
- """
-
- def to_pandas(table):
- return ArrowBatchTransformer.to_pandas(
- table,
- 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,
- )
-
- def row_stream():
- for batch in batches:
- self._update_batch_size_stats(batch)
-
- data_table = flatten_columns(batch, "inputData")
- init_table = flatten_columns(batch, "initState")
-
- # Check column count - empty table has no columns
- has_data = data_table.num_columns > 0
- has_init = init_table.num_columns > 0
-
- assert not (has_data and has_init)
-
- if has_data:
- for row in pd.concat(to_pandas(data_table),
axis=1).itertuples(index=False):
- batch_key = tuple(row[s] for s in self.key_offsets)
- yield (batch_key, row, None)
- elif has_init:
- for row in pd.concat(to_pandas(init_table),
axis=1).itertuples(index=False):
- batch_key = tuple(row[s] for s in
self.init_key_offsets)
- yield (batch_key, None, row)
-
- EMPTY_DATAFRAME = pd.DataFrame()
- for batch_key, group_rows in groupby(row_stream(), key=lambda x:
x[0]):
- rows = []
- init_state_rows = []
- for _, row, init_state_row in group_rows:
- if row is not None:
- rows.append(row)
- if init_state_row is not None:
- init_state_rows.append(init_state_row)
-
- total_len = len(rows) + len(init_state_rows)
- if (
- total_len >= self.arrow_max_records_per_batch
- or total_len * self.average_arrow_row_size >=
self.arrow_max_bytes_per_batch
- ):
- yield (
- batch_key,
- pd.DataFrame(rows) if len(rows) > 0 else
EMPTY_DATAFRAME.copy(),
- (
- pd.DataFrame(init_state_rows)
- if len(init_state_rows) > 0
- else EMPTY_DATAFRAME.copy()
- ),
- )
- rows = []
- init_state_rows = []
- if rows or init_state_rows:
- yield (
- batch_key,
- pd.DataFrame(rows) if len(rows) > 0 else
EMPTY_DATAFRAME.copy(),
- (
- pd.DataFrame(init_state_rows)
- if len(init_state_rows) > 0
- else EMPTY_DATAFRAME.copy()
- ),
- )
-
- _batches = super(ArrowStreamPandasSerializer, self).load_stream(stream)
- data_batches = generate_data_batches(_batches)
-
- for k, g in groupby(data_batches, key=lambda x: x[0]):
- yield (TransformWithStateInPandasFuncMode.PROCESS_DATA, k, g)
-
- yield (TransformWithStateInPandasFuncMode.PROCESS_TIMER, None, None)
-
- yield (TransformWithStateInPandasFuncMode.COMPLETE, None, None)
-
-
class TransformWithStateInPySparkRowSerializer(ArrowStreamUDFSerializer):
"""
Serializer used by Python worker to evaluate UDF for
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
