You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2024/02/16 03:42:14 UTC
(spark) branch branch-3.5 updated: [SPARK-47068][PYTHON][TESTS] Recover -1 and 0 case for spark.sql.execution.arrow.maxRecordsPerBatch
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new 1c1c5faa29dc [SPARK-47068][PYTHON][TESTS] Recover -1 and 0 case for spark.sql.execution.arrow.maxRecordsPerBatch
1c1c5faa29dc is described below
commit 1c1c5faa29dc649faf143fe2eea39ccf15862f85
Author: Hyukjin Kwon <gu...@apache.org>
AuthorDate: Fri Feb 16 12:41:19 2024 +0900
[SPARK-47068][PYTHON][TESTS] Recover -1 and 0 case for spark.sql.execution.arrow.maxRecordsPerBatch
This PR fixes the regression introduced by https://github.com/apache/spark/pull/36683.
```python
import pandas as pd
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
spark.conf.set("spark.sql.execution.arrow.maxRecordsPerBatch", 0)
spark.conf.set("spark.sql.execution.arrow.pyspark.fallback.enabled", False)
spark.createDataFrame(pd.DataFrame({'a': [123]})).toPandas()
spark.conf.set("spark.sql.execution.arrow.maxRecordsPerBatch", -1)
spark.createDataFrame(pd.DataFrame({'a': [123]})).toPandas()
```
**Before**
```
/.../spark/python/pyspark/sql/pandas/conversion.py:371: UserWarning: createDataFrame attempted Arrow optimization because 'spark.sql.execution.arrow.pyspark.enabled' is set to true, but has reached the error below and will not continue because automatic fallback with 'spark.sql.execution.arrow.pyspark.fallback.enabled' has been set to false.
range() arg 3 must not be zero
warn(msg)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/.../spark/python/pyspark/sql/session.py", line 1483, in createDataFrame
return super(SparkSession, self).createDataFrame( # type: ignore[call-overload]
File "/.../spark/python/pyspark/sql/pandas/conversion.py", line 351, in createDataFrame
return self._create_from_pandas_with_arrow(data, schema, timezone)
File "/.../spark/python/pyspark/sql/pandas/conversion.py", line 633, in _create_from_pandas_with_arrow
pdf_slices = (pdf.iloc[start : start + step] for start in range(0, len(pdf), step))
ValueError: range() arg 3 must not be zero
```
```
Empty DataFrame
Columns: [a]
Index: []
```
**After**
```
a
0 123
```
```
a
0 123
```
It fixes a regerssion. This is a documented behaviour. It should be backported to branch-3.4 and branch-3.5.
Yes, it fixes a regression as described above.
Unittest was added.
No.
Closes #45132 from HyukjinKwon/SPARK-47068.
Authored-by: Hyukjin Kwon <gu...@apache.org>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
(cherry picked from commit 3bb762dc032866cfb304019cba6db01125556c2f)
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
python/pyspark/sql/pandas/conversion.py | 1 +
python/pyspark/sql/tests/test_arrow.py | 10 ++++++++++
2 files changed, 11 insertions(+)
diff --git a/python/pyspark/sql/pandas/conversion.py b/python/pyspark/sql/pandas/conversion.py
index 8664c4df73ed..3643cafbb3ba 100644
--- a/python/pyspark/sql/pandas/conversion.py
+++ b/python/pyspark/sql/pandas/conversion.py
@@ -613,6 +613,7 @@ class SparkConversionMixin:
# Slice the DataFrame to be batched
step = self._jconf.arrowMaxRecordsPerBatch()
+ step = step if step > 0 else len(pdf)
pdf_slices = (pdf.iloc[start : start + step] for start in range(0, len(pdf), step))
# Create list of Arrow (columns, arrow_type, spark_type) for serializer dump_stream
diff --git a/python/pyspark/sql/tests/test_arrow.py b/python/pyspark/sql/tests/test_arrow.py
index 73b6067373b0..9e9a7d3ac9b0 100644
--- a/python/pyspark/sql/tests/test_arrow.py
+++ b/python/pyspark/sql/tests/test_arrow.py
@@ -1238,6 +1238,16 @@ class ArrowTestsMixin:
df = self.spark.createDataFrame([MyInheritedTuple(1, 2, MyInheritedTuple(1, 2, 3))])
self.assertEqual(df.first(), Row(a=1, b=2, c=Row(a=1, b=2, c=3)))
+ def test_negative_and_zero_batch_size(self):
+ # SPARK-47068: Negative and zero value should work as unlimited batch size.
+ with self.sql_conf({"spark.sql.execution.arrow.maxRecordsPerBatch": 0}):
+ pdf = pd.DataFrame({"a": [123]})
+ assert_frame_equal(pdf, self.spark.createDataFrame(pdf).toPandas())
+
+ with self.sql_conf({"spark.sql.execution.arrow.maxRecordsPerBatch": -1}):
+ pdf = pd.DataFrame({"a": [123]})
+ assert_frame_equal(pdf, self.spark.createDataFrame(pdf).toPandas())
+
@unittest.skipIf(
not have_pandas or not have_pyarrow,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org