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 2018/05/30 16:46:17 UTC
spark git commit: [SPARK-23754][BRANCH-2.3][PYTHON] Re-raising
StopIteration in client code
Repository: spark
Updated Branches:
refs/heads/branch-2.3 66289a3e0 -> e1c0ab16c
[SPARK-23754][BRANCH-2.3][PYTHON] Re-raising StopIteration in client code
## What changes are proposed
Make sure that `StopIteration`s raised in users' code do not silently interrupt processing by spark, but are raised as exceptions to the users. The users' functions are wrapped in `safe_iter` (in `shuffle.py`), which re-raises `StopIteration`s as `RuntimeError`s
## How were the changes tested
Unit tests, making sure that the exceptions are indeed raised. I am not sure how to check whether a `Py4JJavaError` contains my exception, so I simply looked for the exception message in the java exception's `toString`. Can you propose a better way?
This is my original work, licensed in the same way as spark
---
Author: e-dorigatti <emilio.dorigattigmail.com>
Closes #21383 from e-dorigatti/fix_spark_23754.
(cherry picked from commit 0ebb0c0d4dd3e192464dc5e0e6f01efa55b945ed)
Author: e-dorigatti <em...@gmail.com>
Closes #21463 from e-dorigatti/branch-2.3.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/e1c0ab16
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/e1c0ab16
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/e1c0ab16
Branch: refs/heads/branch-2.3
Commit: e1c0ab16c71f102bfd9f5133647d168e49ae06bc
Parents: 66289a3
Author: e-dorigatti <em...@gmail.com>
Authored: Thu May 31 00:46:09 2018 +0800
Committer: hyukjinkwon <gu...@apache.org>
Committed: Thu May 31 00:46:09 2018 +0800
----------------------------------------------------------------------
python/pyspark/rdd.py | 18 +++++++++++---
python/pyspark/shuffle.py | 7 +++---
python/pyspark/sql/tests.py | 16 ++++++++++++
python/pyspark/sql/udf.py | 4 ++-
python/pyspark/tests.py | 53 ++++++++++++++++++++++++++++++++++++++++
python/pyspark/util.py | 17 +++++++++++++
6 files changed, 108 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/e1c0ab16/python/pyspark/rdd.py
----------------------------------------------------------------------
diff --git a/python/pyspark/rdd.py b/python/pyspark/rdd.py
index acd0fa2..8dc30a4 100644
--- a/python/pyspark/rdd.py
+++ b/python/pyspark/rdd.py
@@ -53,6 +53,7 @@ from pyspark.resultiterable import ResultIterable
from pyspark.shuffle import Aggregator, ExternalMerger, \
get_used_memory, ExternalSorter, ExternalGroupBy
from pyspark.traceback_utils import SCCallSiteSync
+from pyspark.util import fail_on_stopiteration
__all__ = ["RDD"]
@@ -338,7 +339,7 @@ class RDD(object):
[('a', 1), ('b', 1), ('c', 1)]
"""
def func(_, iterator):
- return map(f, iterator)
+ return map(fail_on_stopiteration(f), iterator)
return self.mapPartitionsWithIndex(func, preservesPartitioning)
def flatMap(self, f, preservesPartitioning=False):
@@ -353,7 +354,7 @@ class RDD(object):
[(2, 2), (2, 2), (3, 3), (3, 3), (4, 4), (4, 4)]
"""
def func(s, iterator):
- return chain.from_iterable(map(f, iterator))
+ return chain.from_iterable(map(fail_on_stopiteration(f), iterator))
return self.mapPartitionsWithIndex(func, preservesPartitioning)
def mapPartitions(self, f, preservesPartitioning=False):
@@ -416,7 +417,7 @@ class RDD(object):
[2, 4]
"""
def func(iterator):
- return filter(f, iterator)
+ return filter(fail_on_stopiteration(f), iterator)
return self.mapPartitions(func, True)
def distinct(self, numPartitions=None):
@@ -797,6 +798,8 @@ class RDD(object):
>>> def f(x): print(x)
>>> sc.parallelize([1, 2, 3, 4, 5]).foreach(f)
"""
+ f = fail_on_stopiteration(f)
+
def processPartition(iterator):
for x in iterator:
f(x)
@@ -846,6 +849,8 @@ class RDD(object):
...
ValueError: Can not reduce() empty RDD
"""
+ f = fail_on_stopiteration(f)
+
def func(iterator):
iterator = iter(iterator)
try:
@@ -917,6 +922,8 @@ class RDD(object):
>>> sc.parallelize([1, 2, 3, 4, 5]).fold(0, add)
15
"""
+ op = fail_on_stopiteration(op)
+
def func(iterator):
acc = zeroValue
for obj in iterator:
@@ -949,6 +956,9 @@ class RDD(object):
>>> sc.parallelize([]).aggregate((0, 0), seqOp, combOp)
(0, 0)
"""
+ seqOp = fail_on_stopiteration(seqOp)
+ combOp = fail_on_stopiteration(combOp)
+
def func(iterator):
acc = zeroValue
for obj in iterator:
@@ -1642,6 +1652,8 @@ class RDD(object):
>>> sorted(rdd.reduceByKeyLocally(add).items())
[('a', 2), ('b', 1)]
"""
+ func = fail_on_stopiteration(func)
+
def reducePartition(iterator):
m = {}
for k, v in iterator:
http://git-wip-us.apache.org/repos/asf/spark/blob/e1c0ab16/python/pyspark/shuffle.py
----------------------------------------------------------------------
diff --git a/python/pyspark/shuffle.py b/python/pyspark/shuffle.py
index e974cda..68f9df7 100644
--- a/python/pyspark/shuffle.py
+++ b/python/pyspark/shuffle.py
@@ -27,6 +27,7 @@ import random
import pyspark.heapq3 as heapq
from pyspark.serializers import BatchedSerializer, PickleSerializer, FlattenedValuesSerializer, \
CompressedSerializer, AutoBatchedSerializer
+from pyspark.util import fail_on_stopiteration
try:
@@ -93,9 +94,9 @@ class Aggregator(object):
"""
def __init__(self, createCombiner, mergeValue, mergeCombiners):
- self.createCombiner = createCombiner
- self.mergeValue = mergeValue
- self.mergeCombiners = mergeCombiners
+ self.createCombiner = fail_on_stopiteration(createCombiner)
+ self.mergeValue = fail_on_stopiteration(mergeValue)
+ self.mergeCombiners = fail_on_stopiteration(mergeCombiners)
class SimpleAggregator(Aggregator):
http://git-wip-us.apache.org/repos/asf/spark/blob/e1c0ab16/python/pyspark/sql/tests.py
----------------------------------------------------------------------
diff --git a/python/pyspark/sql/tests.py b/python/pyspark/sql/tests.py
index 7ff09af..818ba83 100644
--- a/python/pyspark/sql/tests.py
+++ b/python/pyspark/sql/tests.py
@@ -853,6 +853,22 @@ class SQLTests(ReusedSQLTestCase):
self.assertEqual(f, f_.func)
self.assertEqual(return_type, f_.returnType)
+ def test_stopiteration_in_udf(self):
+ # test for SPARK-23754
+ from pyspark.sql.functions import udf
+ from py4j.protocol import Py4JJavaError
+
+ def foo(x):
+ raise StopIteration()
+
+ with self.assertRaises(Py4JJavaError) as cm:
+ self.spark.range(0, 1000).withColumn('v', udf(foo)('id')).show()
+
+ self.assertIn(
+ "Caught StopIteration thrown from user's code; failing the task",
+ cm.exception.java_exception.toString()
+ )
+
def test_validate_column_types(self):
from pyspark.sql.functions import udf, to_json
from pyspark.sql.column import _to_java_column
http://git-wip-us.apache.org/repos/asf/spark/blob/e1c0ab16/python/pyspark/sql/udf.py
----------------------------------------------------------------------
diff --git a/python/pyspark/sql/udf.py b/python/pyspark/sql/udf.py
index 671e568..7d813af 100644
--- a/python/pyspark/sql/udf.py
+++ b/python/pyspark/sql/udf.py
@@ -24,6 +24,7 @@ from pyspark.rdd import _prepare_for_python_RDD, PythonEvalType, ignore_unicode_
from pyspark.sql.column import Column, _to_java_column, _to_seq
from pyspark.sql.types import StringType, DataType, StructType, _parse_datatype_string, \
to_arrow_type, to_arrow_schema
+from pyspark.util import fail_on_stopiteration
__all__ = ["UDFRegistration"]
@@ -154,7 +155,8 @@ class UserDefinedFunction(object):
spark = SparkSession.builder.getOrCreate()
sc = spark.sparkContext
- wrapped_func = _wrap_function(sc, self.func, self.returnType)
+ func = fail_on_stopiteration(self.func)
+ wrapped_func = _wrap_function(sc, func, self.returnType)
jdt = spark._jsparkSession.parseDataType(self.returnType.json())
judf = sc._jvm.org.apache.spark.sql.execution.python.UserDefinedPythonFunction(
self._name, wrapped_func, jdt, self.evalType, self.deterministic)
http://git-wip-us.apache.org/repos/asf/spark/blob/e1c0ab16/python/pyspark/tests.py
----------------------------------------------------------------------
diff --git a/python/pyspark/tests.py b/python/pyspark/tests.py
index 18bbeb5..af39450 100644
--- a/python/pyspark/tests.py
+++ b/python/pyspark/tests.py
@@ -161,6 +161,37 @@ class MergerTests(unittest.TestCase):
self.assertEqual(k, len(vs))
self.assertEqual(list(range(k)), list(vs))
+ def test_stopiteration_is_raised(self):
+
+ def stopit(*args, **kwargs):
+ raise StopIteration()
+
+ def legit_create_combiner(x):
+ return [x]
+
+ def legit_merge_value(x, y):
+ return x.append(y) or x
+
+ def legit_merge_combiners(x, y):
+ return x.extend(y) or x
+
+ data = [(x % 2, x) for x in range(100)]
+
+ # wrong create combiner
+ m = ExternalMerger(Aggregator(stopit, legit_merge_value, legit_merge_combiners), 20)
+ with self.assertRaises((Py4JJavaError, RuntimeError)) as cm:
+ m.mergeValues(data)
+
+ # wrong merge value
+ m = ExternalMerger(Aggregator(legit_create_combiner, stopit, legit_merge_combiners), 20)
+ with self.assertRaises((Py4JJavaError, RuntimeError)) as cm:
+ m.mergeValues(data)
+
+ # wrong merge combiners
+ m = ExternalMerger(Aggregator(legit_create_combiner, legit_merge_value, stopit), 20)
+ with self.assertRaises((Py4JJavaError, RuntimeError)) as cm:
+ m.mergeCombiners(map(lambda x_y1: (x_y1[0], [x_y1[1]]), data))
+
class SorterTests(unittest.TestCase):
def test_in_memory_sort(self):
@@ -1239,6 +1270,28 @@ class RDDTests(ReusedPySparkTestCase):
self.assertRaises(Py4JJavaError, rdd.pipe('grep 4', checkCode=True).collect)
self.assertEqual([], rdd.pipe('grep 4').collect())
+ def test_stopiteration_in_client_code(self):
+
+ def stopit(*x):
+ raise StopIteration()
+
+ seq_rdd = self.sc.parallelize(range(10))
+ keyed_rdd = self.sc.parallelize((x % 2, x) for x in range(10))
+
+ self.assertRaises(Py4JJavaError, seq_rdd.map(stopit).collect)
+ self.assertRaises(Py4JJavaError, seq_rdd.filter(stopit).collect)
+ self.assertRaises(Py4JJavaError, seq_rdd.cartesian(seq_rdd).flatMap(stopit).collect)
+ self.assertRaises(Py4JJavaError, seq_rdd.foreach, stopit)
+ self.assertRaises(Py4JJavaError, keyed_rdd.reduceByKeyLocally, stopit)
+ self.assertRaises(Py4JJavaError, seq_rdd.reduce, stopit)
+ self.assertRaises(Py4JJavaError, seq_rdd.fold, 0, stopit)
+
+ # the exception raised is non-deterministic
+ self.assertRaises((Py4JJavaError, RuntimeError),
+ seq_rdd.aggregate, 0, stopit, lambda *x: 1)
+ self.assertRaises((Py4JJavaError, RuntimeError),
+ seq_rdd.aggregate, 0, lambda *x: 1, stopit)
+
class ProfilerTests(PySparkTestCase):
http://git-wip-us.apache.org/repos/asf/spark/blob/e1c0ab16/python/pyspark/util.py
----------------------------------------------------------------------
diff --git a/python/pyspark/util.py b/python/pyspark/util.py
index ad4a0bc..83d528f 100644
--- a/python/pyspark/util.py
+++ b/python/pyspark/util.py
@@ -45,6 +45,23 @@ def _exception_message(excp):
return str(excp)
+def fail_on_stopiteration(f):
+ """
+ Wraps the input function to fail on 'StopIteration' by raising a 'RuntimeError'
+ prevents silent loss of data when 'f' is used in a for loop
+ """
+ def wrapper(*args, **kwargs):
+ try:
+ return f(*args, **kwargs)
+ except StopIteration as exc:
+ raise RuntimeError(
+ "Caught StopIteration thrown from user's code; failing the task",
+ exc
+ )
+
+ return wrapper
+
+
if __name__ == "__main__":
import doctest
(failure_count, test_count) = doctest.testmod()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org