You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by yi...@apache.org on 2026/08/21 01:39:32 UTC
(spark) branch branch-4.3 updated: [SPARK-58699][PYTHON] Short-circuit identity conversion for string and binary in LocalDataToArrowConversion
This is an automated email from the ASF dual-hosted git repository.
Yicong-Huang pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new c9cb18888e58 [SPARK-58699][PYTHON] Short-circuit identity conversion for string and binary in LocalDataToArrowConversion
c9cb18888e58 is described below
commit c9cb18888e58056e86e3eda7a595a7f3f0b22c13
Author: Yicong Huang <17...@users.noreply.github.com>
AuthorDate: Fri Aug 21 01:38:24 2026 +0000
[SPARK-58699][PYTHON] Short-circuit identity conversion for string and binary in LocalDataToArrowConversion
### What changes were proposed in this pull request?
`LocalDataToArrowConversion` converts local Python rows to Arrow element by element. For `string` and `binary` leaf types the per-element converter returns the value unchanged when it is already the target Python type (`str` / `bytes`), yet it still pays a Python function call and the constructor dispatch of `str(value)` / `bytes(value)` per element (`str(s) is s` and `bytes(b) is b` — no copy, but the `str()` / `bytes()` call still dispatches on the argument type every time).
This PR adds a type-check fast path that returns such values directly:
- `convert_string`: `type(value) is str` returns the value, skipping the `str()` call (and the bool check).
- `convert_binary`: `type(value) is bytes` returns the value, skipping the `bytes()` call. `bytearray` still falls through and is copied into an immutable `bytes`.
- `convert_array`: for `string` / `binary` element types, the fast path is inlined into the list comprehension so already-target-typed elements skip the per-element converter call entirely. Any other element — including `None` (whose nullability is still enforced by the reused `element_conv`) and values that need coercion (e.g. a `bool` to string) — falls back to `element_conv`, which is reused unchanged.
The checks are ordered `None`-first in the scalar converters so a column that is entirely `None` (a common, legitimate case for nullable columns) does not regress. This is orthogonal to SPARK-52796, which optimized the outer `convert()` assembly rather than the element-level conversion.
### Why are the changes needed?
Converting local Python data to Arrow is the hot path of Spark Connect `createDataFrame` and Arrow-optimized Python UDF / DataSource output, where `string` is the most common non-trivial leaf type.
This is a heuristic: faster when elements are already `str` / `bytes`, slower when they need coercion. The ASV benchmark (`bench_local_data_to_arrow.py`) sweeps the non-target fraction from 0% to 100% for a scalar, a one-level `array`, and a two-level `array<array<...>>` column. Each cell is `micro / e2e` percent change vs master (negative = faster): `micro` = `LocalDataToArrowConversion.convert` alone (1M rows, median of 3 runs); `e2e` = full Spark Connect `createDataFrame` including [...]
**`StringType`**
| leaf | other | 0% | 30% | 70% | 100% |
|---|---|---|---|---|---|
| `string` | none | -12.6 / -10.1 | -7.9 / -7.2 | -4.5 / -4.2 | +0.4 / +0.0 |
| `array<string>` | none | -25.6 / -17.0 | -19.0 / -10.0 | -6.1 / -4.1 | +9.5 / +4.8 |
| `array<array<string>>` | none | -20.1 / -8.7 | -13.2 / -6.6 | -3.3 / -0.1 | +3.5 / +0.7 |
| `string` | int | -12.1 / -9.2 | -7.5 / -6.5 | -4.2 / -4.6 | -2.1 / -3.0 |
| `array<string>` | int | -25.0 / -12.9 | -15.6 / -8.8 | -4.1 / -3.4 | +3.2 / +1.6 |
| `array<array<string>>` | int | -19.0 / -8.7 | -13.3 / -5.9 | -4.5 / -3.7 | +2.7 / +0.7 |
| `string` | bool | -9.0 / -10.2 | -14.6 / -11.0 | -20.6 / -16.6 | -23.9 / -16.7 |
| `array<string>` | bool | -21.9 / -10.8 | -22.9 / -13.5 | -19.9 / -12.6 | -18.2 / -12.2 |
| `array<array<string>>` | bool | -19.6 / -8.4 | -17.8 / -9.6 | -15.9 / -9.2 | -13.0 / -8.0 |
| `string` | mix | -12.0 / -9.4 | -7.6 / -6.3 | -3.5 / -4.6 | -3.9 / -2.2 |
| `array<string>` | mix | -28.0 / -11.8 | -11.8 / -8.3 | -4.5 / -2.9 | -1.1 / -1.2 |
| `array<array<string>>` | mix | -19.7 / -9.5 | -12.4 / -6.0 | -6.9 / -3.4 | -2.1 / -1.2 |
**`BinaryType`**
| leaf | other | 0% | 30% | 70% | 100% |
|---|---|---|---|---|---|
| `binary` | none | -31.9 / -26.5 | -25.0 / -21.1 | -14.0 / -11.1 | +0.5 / +0.3 |
| `array<binary>` | none | -45.3 / -19.6 | -32.4 / -19.0 | -14.5 / -7.3 | +9.9 / +4.8 |
| `array<array<binary>>` | none | -32.0 / -17.2 | -23.3 / -12.1 | -10.2 / -5.4 | +2.1 / +1.7 |
| `binary` | bytearray | -31.7 / -26.0 | -19.7 / -16.1 | -3.4 / -3.3 | +4.2 / +5.1 |
| `array<binary>` | bytearray | -45.0 / -27.6 | -26.0 / -16.1 | -7.5 / -3.0 | +8.3 / +5.1 |
| `array<array<binary>>` | bytearray | -32.9 / -17.1 | -22.0 / -10.3 | -4.8 / -2.6 | +7.3 / +2.8 |
`mix` is a rotation of int / float / bool / Decimal / date / datetime. Differences within ~2-3pp are at the noise floor (each 0% row is the identical all-target workload, measured once per `other` kind).
### Does this PR introduce _any_ user-facing change?
No. The conversion results are unchanged, including nullability enforcement (a `None` element in a non-nullable array still raises) and the coercion of non-target values (bool/int to string, `bytearray` to bytes). This is a performance-only change.
### How was this patch tested?
- Added `test_string_binary_identity_fast_path` in `python/pyspark/sql/tests/test_conversion.py`, covering the identity fast path for `str` / `bytes` and the fallback for non-target elements (bool/int coerced to string, `bytearray` materialized as immutable `bytes`) across scalar, array, map-value and struct leaves.
- Existing `test_conversion.py` suite passes.
- Added `bench_local_data_to_arrow.py` ASV benchmark, sweeping scalar / `array` / `array<array<...>>` leaves against `none` / `int` / `bool` / `mix` (string) and `none` / `bytearray` (binary) non-target values.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #57911 from Yicong-Huang/bench-localdata-string.
Authored-by: Yicong Huang <17...@users.noreply.github.com>
Signed-off-by: Yicong-Huang <17...@users.noreply.github.com>
(cherry picked from commit 57488264323e8ffe423cb90e45cd798ff3477c2b)
Signed-off-by: Yicong-Huang <17...@users.noreply.github.com>
---
python/benchmarks/bench_local_data_to_arrow.py | 160 +++++++++++++++++++++++++
python/pyspark/sql/conversion.py | 45 ++++++-
python/pyspark/sql/tests/test_conversion.py | 7 ++
3 files changed, 206 insertions(+), 6 deletions(-)
diff --git a/python/benchmarks/bench_local_data_to_arrow.py b/python/benchmarks/bench_local_data_to_arrow.py
new file mode 100644
index 000000000000..55f3100311ce
--- /dev/null
+++ b/python/benchmarks/bench_local_data_to_arrow.py
@@ -0,0 +1,160 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Microbenchmarks for ``LocalDataToArrowConversion.convert``, the hot path of
+Spark Connect ``createDataFrame`` and Arrow-optimized Python UDF output.
+
+``string`` / ``binary`` columns are converted element by element: the per-element
+converter is fast when the element is already the target type (``str`` /
+``bytes``) and slower when it needs coercion. Each benchmark sweeps the fraction
+of non-target ("other") values from 0% (all fast path) to 100% (all slow path),
+for a scalar column, a one-level ``array`` column, and a two-level
+``array<array<...>>`` column (which drives the inlined array fast path hardest).
+"""
+
+
+def _build(convert, is_string, leaf, other_val, other_frac, n_rows):
+ import random
+ from pyspark.sql.types import ArrayType, BinaryType, StringType, StructField, StructType
+
+ leaf_type = StringType() if is_string else BinaryType()
+
+ def target(i):
+ return f"s{i}" if is_string else b"s%d" % i
+
+ rnd = random.Random(0)
+
+ def elem(i):
+ return other_val(i) if rnd.random() * 100 < other_frac else target(i)
+
+ if leaf == "array2":
+ schema = StructType([StructField("c", ArrayType(ArrayType(leaf_type)), True)])
+ data = [([[elem(i), elem(i + 1)], [elem(i + 2), elem(i + 3)]],) for i in range(n_rows)]
+ elif leaf == "array":
+ schema = StructType([StructField("c", ArrayType(leaf_type), True)])
+ data = [([elem(i), elem(i + 1), elem(i + 2)],) for i in range(n_rows)]
+ else: # scalar
+ schema = StructType([StructField("c", leaf_type, True)])
+ data = [(elem(i),) for i in range(n_rows)]
+ return data, schema
+
+
+class LocalDataToArrowStringBenchmark:
+ """
+ Benchmark ``convert`` on a ``string`` column, sweeping the kind and fraction
+ of non-target values.
+
+ - ``leaf``: ``scalar``, one-level ``array``, or two-level ``array2``
+ (``array<array<string>>``).
+ - ``other``: the non-target values -- ``none`` (null), ``int`` or ``bool``
+ (a single non-str type, isolating one coercion branch), or ``mix`` (a
+ rotation of int / float / bool / Decimal / date / datetime).
+ - ``other_frac``: fraction of elements that are non-target, ``0`` to ``100``.
+ """
+
+ params = [
+ [1000000],
+ ["scalar", "array", "array2"],
+ ["none", "int", "bool", "mix"],
+ [0, 30, 70, 100],
+ ]
+ param_names = ["n_rows", "leaf", "other", "other_frac"]
+
+ def setup(self, n_rows, leaf, other, other_frac):
+ import datetime
+ import decimal
+ from pyspark.sql.conversion import LocalDataToArrowConversion
+
+ self.convert = LocalDataToArrowConversion.convert
+
+ if other == "none":
+
+ def other_val(i):
+ return None
+ elif other == "int":
+
+ def other_val(i):
+ return i # int coerced to string via str()
+ elif other == "bool":
+
+ def other_val(i):
+ return i % 2 == 0 # bool coerced to "true" / "false"
+ else: # mix: a rotation of non-str types, each coerced to string
+ _pool = [
+ 123,
+ 4.5,
+ True,
+ False,
+ decimal.Decimal("1.50"),
+ datetime.date(2020, 1, 1),
+ datetime.datetime(2020, 1, 1, 3, 4, 5),
+ ]
+
+ def other_val(i):
+ return _pool[i % len(_pool)]
+
+ self.data, self.schema = _build(self.convert, True, leaf, other_val, other_frac, n_rows)
+
+ def time_convert(self, n_rows, leaf, other, other_frac):
+ self.convert(self.data, self.schema, False)
+
+ def peakmem_convert(self, n_rows, leaf, other, other_frac):
+ self.convert(self.data, self.schema, False)
+
+
+class LocalDataToArrowBinaryBenchmark:
+ """
+ Benchmark ``convert`` on a ``binary`` column, sweeping the fraction of
+ non-target values.
+
+ - ``leaf``: ``scalar``, one-level ``array``, or two-level ``array2``
+ (``array<array<binary>>``).
+ - ``other``: ``none`` (null) or ``bytearray`` (the only non-bytes input
+ binary accepts, copied to immutable bytes).
+ - ``other_frac``: fraction of elements that are non-target, ``0`` to ``100``.
+ """
+
+ params = [
+ [1000000],
+ ["scalar", "array", "array2"],
+ ["none", "bytearray"],
+ [0, 30, 70, 100],
+ ]
+ param_names = ["n_rows", "leaf", "other", "other_frac"]
+
+ def setup(self, n_rows, leaf, other, other_frac):
+ from pyspark.sql.conversion import LocalDataToArrowConversion
+
+ self.convert = LocalDataToArrowConversion.convert
+
+ if other == "none":
+
+ def other_val(i):
+ return None
+ else: # bytearray coerced to immutable bytes
+
+ def other_val(i):
+ return bytearray(b"s%d" % i)
+
+ self.data, self.schema = _build(self.convert, False, leaf, other_val, other_frac, n_rows)
+
+ def time_convert(self, n_rows, leaf, other, other_frac):
+ self.convert(self.data, self.schema, False)
+
+ def peakmem_convert(self, n_rows, leaf, other, other_frac):
+ self.convert(self.data, self.schema, False)
diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py
index e472c7304e2e..20d37def3591 100644
--- a/python/pyspark/sql/conversion.py
+++ b/python/pyspark/sql/conversion.py
@@ -682,6 +682,27 @@ class LocalDataToArrowConversion:
assert isinstance(value, (list, array.array))
return list(value)
+ elif isinstance(dataType.elementType, (StringType, BinaryType)):
+ # Inline the scalar identity fast path so elements that are
+ # already the target Python type skip the per-element converter
+ # call entirely: `convert_string`/`convert_binary` return such
+ # elements unchanged. `str` and immutable `bytes` are the two
+ # element types whose converter is a no-op on a matching value.
+ # Any other element -- including `None` (whose nullability is
+ # enforced by `element_conv`) and values that need coercion (e.g.
+ # a bool to string) -- falls back to `element_conv`, reused
+ # unchanged.
+ fast_type = str if isinstance(dataType.elementType, StringType) else bytes
+
+ def convert_array(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (list, array.array))
+ return [v if type(v) is fast_type else element_conv(v) for v in value]
+
else:
def convert_array(value: Any) -> Any:
@@ -739,6 +760,12 @@ class LocalDataToArrowConversion:
if not nullable:
raise PySparkValueError(f"input for {dataType} must not be None")
return None
+ elif type(value) is bytes:
+ # Fast path: `bytes(value)` returns `value` itself for a `bytes`
+ # input (no copy, as `bytes` is immutable), but still pays the
+ # constructor dispatch per element. Returning it directly skips
+ # that. `bytearray` falls through and is copied into `bytes`.
+ return value
else:
assert isinstance(value, (bytes, bytearray))
return bytes(value)
@@ -801,13 +828,19 @@ class LocalDataToArrowConversion:
if not nullable:
raise PySparkValueError(f"input for {dataType} must not be None")
return None
+ elif type(value) is str:
+ # Fast path: `str(value)` returns `value` itself for a `str`
+ # input (no copy), but still pays the constructor dispatch per
+ # element. Returning it directly skips that and the bool check.
+ return value
+ elif value is True:
+ # To match the PySpark Classic which convert bool to string in
+ # the JVM side (python.EvaluatePython.makeFromJava)
+ return "true"
+ elif value is False:
+ return "false"
else:
- if isinstance(value, bool):
- # To match the PySpark Classic which convert bool to string in
- # the JVM side (python.EvaluatePython.makeFromJava)
- return str(value).lower()
- else:
- return str(value)
+ return str(value)
return convert_string
diff --git a/python/pyspark/sql/tests/test_conversion.py b/python/pyspark/sql/tests/test_conversion.py
index d02dd8e73497..eab91d6b34a2 100644
--- a/python/pyspark/sql/tests/test_conversion.py
+++ b/python/pyspark/sql/tests/test_conversion.py
@@ -471,12 +471,19 @@ class ConversionTests(unittest.TestCase):
(IntegerType(), (1,), (None,)),
((IntegerType(), {"nullable": False}), (1,)),
(StringType(), ("a",)),
+ # bool coerced to string matches the JVM (EvaluatePython.makeFromJava).
+ (StringType(), (True, "true"), (False, "false")),
(BinaryType(), (b"a",)),
(GeographyType("ANY"), (None,)),
(GeometryType("ANY"), (None,)),
(ArrayType(IntegerType()), ([1, None],)),
(ArrayType(IntegerType(), containsNull=False), ([1, 2],)),
(ArrayType(BinaryType()), ([b"a", b"b"],)),
+ # array<string> with already-str, coerced (int/bool) and null elements.
+ (
+ ArrayType(StringType()),
+ (["ok", 42, True, False, None], ["ok", "42", "true", "false", None]),
+ ),
(MapType(StringType(), IntegerType()), ({"a": 1, "b": None},)),
(
MapType(StringType(), IntegerType(), valueContainsNull=False),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org