|
#
# 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 numpy as np
from pyspark import keyword_only
from pyspark.ml import Estimator, Model, Transformer, UnaryTransformer
from pyspark.ml.evaluation import Evaluator
from pyspark.ml.param import Param, Params, TypeConverters
from pyspark.ml.param.shared import HasMaxIter, HasRegParam
from pyspark.ml.classification import Classifier, ClassificationModel
from pyspark.ml.util import DefaultParamsReadable, DefaultParamsWritable
from pyspark.ml.wrapper import _java2py # type: ignore
from pyspark.sql import DataFrame, SparkSession
from pyspark.sql.types import DoubleType
from pyspark.testing.utils import ReusedPySparkTestCase as PySparkTestCase
def check_params(test_self, py_stage, check_params_exist=True):
"""
Checks common requirements for :py:class:`PySpark.ml.Params.params`:
- set of params exist in Java and Python and are ordered by names
- param parent has the same UID as the object's UID
- default param value from Java matches value in Python
- optionally check if all params from Java also exist in Python
"""
py_stage_str = "%s %s" % (type(py_stage), py_stage)
43 ↛ 44line 43 didn't jump to line 44, because the condition on line 43 was never true if not hasattr(py_stage, "_to_java"):
return
java_stage = py_stage._to_java()
46 ↛ 47line 46 didn't jump to line 47, because the condition on line 46 was never true if java_stage is None:
return
test_self.assertEqual(py_stage.uid, java_stage.uid(), msg=py_stage_str)
49 ↛ 57line 49 didn't jump to line 57, because the condition on line 49 was never false if check_params_exist:
param_names = [p.name for p in py_stage.params]
java_params = list(java_stage.params())
java_param_names = [jp.name() for jp in java_params]
test_self.assertEqual(
param_names, sorted(java_param_names),
"Param list in Python does not match Java for %s:\nJava = %s\nPython = %s"
% (py_stage_str, java_param_names, param_names))
for p in py_stage.params:
test_self.assertEqual(p.parent, py_stage.uid)
java_param = java_stage.getParam(p.name)
py_has_default = py_stage.hasDefault(p)
java_has_default = java_stage.hasDefault(java_param)
test_self.assertEqual(py_has_default, java_has_default,
"Default value mismatch of param %s for Params %s"
% (p.name, str(py_stage)))
if py_has_default:
if p.name == "seed":
continue # Random seeds between Spark and PySpark are different
java_default = _java2py(test_self.sc,
java_stage.clear(java_param).getOrDefault(java_param))
py_stage.clear(p)
py_default = py_stage.getOrDefault(p)
# equality test for NaN is always False
if isinstance(java_default, float) and np.isnan(java_default):
java_default = "NaN"
py_default = "NaN" if np.isnan(py_default) else "not NaN"
test_self.assertEqual(
java_default, py_default,
"Java default %s != python default %s of param %s for Params %s"
% (str(java_default), str(py_default), p.name, str(py_stage)))
class SparkSessionTestCase(PySparkTestCase):
@classmethod
def setUpClass(cls):
PySparkTestCase.setUpClass()
cls.spark = SparkSession(cls.sc)
@classmethod
def tearDownClass(cls):
PySparkTestCase.tearDownClass()
cls.spark.stop()
class MockDataset(DataFrame):
def __init__(self):
self.index = 0
class HasFake(Params):
def __init__(self):
super(HasFake, self).__init__()
self.fake = Param(self, "fake", "fake param")
def getFake(self):
return self.getOrDefault(self.fake)
class MockTransformer(Transformer, HasFake):
def __init__(self):
super(MockTransformer, self).__init__()
self.dataset_index = None
def _transform(self, dataset):
self.dataset_index = dataset.index
dataset.index += 1
return dataset
class MockUnaryTransformer(UnaryTransformer, DefaultParamsReadable, DefaultParamsWritable):
shift = Param(Params._dummy(), # type: ignore
"shift", "The amount by which to shift " +
"data in a DataFrame",
typeConverter=TypeConverters.toFloat)
def __init__(self, shiftVal=1):
super(MockUnaryTransformer, self).__init__()
self._setDefault(shift=1)
self._set(shift=shiftVal)
def getShift(self):
return self.getOrDefault(self.shift)
def setShift(self, shift):
self._set(shift=shift)
def createTransformFunc(self):
shiftVal = self.getShift()
return lambda x: x + shiftVal
def outputDataType(self):
return DoubleType()
def validateInputType(self, inputType):
if inputType != DoubleType():
raise TypeError("Bad input type: {}. ".format(inputType) +
"Requires Double.")
class MockEstimator(Estimator, HasFake):
def __init__(self):
super(MockEstimator, self).__init__()
self.dataset_index = None
def _fit(self, dataset):
self.dataset_index = dataset.index
model = MockModel()
self._copyValues(model)
return model
class MockModel(MockTransformer, Model, HasFake):
pass
class _DummyLogisticRegressionParams(HasMaxIter, HasRegParam):
def setMaxIter(self, value):
return self._set(maxIter=value)
def setRegParam(self, value):
return self._set(regParam=value)
# This is a dummy LogisticRegression used in test for python backend estimator/model
class DummyLogisticRegression(Classifier, _DummyLogisticRegressionParams,
DefaultParamsReadable, DefaultParamsWritable):
@keyword_only
def __init__(self, *, featuresCol="features", labelCol="label", predictionCol="prediction",
maxIter=100, regParam=0.0, rawPredictionCol="rawPrediction"):
super(DummyLogisticRegression, self).__init__()
kwargs = self._input_kwargs
self.setParams(**kwargs)
@keyword_only
def setParams(self, *, featuresCol="features", labelCol="label", predictionCol="prediction",
maxIter=100, regParam=0.0, rawPredictionCol="rawPrediction"):
kwargs = self._input_kwargs
self._set(**kwargs)
return self
def _fit(self, dataset):
# Do nothing but create a dummy model
return self._copyValues(DummyLogisticRegressionModel())
class DummyLogisticRegressionModel(ClassificationModel, _DummyLogisticRegressionParams,
DefaultParamsReadable, DefaultParamsWritable):
def __init__(self):
super(DummyLogisticRegressionModel, self).__init__()
def _transform(self, dataset):
# A dummy transform impl which always predict label 1
from pyspark.sql.functions import array, lit
from pyspark.ml.functions import array_to_vector
rawPredCol = self.getRawPredictionCol()
211 ↛ 214line 211 didn't jump to line 214, because the condition on line 211 was never false if rawPredCol:
dataset = dataset.withColumn(
rawPredCol, array_to_vector(array(lit(-100.0), lit(100.0))))
predCol = self.getPredictionCol()
215 ↛ 218line 215 didn't jump to line 218, because the condition on line 215 was never false if predCol:
dataset = dataset.withColumn(predCol, lit(1.0))
return dataset
@property
def numClasses(self):
# a dummy implementation for test.
return 2
@property
def intercept(self):
# a dummy implementation for test.
return 0.0
# This class only used in test. The following methods/properties are not used in tests.
@property
def coefficients(self):
raise NotImplementedError()
def predictRaw(self, value):
raise NotImplementedError()
def numFeatures(self):
raise NotImplementedError()
def predict(self, value):
raise NotImplementedError()
class DummyEvaluator(Evaluator, DefaultParamsReadable, DefaultParamsWritable):
def _evaluate(self, dataset):
# a dummy implementation for test.
return 1.0
|
