4ad9bfd53b
### What changes were proposed in this pull request? This PR aims to drop Python 2.7, 3.4 and 3.5. Roughly speaking, it removes all the widely known Python 2 compatibility workarounds such as `sys.version` comparison, `__future__`. Also, it removes the Python 2 dedicated codes such as `ArrayConstructor` in Spark. ### Why are the changes needed? 1. Unsupport EOL Python versions 2. Reduce maintenance overhead and remove a bit of legacy codes and hacks for Python 2. 3. PyPy2 has a critical bug that causes a flaky test, SPARK-28358 given my testing and investigation. 4. Users can use Python type hints with Pandas UDFs without thinking about Python version 5. Users can leverage one latest cloudpickle, https://github.com/apache/spark/pull/28950. With Python 3.8+ it can also leverage C pickle. ### Does this PR introduce _any_ user-facing change? Yes, users cannot use Python 2.7, 3.4 and 3.5 in the upcoming Spark version. ### How was this patch tested? Manually tested and also tested in Jenkins. Closes #28957 from HyukjinKwon/SPARK-32138. Authored-by: HyukjinKwon <gurwls223@apache.org> Signed-off-by: HyukjinKwon <gurwls223@apache.org>
134 lines
4.2 KiB
Python
134 lines
4.2 KiB
Python
#
|
|
# 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 py4j.protocol
|
|
from py4j.protocol import Py4JJavaError
|
|
from py4j.java_gateway import JavaObject
|
|
from py4j.java_collections import JavaArray, JavaList
|
|
|
|
from pyspark import RDD, SparkContext
|
|
from pyspark.serializers import PickleSerializer, AutoBatchedSerializer
|
|
from pyspark.sql import DataFrame, SQLContext
|
|
|
|
# Hack for support float('inf') in Py4j
|
|
_old_smart_decode = py4j.protocol.smart_decode
|
|
|
|
_float_str_mapping = {
|
|
'nan': 'NaN',
|
|
'inf': 'Infinity',
|
|
'-inf': '-Infinity',
|
|
}
|
|
|
|
|
|
def _new_smart_decode(obj):
|
|
if isinstance(obj, float):
|
|
s = str(obj)
|
|
return _float_str_mapping.get(s, s)
|
|
return _old_smart_decode(obj)
|
|
|
|
py4j.protocol.smart_decode = _new_smart_decode
|
|
|
|
|
|
_picklable_classes = [
|
|
'SparseVector',
|
|
'DenseVector',
|
|
'SparseMatrix',
|
|
'DenseMatrix',
|
|
]
|
|
|
|
|
|
# this will call the ML version of pythonToJava()
|
|
def _to_java_object_rdd(rdd):
|
|
""" Return an JavaRDD of Object by unpickling
|
|
|
|
It will convert each Python object into Java object by Pyrolite, whenever the
|
|
RDD is serialized in batch or not.
|
|
"""
|
|
rdd = rdd._reserialize(AutoBatchedSerializer(PickleSerializer()))
|
|
return rdd.ctx._jvm.org.apache.spark.ml.python.MLSerDe.pythonToJava(rdd._jrdd, True)
|
|
|
|
|
|
def _py2java(sc, obj):
|
|
""" Convert Python object into Java """
|
|
if isinstance(obj, RDD):
|
|
obj = _to_java_object_rdd(obj)
|
|
elif isinstance(obj, DataFrame):
|
|
obj = obj._jdf
|
|
elif isinstance(obj, SparkContext):
|
|
obj = obj._jsc
|
|
elif isinstance(obj, list):
|
|
obj = [_py2java(sc, x) for x in obj]
|
|
elif isinstance(obj, JavaObject):
|
|
pass
|
|
elif isinstance(obj, (int, float, bool, bytes, str)):
|
|
pass
|
|
else:
|
|
data = bytearray(PickleSerializer().dumps(obj))
|
|
obj = sc._jvm.org.apache.spark.ml.python.MLSerDe.loads(data)
|
|
return obj
|
|
|
|
|
|
def _java2py(sc, r, encoding="bytes"):
|
|
if isinstance(r, JavaObject):
|
|
clsName = r.getClass().getSimpleName()
|
|
# convert RDD into JavaRDD
|
|
if clsName != 'JavaRDD' and clsName.endswith("RDD"):
|
|
r = r.toJavaRDD()
|
|
clsName = 'JavaRDD'
|
|
|
|
if clsName == 'JavaRDD':
|
|
jrdd = sc._jvm.org.apache.spark.ml.python.MLSerDe.javaToPython(r)
|
|
return RDD(jrdd, sc)
|
|
|
|
if clsName == 'Dataset':
|
|
return DataFrame(r, SQLContext.getOrCreate(sc))
|
|
|
|
if clsName in _picklable_classes:
|
|
r = sc._jvm.org.apache.spark.ml.python.MLSerDe.dumps(r)
|
|
elif isinstance(r, (JavaArray, JavaList)):
|
|
try:
|
|
r = sc._jvm.org.apache.spark.ml.python.MLSerDe.dumps(r)
|
|
except Py4JJavaError:
|
|
pass # not pickable
|
|
|
|
if isinstance(r, (bytearray, bytes)):
|
|
r = PickleSerializer().loads(bytes(r), encoding=encoding)
|
|
return r
|
|
|
|
|
|
def callJavaFunc(sc, func, *args):
|
|
""" Call Java Function """
|
|
args = [_py2java(sc, a) for a in args]
|
|
return _java2py(sc, func(*args))
|
|
|
|
|
|
def inherit_doc(cls):
|
|
"""
|
|
A decorator that makes a class inherit documentation from its parents.
|
|
"""
|
|
for name, func in vars(cls).items():
|
|
# only inherit docstring for public functions
|
|
if name.startswith("_"):
|
|
continue
|
|
if not func.__doc__:
|
|
for parent in cls.__bases__:
|
|
parent_func = getattr(parent, name, None)
|
|
if parent_func and getattr(parent_func, "__doc__", None):
|
|
func.__doc__ = parent_func.__doc__
|
|
break
|
|
return cls
|