This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/spark.git
commit 514d1899a1baf7c1bb5af68aa05e3886a80a0843
Author: Hyukjin Kwon <gurwls...@apache.org>
AuthorDate: Tue May 3 08:33:22 2022 +0900
Revert "[SPARK-39084][PYSPARK] Fix df.rdd.isEmpty() by using TaskContext to
stop iterator on task completion"
This reverts commit 4dba99ae359b07f814f68707073414f60616b564.
---
python/pyspark/sql/tests/test_dataframe.py | 38 +---------------------
.../sql/execution/python/EvaluatePython.scala | 3 +-
2 files changed, 2 insertions(+), 39 deletions(-)
diff --git a/python/pyspark/sql/tests/test_dataframe.py
b/python/pyspark/sql/tests/test_dataframe.py
index f2826e29d36..8c9f3304e00 100644
--- a/python/pyspark/sql/tests/test_dataframe.py
+++ b/python/pyspark/sql/tests/test_dataframe.py
@@ -20,8 +20,7 @@ import pydoc
import shutil
import tempfile
import time
-import uuid
-from typing import cast
+import unittest
from pyspark.sql import SparkSession, Row
from pyspark.sql.types import StringType, IntegerType, DoubleType, StructType,
StructField, \
@@ -874,41 +873,6 @@ class DataFrameTests(ReusedSQLTestCase):
with self.assertRaisesRegex(TypeError, "Parameter 'truncate=foo'"):
df.show(truncate='foo')
- def test_df_is_empty(self):
- # SPARK-39084: Fix df.rdd.isEmpty() resulting in JVM crash.
-
- # This particular example of DataFrame reproduces an issue in isEmpty
call
- # which could result in JVM crash.
- data = []
- for t in range(0, 10000):
- id = str(uuid.uuid4())
- if t == 0:
- for i in range(0, 99):
- data.append((id,))
- elif t < 10:
- for i in range(0, 75):
- data.append((id,))
- elif t < 100:
- for i in range(0, 50):
- data.append((id,))
- elif t < 1000:
- for i in range(0, 25):
- data.append((id,))
- else:
- for i in range(0, 10):
- data.append((id,))
-
- tmpPath = tempfile.mkdtemp()
- shutil.rmtree(tmpPath)
- try:
- df = self.spark.createDataFrame(data, ["col"])
- df.coalesce(1).write.parquet(tmpPath)
-
- res = self.spark.read.parquet(tmpPath).groupBy("col").count()
- self.assertFalse(res.rdd.isEmpty())
- finally:
- shutil.rmtree(tmpPath)
-
@unittest.skipIf(
not have_pandas or not have_pyarrow,
pandas_requirement_message or pyarrow_requirement_message) # type:
ignore
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala
index 667f2c030d5..4885f631138 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala
@@ -24,7 +24,6 @@ import scala.collection.JavaConverters._
import net.razorvine.pickle.{IObjectPickler, Opcodes, Pickler}
-import org.apache.spark.{ContextAwareIterator, TaskContext}
import org.apache.spark.api.python.SerDeUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
@@ -301,7 +300,7 @@ object EvaluatePython {
def javaToPython(rdd: RDD[Any]): RDD[Array[Byte]] = {
rdd.mapPartitions { iter =>
registerPicklers() // let it called in executor
- new SerDeUtil.AutoBatchedPickler(new
ContextAwareIterator(TaskContext.get, iter))
+ new SerDeUtil.AutoBatchedPickler(iter)
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
