Yicong-Huang commented on code in PR #57099:
URL: https://github.com/apache/spark/pull/57099#discussion_r3546858615
##########
python/benchmarks/bench_arrow.py:
##########
@@ -114,3 +114,43 @@ def time_long_with_nulls_to_pandas_ext(self, n_rows,
method):
def peakmem_long_with_nulls_to_pandas_ext(self, n_rows, method):
self.run_long_with_nulls_to_pandas_ext(n_rows, method)
+
+
+class ArrowListColumnToRowsBenchmark:
+ """
+ Benchmark for converting Arrow list-typed columns to Python rows, the hot
+ path of Arrow-optimized Python UDF inputs and Spark Connect collect().
+
+ ``baseline`` measures plain ``column.to_pylist()``; ``bulk`` measures
+ ``ArrowTableToRowsConversion._to_pylist`` (see apache/arrow#50326).
+ """
+
+ params = [
+ [100000, 1000000],
+ ["baseline", "bulk"],
+ ]
+ param_names = ["n_rows", "method"]
+
+ def setup(self, n_rows, method):
+ from pyspark.sql.conversion import ArrowTableToRowsConversion
+
+ self.list_of_strings = pa.array(
+ [[f"s{i}", f"t{i}"] for i in range(n_rows)],
type=pa.list_(pa.string())
+ )
+ self.nested_ints_with_nulls = pa.array(
+ [[[i, i + 1], None, [i + 2]] if i % 10 != 0 else None for i in
range(n_rows)],
+ type=pa.list_(pa.list_(pa.int32())),
+ )
+ if method == "bulk":
+ self.convert = ArrowTableToRowsConversion._to_pylist
+ else:
+ self.convert = lambda column: column.to_pylist()
+
+ def time_list_of_strings_to_rows(self, n_rows, method):
+ self.convert(self.list_of_strings)
+
+ def time_nested_ints_with_nulls_to_rows(self, n_rows, method):
+ self.convert(self.nested_ints_with_nulls)
+
+ def peakmem_list_of_strings_to_rows(self, n_rows, method):
Review Comment:
Could we add a peakmem benchmark for the nested list case, and ideally one
nested `array<struct<...>>` like case too?
The new path materializes the flattened child values first and do slices to
create rows. this is supposed to cause the memory usage higher.
##########
python/pyspark/sql/conversion.py:
##########
@@ -980,6 +980,63 @@ class ArrowTableToRowsConversion:
Conversion from Arrow Table to Rows.
"""
+ @staticmethod
+ def _to_pylist(column: Union["pa.Array", "pa.ChunkedArray"]) -> List[Any]:
+ """
+ Equivalent to ``column.to_pylist()``, but converts (nested) list
columns in bulk
+ instead of one scalar at a time.
+
+ ``Array.to_pylist()`` materializes one Scalar per element; for list
types each row
+ additionally allocates a C++ scalar, a Python Scalar wrapper and a
Python Array
+ wrapper for the row's values before converting elements one by one,
which is
+ several times slower than converting the flattened child values in a
single pass
+ and slicing the resulting Python list per row (see
apache/arrow#50326). The values
+ themselves are still converted by Arrow's own ``to_pylist``, so
results are exactly
+ identical: ``None`` stays ``None`` and values inside numeric lists
stay Python ints,
+ unlike a pandas round trip which would coerce them to floats/NaN.
NumPy is used
+ only for the offsets (non-null integers) and the validity bitmap
(booleans), so no
+ value coercion can occur.
+
+ This can be removed once the minimum supported PyArrow version
includes the fix
+ for apache/arrow#50326.
+ """
+ import pyarrow as pa
+ import pyarrow.types as pa_types
+
+ try:
+ import numpy # noqa: F401
+ except ImportError:
+ return column.to_pylist()
Review Comment:
minor: since we are recursively calling this method, this try except and
import can be executed multiple times. do you think we can improve here to
treat with or without numpy as a argument, instead of try to import it as
detection in every invocation?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
