You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ru...@apache.org on 2026/09/04 05:21:40 UTC
(spark) branch branch-4.x updated: [SPARK-59219][PYTHON] Avoid copying the Series to name it in convert_numpy
This is an automated email from the ASF dual-hosted git repository.
zhengruifeng 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 e94c638e5b89 [SPARK-59219][PYTHON] Avoid copying the Series to name it in convert_numpy
e94c638e5b89 is described below
commit e94c638e5b8953db257435cb474444157d20d5c3
Author: Spenser Sun <hs...@gmail.com>
AuthorDate: Fri Sep 4 13:21:16 2026 +0800
[SPARK-59219][PYTHON] Avoid copying the Series to name it in convert_numpy
### What changes were proposed in this pull request?
`ArrowArrayToPandasConversion.convert_numpy` ended with `series.rename(ser_name)`, but `Series.rename` is not an in-place rename: pandas implements it as `self.copy(deep=False)` followed by setting the name. The Series is created inside `convert_numpy`, so this names it in place instead.
This also adds `python/benchmarks/bench_arrow_to_pandas.py`, since nothing in the committed ASV suite exercised the Arrow-to-pandas converter.
### Why are the changes needed?
The copy runs per column per batch and is fixed cost that does not scale with row count, so it matters most for wide or small batches. For 50 int64 columns at 10k rows the path costs 2.71ms, against 1.11ms for a bare per-column `to_pandas()` loop; this removes ~21us of that ~32us per-column gap. (The bare loop does not set the Series name, so it is a floor this path cannot reach.)
`bench_arrow_to_pandas.py`, master vs this PR:
| n_rows | n_cols | long | timestamp | string |
| --- | --- | --- | --- | --- |
| 128 | 1 | 60.7 -> 41.3us (1.47x) | 196 -> 139us (1.41x) | 160 -> 157us |
| 128 | 50 | 2.75 -> 1.61ms (1.71x) | 7.05 -> 5.41ms (1.30x) | 6.34 -> 6.32ms |
| 10000 | 1 | 63.3 -> 42.3us (1.50x) | 732 -> 690us (1.06x) | 199 -> 196us |
| 10000 | 50 | 2.64 -> 1.66ms (1.59x) | 35.6 -> 33.5ms (1.06x) | 7.18 -> 7.07ms |
`long` and `timestamp` take `convert_numpy`; `string` takes `convert_legacy` and is unaffected. Since the saving is fixed per column, the ratio shrinks as per-row work grows: `timestamp` is 1.30x at 128 rows but 1.06x at 10k.
### Does this PR introduce _any_ user-facing change?
No. The returned Series is identical.
### How was this patch tested?
Existing `test_conversion.py` and `test_pandas_udf_scalar.py` pass, plus a new test pinning that `convert_numpy` reads the Arrow field name before `preprocess_time`, whose `pa.compute` kernels return a new array without it; moving that capture below `preprocess_time` fails only the new test.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)
Closes #58504 from Spenserrrr/conversion-bench.
Authored-by: Spenser Sun <hs...@gmail.com>
Signed-off-by: Ruifeng Zheng <ru...@foxmail.com>
(cherry picked from commit 7b60b3eb76c4f98f328360116373fa3714b7d9e2)
Signed-off-by: Ruifeng Zheng <ru...@foxmail.com>
---
python/benchmarks/bench_arrow_to_pandas.py | 80 +++++++++++++++++++++++++++++
python/pyspark/sql/conversion.py | 4 +-
python/pyspark/sql/tests/test_conversion.py | 15 ++++++
3 files changed, 98 insertions(+), 1 deletion(-)
diff --git a/python/benchmarks/bench_arrow_to_pandas.py b/python/benchmarks/bench_arrow_to_pandas.py
new file mode 100644
index 000000000000..80e954b8ae0e
--- /dev/null
+++ b/python/benchmarks/bench_arrow_to_pandas.py
@@ -0,0 +1,80 @@
+#
+# 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 ``ArrowBatchTransformer.to_pandas``, the hot path of pandas
+UDF inputs: every pandas UDF eval type calls it once per batch to build the
+Series it passes to the user's function.
+
+Part of the per-batch cost is fixed per COLUMN and does not scale with row
+count, so ``n_cols`` is swept alongside ``n_rows``: wide batches and small
+batches are the shapes where that fixed cost dominates.
+
+``ArrowArrayToPandasConversion.convert`` routes each column by type. ``long`` and
+``timestamp`` are in the ``_prefer_convert_numpy`` allowlist and take
+``convert_numpy``; ``string`` is not, and takes ``convert_legacy``. The two
+allowlist types differ by an order of magnitude in conversion cost -- a timestamp
+column is localized by pyarrow compute kernels -- so sweeping both shows how much
+of a change is fixed per-column cost rather than per-row work.
+"""
+
+import numpy as np
+import pyarrow as pa
+
+
+class ArrowBatchToPandasBenchmark:
+ """Benchmark ``ArrowBatchTransformer.to_pandas`` over a whole RecordBatch."""
+
+ params = [
+ [128, 10000],
+ [1, 50],
+ ["long", "timestamp", "string"],
+ ]
+ param_names = ["n_rows", "n_cols", "col_type"]
+
+ def setup(self, n_rows, n_cols, col_type):
+ from pyspark.sql.conversion import ArrowBatchTransformer
+ from pyspark.sql.types import (
+ LongType,
+ StringType,
+ StructField,
+ StructType,
+ TimestampType,
+ )
+
+ if col_type == "long":
+ column = pa.array(np.arange(n_rows, dtype=np.int64))
+ spark_type = LongType()
+ elif col_type == "timestamp":
+ base = np.datetime64("2020-01-01T00:00:00", "us")
+ column = pa.array(base + np.arange(n_rows) * np.timedelta64(1, "s")).cast(
+ pa.timestamp("us", tz="UTC")
+ )
+ spark_type = TimestampType()
+ elif col_type == "string":
+ column = pa.array([f"s{i:07d}" for i in range(n_rows)], type=pa.string())
+ spark_type = StringType()
+ else:
+ raise ValueError(f"unknown col_type: {col_type}")
+
+ names = [f"c{i}" for i in range(n_cols)]
+ self.batch = pa.RecordBatch.from_arrays([column] * n_cols, names)
+ self.schema = StructType([StructField(name, spark_type) for name in names])
+ self.to_pandas = ArrowBatchTransformer.to_pandas
+
+ def time_batch_to_pandas(self, n_rows, n_cols, col_type):
+ self.to_pandas(self.batch, timezone="UTC", schema=self.schema)
diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py
index 0585a730fb08..f2fd2c6b0c3b 100644
--- a/python/pyspark/sql/conversion.py
+++ b/python/pyspark/sql/conversion.py
@@ -2045,4 +2045,6 @@ class ArrowArrayToPandasConversion:
else: # pragma: no cover
assert False, f"Need converter for {spark_type} but failed to find one."
- return series.rename(ser_name)
+ # `series` is created in this method, so naming it in place is safe; rename() copies it.
+ series.name = ser_name
+ return series
diff --git a/python/pyspark/sql/tests/test_conversion.py b/python/pyspark/sql/tests/test_conversion.py
index c5e9fb55aba2..f660ffa0e7d3 100644
--- a/python/pyspark/sql/tests/test_conversion.py
+++ b/python/pyspark/sql/tests/test_conversion.py
@@ -46,6 +46,7 @@ from pyspark.sql.types import (
StringType,
StructField,
StructType,
+ TimestampNTZType,
TimestampType,
UserDefinedType,
VariantType,
@@ -719,6 +720,20 @@ class ConversionTests(unittest.TestCase):
@unittest.skipIf(not have_pyarrow, pyarrow_requirement_message)
class ArrowArrayToPandasConversionTests(unittest.TestCase):
+ def test_convert_numpy_ser_name_survives_preprocess_time(self):
+ # convert_numpy reads the Arrow field name before preprocess_time, because the
+ # pa.compute kernels it runs for timestamps return a new array with no field name.
+ import pyarrow as pa
+
+ for pa_type in [pa.timestamp("us", tz="UTC"), pa.timestamp("s"), pa.timestamp("ns")]:
+ ts = pa.array([datetime.datetime(2020, 6, 15, 12, 30)], type=pa.timestamp("us")).cast(
+ pa_type
+ )
+ col = pa.RecordBatch.from_arrays([ts], ["tscol"]).column(0)
+ spark_type = TimestampType() if pa_type.tz is not None else TimestampNTZType()
+ result = ArrowArrayToPandasConversion.convert_numpy(col, spark_type, timezone="UTC")
+ self.assertEqual(result.name, "tscol", f"name lost for {pa_type}")
+
def test_udt_convert_numpy(self):
import pyarrow as pa
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org