Skip to content

Commit ae7f6e3

Browse files
jiateohHeartSaVioR
authored andcommitted
[SPARK-56248][PYTHON][SS] Optimize python stateful processor serialization to skip unnecessary list/dict/row construction
### What changes were proposed in this pull request? Eliminate per-call Row(**dict(zip(...))) construction in _serialize_to_bytes, pass normalized tuples directly to schema.toInternal which handles them by index To better explain the removal of `Row` usage: - The positional ordering is retained because Row was constructed purely positionally: - Row.new stores values in insertion order ([link](https://github.com/jiateoh/spark/blob/615af5d154d1b15b26dfde3cc3441550eac5a615/python/pyspark/sql/types.py#L3547)). [This is also noted in the change notes since 3.0.0](https://github.com/jiateoh/spark/blob/615af5d154d1b15b26dfde3cc3441550eac5a615/python/pyspark/sql/types.py#L3492) - `dict(zip(...))` preserves insertion order in python 3.7+ (dicts preserve insertion order) - Inputs to `zip` are field_names, assumed to be same-ordered as the `converted` input data which is derived from the original input tuple in the same order. - `Row` is a tuple subclass, so it always hit [Schema.toInternal](https://github.com/jiateoh/spark/blob/615af5d154d1b15b26dfde3cc3441550eac5a615/python/pyspark/sql/types.py#L1763)'s tuple/list positional branch. Replacing it with the input tuple or a converted list will execute the same Schema.toInternal branch as before. The result is that the extra list, zip, dict, and Row construction is no longer necessary: the end result remains equivalent to the input to the entire `Row(**dict(zip(...)))` sequence: a positionally-ordered tuple/list AI-assisted + human reviewed/updated ### Why are the changes needed? This is a code cleanup/performance optimization. Original code has unnecessary operations that are executed for every row, including: rebuilding closures, extracting field names, building intermediate lists + dicts, and constructing Row objects (which sort by field unnecessarily). These can all add minor overhead while having no effect on the underlying usage. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Unit tests ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 4.6) Closes #55039 from jiateoh/tws_python_serialization_improvements. Authored-by: Jia Teoh <jiateoh@gmail.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>
1 parent e42a561 commit ae7f6e3

1 file changed

Lines changed: 3 additions & 6 deletions

File tree

python/pyspark/sql/streaming/stateful_processor_api_client.py

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -516,14 +516,11 @@ def normalize_value(v: Any) -> Any:
516516
else:
517517
return v
518518

519-
converted = [normalize_value(v) for v in data]
519+
converted = tuple(normalize_value(v) for v in data)
520520
else:
521-
converted = list(data)
521+
converted = data
522522

523-
field_names = [f.name for f in schema.fields]
524-
row_value = Row(**dict(zip(field_names, converted)))
525-
526-
return self.pickleSer.dumps(schema.toInternal(row_value))
523+
return self.pickleSer.dumps(schema.toInternal(converted))
527524

528525
def _deserialize_from_bytes(self, value: bytes) -> Any:
529526
return self.pickleSer.loads(value)

0 commit comments

Comments
 (0)