3e5b4ae63a
## What changes were proposed in this pull request? SPARK-23754 was fixed in #21383 by changing the UDF code to wrap the user function, but this required a hack to save its argspec. This PR reverts this change and fixes the `StopIteration` bug in the worker ## How does this work? The root of the problem is that when an user-supplied function raises a `StopIteration`, pyspark might stop processing data, if this function is used in a for-loop. The solution is to catch `StopIteration`s exceptions and re-raise them as `RuntimeError`s, so that the execution fails and the error is reported to the user. This is done using the `fail_on_stopiteration` wrapper, in different ways depending on where the function is used: - In RDDs, the user function is wrapped in the driver, because this function is also called in the driver itself. - In SQL UDFs, the function is wrapped in the worker, since all processing happens there. Moreover, the worker needs the signature of the user function, which is lost when wrapping it, but passing this signature to the worker requires a not so nice hack. ## How was this patch tested? Same tests, plus tests for pandas UDFs Author: edorigatti <emilio.dorigatti@gmail.com> Closes #21467 from e-dorigatti/fix_udf_hack.
114 lines
3.7 KiB
Python
114 lines
3.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
#
|
|
# 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.
|
|
#
|
|
|
|
import re
|
|
import sys
|
|
import inspect
|
|
from py4j.protocol import Py4JJavaError
|
|
|
|
__all__ = []
|
|
|
|
|
|
def _exception_message(excp):
|
|
"""Return the message from an exception as either a str or unicode object. Supports both
|
|
Python 2 and Python 3.
|
|
|
|
>>> msg = "Exception message"
|
|
>>> excp = Exception(msg)
|
|
>>> msg == _exception_message(excp)
|
|
True
|
|
|
|
>>> msg = u"unicöde"
|
|
>>> excp = Exception(msg)
|
|
>>> msg == _exception_message(excp)
|
|
True
|
|
"""
|
|
if isinstance(excp, Py4JJavaError):
|
|
# 'Py4JJavaError' doesn't contain the stack trace available on the Java side in 'message'
|
|
# attribute in Python 2. We should call 'str' function on this exception in general but
|
|
# 'Py4JJavaError' has an issue about addressing non-ascii strings. So, here we work
|
|
# around by the direct call, '__str__()'. Please see SPARK-23517.
|
|
return excp.__str__()
|
|
if hasattr(excp, "message"):
|
|
return excp.message
|
|
return str(excp)
|
|
|
|
|
|
def _get_argspec(f):
|
|
"""
|
|
Get argspec of a function. Supports both Python 2 and Python 3.
|
|
"""
|
|
if sys.version_info[0] < 3:
|
|
argspec = inspect.getargspec(f)
|
|
else:
|
|
# `getargspec` is deprecated since python3.0 (incompatible with function annotations).
|
|
# See SPARK-23569.
|
|
argspec = inspect.getfullargspec(f)
|
|
return argspec
|
|
|
|
|
|
class VersionUtils(object):
|
|
"""
|
|
Provides utility method to determine Spark versions with given input string.
|
|
"""
|
|
@staticmethod
|
|
def majorMinorVersion(sparkVersion):
|
|
"""
|
|
Given a Spark version string, return the (major version number, minor version number).
|
|
E.g., for 2.0.1-SNAPSHOT, return (2, 0).
|
|
|
|
>>> sparkVersion = "2.4.0"
|
|
>>> VersionUtils.majorMinorVersion(sparkVersion)
|
|
(2, 4)
|
|
>>> sparkVersion = "2.3.0-SNAPSHOT"
|
|
>>> VersionUtils.majorMinorVersion(sparkVersion)
|
|
(2, 3)
|
|
|
|
"""
|
|
m = re.search('^(\d+)\.(\d+)(\..*)?$', sparkVersion)
|
|
if m is not None:
|
|
return (int(m.group(1)), int(m.group(2)))
|
|
else:
|
|
raise ValueError("Spark tried to parse '%s' as a Spark" % sparkVersion +
|
|
" version string, but it could not find the major and minor" +
|
|
" version numbers.")
|
|
|
|
|
|
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 in Spark code
|
|
"""
|
|
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()
|
|
if failure_count:
|
|
sys.exit(-1)
|