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/11/18 08:02:22 UTC
[1/4] spark git commit: [SPARK-26033][PYTHON][TESTS] Break large
ml/tests.py file into smaller files
Repository: spark
Updated Branches:
refs/heads/master e00cac989 -> 034ae305c
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/test_persistence.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/test_persistence.py b/python/pyspark/ml/tests/test_persistence.py
new file mode 100644
index 0000000..b5a2e16
--- /dev/null
+++ b/python/pyspark/ml/tests/test_persistence.py
@@ -0,0 +1,369 @@
+#
+# 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 json
+from shutil import rmtree
+import sys
+import tempfile
+if sys.version_info[:2] <= (2, 6):
+ try:
+ import unittest2 as unittest
+ except ImportError:
+ sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+ sys.exit(1)
+else:
+ import unittest
+
+from pyspark.ml import Transformer
+from pyspark.ml.classification import DecisionTreeClassifier, LogisticRegression, OneVsRest, \
+ OneVsRestModel
+from pyspark.ml.feature import Binarizer, HashingTF, PCA
+from pyspark.ml.linalg import Vectors
+from pyspark.ml.param import Params
+from pyspark.ml.pipeline import Pipeline, PipelineModel
+from pyspark.ml.regression import DecisionTreeRegressor, LinearRegression
+from pyspark.ml.util import DefaultParamsReadable, DefaultParamsWriter
+from pyspark.ml.wrapper import JavaParams
+from pyspark.testing.mlutils import MockUnaryTransformer, SparkSessionTestCase
+
+
+class PersistenceTest(SparkSessionTestCase):
+
+ def test_linear_regression(self):
+ lr = LinearRegression(maxIter=1)
+ path = tempfile.mkdtemp()
+ lr_path = path + "/lr"
+ lr.save(lr_path)
+ lr2 = LinearRegression.load(lr_path)
+ self.assertEqual(lr.uid, lr2.uid)
+ self.assertEqual(type(lr.uid), type(lr2.uid))
+ self.assertEqual(lr2.uid, lr2.maxIter.parent,
+ "Loaded LinearRegression instance uid (%s) did not match Param's uid (%s)"
+ % (lr2.uid, lr2.maxIter.parent))
+ self.assertEqual(lr._defaultParamMap[lr.maxIter], lr2._defaultParamMap[lr2.maxIter],
+ "Loaded LinearRegression instance default params did not match " +
+ "original defaults")
+ try:
+ rmtree(path)
+ except OSError:
+ pass
+
+ def test_linear_regression_pmml_basic(self):
+ # Most of the validation is done in the Scala side, here we just check
+ # that we output text rather than parquet (e.g. that the format flag
+ # was respected).
+ df = self.spark.createDataFrame([(1.0, 2.0, Vectors.dense(1.0)),
+ (0.0, 2.0, Vectors.sparse(1, [], []))],
+ ["label", "weight", "features"])
+ lr = LinearRegression(maxIter=1)
+ model = lr.fit(df)
+ path = tempfile.mkdtemp()
+ lr_path = path + "/lr-pmml"
+ model.write().format("pmml").save(lr_path)
+ pmml_text_list = self.sc.textFile(lr_path).collect()
+ pmml_text = "\n".join(pmml_text_list)
+ self.assertIn("Apache Spark", pmml_text)
+ self.assertIn("PMML", pmml_text)
+
+ def test_logistic_regression(self):
+ lr = LogisticRegression(maxIter=1)
+ path = tempfile.mkdtemp()
+ lr_path = path + "/logreg"
+ lr.save(lr_path)
+ lr2 = LogisticRegression.load(lr_path)
+ self.assertEqual(lr2.uid, lr2.maxIter.parent,
+ "Loaded LogisticRegression instance uid (%s) "
+ "did not match Param's uid (%s)"
+ % (lr2.uid, lr2.maxIter.parent))
+ self.assertEqual(lr._defaultParamMap[lr.maxIter], lr2._defaultParamMap[lr2.maxIter],
+ "Loaded LogisticRegression instance default params did not match " +
+ "original defaults")
+ try:
+ rmtree(path)
+ except OSError:
+ pass
+
+ def _compare_params(self, m1, m2, param):
+ """
+ Compare 2 ML Params instances for the given param, and assert both have the same param value
+ and parent. The param must be a parameter of m1.
+ """
+ # Prevent key not found error in case of some param in neither paramMap nor defaultParamMap.
+ if m1.isDefined(param):
+ paramValue1 = m1.getOrDefault(param)
+ paramValue2 = m2.getOrDefault(m2.getParam(param.name))
+ if isinstance(paramValue1, Params):
+ self._compare_pipelines(paramValue1, paramValue2)
+ else:
+ self.assertEqual(paramValue1, paramValue2) # for general types param
+ # Assert parents are equal
+ self.assertEqual(param.parent, m2.getParam(param.name).parent)
+ else:
+ # If m1 is not defined param, then m2 should not, too. See SPARK-14931.
+ self.assertFalse(m2.isDefined(m2.getParam(param.name)))
+
+ def _compare_pipelines(self, m1, m2):
+ """
+ Compare 2 ML types, asserting that they are equivalent.
+ This currently supports:
+ - basic types
+ - Pipeline, PipelineModel
+ - OneVsRest, OneVsRestModel
+ This checks:
+ - uid
+ - type
+ - Param values and parents
+ """
+ self.assertEqual(m1.uid, m2.uid)
+ self.assertEqual(type(m1), type(m2))
+ if isinstance(m1, JavaParams) or isinstance(m1, Transformer):
+ self.assertEqual(len(m1.params), len(m2.params))
+ for p in m1.params:
+ self._compare_params(m1, m2, p)
+ elif isinstance(m1, Pipeline):
+ self.assertEqual(len(m1.getStages()), len(m2.getStages()))
+ for s1, s2 in zip(m1.getStages(), m2.getStages()):
+ self._compare_pipelines(s1, s2)
+ elif isinstance(m1, PipelineModel):
+ self.assertEqual(len(m1.stages), len(m2.stages))
+ for s1, s2 in zip(m1.stages, m2.stages):
+ self._compare_pipelines(s1, s2)
+ elif isinstance(m1, OneVsRest) or isinstance(m1, OneVsRestModel):
+ for p in m1.params:
+ self._compare_params(m1, m2, p)
+ if isinstance(m1, OneVsRestModel):
+ self.assertEqual(len(m1.models), len(m2.models))
+ for x, y in zip(m1.models, m2.models):
+ self._compare_pipelines(x, y)
+ else:
+ raise RuntimeError("_compare_pipelines does not yet support type: %s" % type(m1))
+
+ def test_pipeline_persistence(self):
+ """
+ Pipeline[HashingTF, PCA]
+ """
+ temp_path = tempfile.mkdtemp()
+
+ try:
+ df = self.spark.createDataFrame([(["a", "b", "c"],), (["c", "d", "e"],)], ["words"])
+ tf = HashingTF(numFeatures=10, inputCol="words", outputCol="features")
+ pca = PCA(k=2, inputCol="features", outputCol="pca_features")
+ pl = Pipeline(stages=[tf, pca])
+ model = pl.fit(df)
+
+ pipeline_path = temp_path + "/pipeline"
+ pl.save(pipeline_path)
+ loaded_pipeline = Pipeline.load(pipeline_path)
+ self._compare_pipelines(pl, loaded_pipeline)
+
+ model_path = temp_path + "/pipeline-model"
+ model.save(model_path)
+ loaded_model = PipelineModel.load(model_path)
+ self._compare_pipelines(model, loaded_model)
+ finally:
+ try:
+ rmtree(temp_path)
+ except OSError:
+ pass
+
+ def test_nested_pipeline_persistence(self):
+ """
+ Pipeline[HashingTF, Pipeline[PCA]]
+ """
+ temp_path = tempfile.mkdtemp()
+
+ try:
+ df = self.spark.createDataFrame([(["a", "b", "c"],), (["c", "d", "e"],)], ["words"])
+ tf = HashingTF(numFeatures=10, inputCol="words", outputCol="features")
+ pca = PCA(k=2, inputCol="features", outputCol="pca_features")
+ p0 = Pipeline(stages=[pca])
+ pl = Pipeline(stages=[tf, p0])
+ model = pl.fit(df)
+
+ pipeline_path = temp_path + "/pipeline"
+ pl.save(pipeline_path)
+ loaded_pipeline = Pipeline.load(pipeline_path)
+ self._compare_pipelines(pl, loaded_pipeline)
+
+ model_path = temp_path + "/pipeline-model"
+ model.save(model_path)
+ loaded_model = PipelineModel.load(model_path)
+ self._compare_pipelines(model, loaded_model)
+ finally:
+ try:
+ rmtree(temp_path)
+ except OSError:
+ pass
+
+ def test_python_transformer_pipeline_persistence(self):
+ """
+ Pipeline[MockUnaryTransformer, Binarizer]
+ """
+ temp_path = tempfile.mkdtemp()
+
+ try:
+ df = self.spark.range(0, 10).toDF('input')
+ tf = MockUnaryTransformer(shiftVal=2)\
+ .setInputCol("input").setOutputCol("shiftedInput")
+ tf2 = Binarizer(threshold=6, inputCol="shiftedInput", outputCol="binarized")
+ pl = Pipeline(stages=[tf, tf2])
+ model = pl.fit(df)
+
+ pipeline_path = temp_path + "/pipeline"
+ pl.save(pipeline_path)
+ loaded_pipeline = Pipeline.load(pipeline_path)
+ self._compare_pipelines(pl, loaded_pipeline)
+
+ model_path = temp_path + "/pipeline-model"
+ model.save(model_path)
+ loaded_model = PipelineModel.load(model_path)
+ self._compare_pipelines(model, loaded_model)
+ finally:
+ try:
+ rmtree(temp_path)
+ except OSError:
+ pass
+
+ def test_onevsrest(self):
+ temp_path = tempfile.mkdtemp()
+ df = self.spark.createDataFrame([(0.0, Vectors.dense(1.0, 0.8)),
+ (1.0, Vectors.sparse(2, [], [])),
+ (2.0, Vectors.dense(0.5, 0.5))] * 10,
+ ["label", "features"])
+ lr = LogisticRegression(maxIter=5, regParam=0.01)
+ ovr = OneVsRest(classifier=lr)
+ model = ovr.fit(df)
+ ovrPath = temp_path + "/ovr"
+ ovr.save(ovrPath)
+ loadedOvr = OneVsRest.load(ovrPath)
+ self._compare_pipelines(ovr, loadedOvr)
+ modelPath = temp_path + "/ovrModel"
+ model.save(modelPath)
+ loadedModel = OneVsRestModel.load(modelPath)
+ self._compare_pipelines(model, loadedModel)
+
+ def test_decisiontree_classifier(self):
+ dt = DecisionTreeClassifier(maxDepth=1)
+ path = tempfile.mkdtemp()
+ dtc_path = path + "/dtc"
+ dt.save(dtc_path)
+ dt2 = DecisionTreeClassifier.load(dtc_path)
+ self.assertEqual(dt2.uid, dt2.maxDepth.parent,
+ "Loaded DecisionTreeClassifier instance uid (%s) "
+ "did not match Param's uid (%s)"
+ % (dt2.uid, dt2.maxDepth.parent))
+ self.assertEqual(dt._defaultParamMap[dt.maxDepth], dt2._defaultParamMap[dt2.maxDepth],
+ "Loaded DecisionTreeClassifier instance default params did not match " +
+ "original defaults")
+ try:
+ rmtree(path)
+ except OSError:
+ pass
+
+ def test_decisiontree_regressor(self):
+ dt = DecisionTreeRegressor(maxDepth=1)
+ path = tempfile.mkdtemp()
+ dtr_path = path + "/dtr"
+ dt.save(dtr_path)
+ dt2 = DecisionTreeClassifier.load(dtr_path)
+ self.assertEqual(dt2.uid, dt2.maxDepth.parent,
+ "Loaded DecisionTreeRegressor instance uid (%s) "
+ "did not match Param's uid (%s)"
+ % (dt2.uid, dt2.maxDepth.parent))
+ self.assertEqual(dt._defaultParamMap[dt.maxDepth], dt2._defaultParamMap[dt2.maxDepth],
+ "Loaded DecisionTreeRegressor instance default params did not match " +
+ "original defaults")
+ try:
+ rmtree(path)
+ except OSError:
+ pass
+
+ def test_default_read_write(self):
+ temp_path = tempfile.mkdtemp()
+
+ lr = LogisticRegression()
+ lr.setMaxIter(50)
+ lr.setThreshold(.75)
+ writer = DefaultParamsWriter(lr)
+
+ savePath = temp_path + "/lr"
+ writer.save(savePath)
+
+ reader = DefaultParamsReadable.read()
+ lr2 = reader.load(savePath)
+
+ self.assertEqual(lr.uid, lr2.uid)
+ self.assertEqual(lr.extractParamMap(), lr2.extractParamMap())
+
+ # test overwrite
+ lr.setThreshold(.8)
+ writer.overwrite().save(savePath)
+
+ reader = DefaultParamsReadable.read()
+ lr3 = reader.load(savePath)
+
+ self.assertEqual(lr.uid, lr3.uid)
+ self.assertEqual(lr.extractParamMap(), lr3.extractParamMap())
+
+ def test_default_read_write_default_params(self):
+ lr = LogisticRegression()
+ self.assertFalse(lr.isSet(lr.getParam("threshold")))
+
+ lr.setMaxIter(50)
+ lr.setThreshold(.75)
+
+ # `threshold` is set by user, default param `predictionCol` is not set by user.
+ self.assertTrue(lr.isSet(lr.getParam("threshold")))
+ self.assertFalse(lr.isSet(lr.getParam("predictionCol")))
+ self.assertTrue(lr.hasDefault(lr.getParam("predictionCol")))
+
+ writer = DefaultParamsWriter(lr)
+ metadata = json.loads(writer._get_metadata_to_save(lr, self.sc))
+ self.assertTrue("defaultParamMap" in metadata)
+
+ reader = DefaultParamsReadable.read()
+ metadataStr = json.dumps(metadata, separators=[',', ':'])
+ loadedMetadata = reader._parseMetaData(metadataStr, )
+ reader.getAndSetParams(lr, loadedMetadata)
+
+ self.assertTrue(lr.isSet(lr.getParam("threshold")))
+ self.assertFalse(lr.isSet(lr.getParam("predictionCol")))
+ self.assertTrue(lr.hasDefault(lr.getParam("predictionCol")))
+
+ # manually create metadata without `defaultParamMap` section.
+ del metadata['defaultParamMap']
+ metadataStr = json.dumps(metadata, separators=[',', ':'])
+ loadedMetadata = reader._parseMetaData(metadataStr, )
+ with self.assertRaisesRegexp(AssertionError, "`defaultParamMap` section not found"):
+ reader.getAndSetParams(lr, loadedMetadata)
+
+ # Prior to 2.4.0, metadata doesn't have `defaultParamMap`.
+ metadata['sparkVersion'] = '2.3.0'
+ metadataStr = json.dumps(metadata, separators=[',', ':'])
+ loadedMetadata = reader._parseMetaData(metadataStr, )
+ reader.getAndSetParams(lr, loadedMetadata)
+
+
+if __name__ == "__main__":
+ from pyspark.ml.tests.test_persistence import *
+
+ try:
+ import xmlrunner
+ testRunner = xmlrunner.XMLTestRunner(output='target/test-reports')
+ except ImportError:
+ testRunner = None
+ unittest.main(testRunner=testRunner, verbosity=2)
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/test_pipeline.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/test_pipeline.py b/python/pyspark/ml/tests/test_pipeline.py
new file mode 100644
index 0000000..31ef02c
--- /dev/null
+++ b/python/pyspark/ml/tests/test_pipeline.py
@@ -0,0 +1,77 @@
+#
+# 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 sys
+if sys.version_info[:2] <= (2, 6):
+ try:
+ import unittest2 as unittest
+ except ImportError:
+ sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+ sys.exit(1)
+else:
+ import unittest
+
+from pyspark.ml.pipeline import Pipeline
+from pyspark.testing.mlutils import MockDataset, MockEstimator, MockTransformer, PySparkTestCase
+
+
+class PipelineTests(PySparkTestCase):
+
+ def test_pipeline(self):
+ dataset = MockDataset()
+ estimator0 = MockEstimator()
+ transformer1 = MockTransformer()
+ estimator2 = MockEstimator()
+ transformer3 = MockTransformer()
+ pipeline = Pipeline(stages=[estimator0, transformer1, estimator2, transformer3])
+ pipeline_model = pipeline.fit(dataset, {estimator0.fake: 0, transformer1.fake: 1})
+ model0, transformer1, model2, transformer3 = pipeline_model.stages
+ self.assertEqual(0, model0.dataset_index)
+ self.assertEqual(0, model0.getFake())
+ self.assertEqual(1, transformer1.dataset_index)
+ self.assertEqual(1, transformer1.getFake())
+ self.assertEqual(2, dataset.index)
+ self.assertIsNone(model2.dataset_index, "The last model shouldn't be called in fit.")
+ self.assertIsNone(transformer3.dataset_index,
+ "The last transformer shouldn't be called in fit.")
+ dataset = pipeline_model.transform(dataset)
+ self.assertEqual(2, model0.dataset_index)
+ self.assertEqual(3, transformer1.dataset_index)
+ self.assertEqual(4, model2.dataset_index)
+ self.assertEqual(5, transformer3.dataset_index)
+ self.assertEqual(6, dataset.index)
+
+ def test_identity_pipeline(self):
+ dataset = MockDataset()
+
+ def doTransform(pipeline):
+ pipeline_model = pipeline.fit(dataset)
+ return pipeline_model.transform(dataset)
+ # check that empty pipeline did not perform any transformation
+ self.assertEqual(dataset.index, doTransform(Pipeline(stages=[])).index)
+ # check that failure to set stages param will raise KeyError for missing param
+ self.assertRaises(KeyError, lambda: doTransform(Pipeline()))
+
+
+if __name__ == "__main__":
+ from pyspark.ml.tests.test_pipeline import *
+
+ try:
+ import xmlrunner
+ testRunner = xmlrunner.XMLTestRunner(output='target/test-reports')
+ except ImportError:
+ testRunner = None
+ unittest.main(testRunner=testRunner, verbosity=2)
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/test_stat.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/test_stat.py b/python/pyspark/ml/tests/test_stat.py
new file mode 100644
index 0000000..bdc4853
--- /dev/null
+++ b/python/pyspark/ml/tests/test_stat.py
@@ -0,0 +1,58 @@
+#
+# 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 sys
+if sys.version_info[:2] <= (2, 6):
+ try:
+ import unittest2 as unittest
+ except ImportError:
+ sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+ sys.exit(1)
+else:
+ import unittest
+
+from pyspark.ml.linalg import Vectors
+from pyspark.ml.stat import ChiSquareTest
+from pyspark.sql import DataFrame
+from pyspark.testing.mlutils import SparkSessionTestCase
+
+
+class ChiSquareTestTests(SparkSessionTestCase):
+
+ def test_chisquaretest(self):
+ data = [[0, Vectors.dense([0, 1, 2])],
+ [1, Vectors.dense([1, 1, 1])],
+ [2, Vectors.dense([2, 1, 0])]]
+ df = self.spark.createDataFrame(data, ['label', 'feat'])
+ res = ChiSquareTest.test(df, 'feat', 'label')
+ # This line is hitting the collect bug described in #17218, commented for now.
+ # pValues = res.select("degreesOfFreedom").collect())
+ self.assertIsInstance(res, DataFrame)
+ fieldNames = set(field.name for field in res.schema.fields)
+ expectedFields = ["pValues", "degreesOfFreedom", "statistics"]
+ self.assertTrue(all(field in fieldNames for field in expectedFields))
+
+
+if __name__ == "__main__":
+ from pyspark.ml.tests.test_stat import *
+
+ try:
+ import xmlrunner
+ testRunner = xmlrunner.XMLTestRunner(output='target/test-reports')
+ except ImportError:
+ testRunner = None
+ unittest.main(testRunner=testRunner, verbosity=2)
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/test_training_summary.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/test_training_summary.py b/python/pyspark/ml/tests/test_training_summary.py
new file mode 100644
index 0000000..d5464f7
--- /dev/null
+++ b/python/pyspark/ml/tests/test_training_summary.py
@@ -0,0 +1,258 @@
+#
+# 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 sys
+if sys.version_info[:2] <= (2, 6):
+ try:
+ import unittest2 as unittest
+ except ImportError:
+ sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+ sys.exit(1)
+else:
+ import unittest
+
+if sys.version > '3':
+ basestring = str
+
+from pyspark.ml.classification import LogisticRegression
+from pyspark.ml.clustering import BisectingKMeans, GaussianMixture, KMeans
+from pyspark.ml.linalg import Vectors
+from pyspark.ml.regression import GeneralizedLinearRegression, LinearRegression
+from pyspark.sql import DataFrame
+from pyspark.testing.mlutils import SparkSessionTestCase
+
+
+class TrainingSummaryTest(SparkSessionTestCase):
+
+ def test_linear_regression_summary(self):
+ df = self.spark.createDataFrame([(1.0, 2.0, Vectors.dense(1.0)),
+ (0.0, 2.0, Vectors.sparse(1, [], []))],
+ ["label", "weight", "features"])
+ lr = LinearRegression(maxIter=5, regParam=0.0, solver="normal", weightCol="weight",
+ fitIntercept=False)
+ model = lr.fit(df)
+ self.assertTrue(model.hasSummary)
+ s = model.summary
+ # test that api is callable and returns expected types
+ self.assertGreater(s.totalIterations, 0)
+ self.assertTrue(isinstance(s.predictions, DataFrame))
+ self.assertEqual(s.predictionCol, "prediction")
+ self.assertEqual(s.labelCol, "label")
+ self.assertEqual(s.featuresCol, "features")
+ objHist = s.objectiveHistory
+ self.assertTrue(isinstance(objHist, list) and isinstance(objHist[0], float))
+ self.assertAlmostEqual(s.explainedVariance, 0.25, 2)
+ self.assertAlmostEqual(s.meanAbsoluteError, 0.0)
+ self.assertAlmostEqual(s.meanSquaredError, 0.0)
+ self.assertAlmostEqual(s.rootMeanSquaredError, 0.0)
+ self.assertAlmostEqual(s.r2, 1.0, 2)
+ self.assertAlmostEqual(s.r2adj, 1.0, 2)
+ self.assertTrue(isinstance(s.residuals, DataFrame))
+ self.assertEqual(s.numInstances, 2)
+ self.assertEqual(s.degreesOfFreedom, 1)
+ devResiduals = s.devianceResiduals
+ self.assertTrue(isinstance(devResiduals, list) and isinstance(devResiduals[0], float))
+ coefStdErr = s.coefficientStandardErrors
+ self.assertTrue(isinstance(coefStdErr, list) and isinstance(coefStdErr[0], float))
+ tValues = s.tValues
+ self.assertTrue(isinstance(tValues, list) and isinstance(tValues[0], float))
+ pValues = s.pValues
+ self.assertTrue(isinstance(pValues, list) and isinstance(pValues[0], float))
+ # test evaluation (with training dataset) produces a summary with same values
+ # one check is enough to verify a summary is returned
+ # The child class LinearRegressionTrainingSummary runs full test
+ sameSummary = model.evaluate(df)
+ self.assertAlmostEqual(sameSummary.explainedVariance, s.explainedVariance)
+
+ def test_glr_summary(self):
+ from pyspark.ml.linalg import Vectors
+ df = self.spark.createDataFrame([(1.0, 2.0, Vectors.dense(1.0)),
+ (0.0, 2.0, Vectors.sparse(1, [], []))],
+ ["label", "weight", "features"])
+ glr = GeneralizedLinearRegression(family="gaussian", link="identity", weightCol="weight",
+ fitIntercept=False)
+ model = glr.fit(df)
+ self.assertTrue(model.hasSummary)
+ s = model.summary
+ # test that api is callable and returns expected types
+ self.assertEqual(s.numIterations, 1) # this should default to a single iteration of WLS
+ self.assertTrue(isinstance(s.predictions, DataFrame))
+ self.assertEqual(s.predictionCol, "prediction")
+ self.assertEqual(s.numInstances, 2)
+ self.assertTrue(isinstance(s.residuals(), DataFrame))
+ self.assertTrue(isinstance(s.residuals("pearson"), DataFrame))
+ coefStdErr = s.coefficientStandardErrors
+ self.assertTrue(isinstance(coefStdErr, list) and isinstance(coefStdErr[0], float))
+ tValues = s.tValues
+ self.assertTrue(isinstance(tValues, list) and isinstance(tValues[0], float))
+ pValues = s.pValues
+ self.assertTrue(isinstance(pValues, list) and isinstance(pValues[0], float))
+ self.assertEqual(s.degreesOfFreedom, 1)
+ self.assertEqual(s.residualDegreeOfFreedom, 1)
+ self.assertEqual(s.residualDegreeOfFreedomNull, 2)
+ self.assertEqual(s.rank, 1)
+ self.assertTrue(isinstance(s.solver, basestring))
+ self.assertTrue(isinstance(s.aic, float))
+ self.assertTrue(isinstance(s.deviance, float))
+ self.assertTrue(isinstance(s.nullDeviance, float))
+ self.assertTrue(isinstance(s.dispersion, float))
+ # test evaluation (with training dataset) produces a summary with same values
+ # one check is enough to verify a summary is returned
+ # The child class GeneralizedLinearRegressionTrainingSummary runs full test
+ sameSummary = model.evaluate(df)
+ self.assertAlmostEqual(sameSummary.deviance, s.deviance)
+
+ def test_binary_logistic_regression_summary(self):
+ df = self.spark.createDataFrame([(1.0, 2.0, Vectors.dense(1.0)),
+ (0.0, 2.0, Vectors.sparse(1, [], []))],
+ ["label", "weight", "features"])
+ lr = LogisticRegression(maxIter=5, regParam=0.01, weightCol="weight", fitIntercept=False)
+ model = lr.fit(df)
+ self.assertTrue(model.hasSummary)
+ s = model.summary
+ # test that api is callable and returns expected types
+ self.assertTrue(isinstance(s.predictions, DataFrame))
+ self.assertEqual(s.probabilityCol, "probability")
+ self.assertEqual(s.labelCol, "label")
+ self.assertEqual(s.featuresCol, "features")
+ self.assertEqual(s.predictionCol, "prediction")
+ objHist = s.objectiveHistory
+ self.assertTrue(isinstance(objHist, list) and isinstance(objHist[0], float))
+ self.assertGreater(s.totalIterations, 0)
+ self.assertTrue(isinstance(s.labels, list))
+ self.assertTrue(isinstance(s.truePositiveRateByLabel, list))
+ self.assertTrue(isinstance(s.falsePositiveRateByLabel, list))
+ self.assertTrue(isinstance(s.precisionByLabel, list))
+ self.assertTrue(isinstance(s.recallByLabel, list))
+ self.assertTrue(isinstance(s.fMeasureByLabel(), list))
+ self.assertTrue(isinstance(s.fMeasureByLabel(1.0), list))
+ self.assertTrue(isinstance(s.roc, DataFrame))
+ self.assertAlmostEqual(s.areaUnderROC, 1.0, 2)
+ self.assertTrue(isinstance(s.pr, DataFrame))
+ self.assertTrue(isinstance(s.fMeasureByThreshold, DataFrame))
+ self.assertTrue(isinstance(s.precisionByThreshold, DataFrame))
+ self.assertTrue(isinstance(s.recallByThreshold, DataFrame))
+ self.assertAlmostEqual(s.accuracy, 1.0, 2)
+ self.assertAlmostEqual(s.weightedTruePositiveRate, 1.0, 2)
+ self.assertAlmostEqual(s.weightedFalsePositiveRate, 0.0, 2)
+ self.assertAlmostEqual(s.weightedRecall, 1.0, 2)
+ self.assertAlmostEqual(s.weightedPrecision, 1.0, 2)
+ self.assertAlmostEqual(s.weightedFMeasure(), 1.0, 2)
+ self.assertAlmostEqual(s.weightedFMeasure(1.0), 1.0, 2)
+ # test evaluation (with training dataset) produces a summary with same values
+ # one check is enough to verify a summary is returned, Scala version runs full test
+ sameSummary = model.evaluate(df)
+ self.assertAlmostEqual(sameSummary.areaUnderROC, s.areaUnderROC)
+
+ def test_multiclass_logistic_regression_summary(self):
+ df = self.spark.createDataFrame([(1.0, 2.0, Vectors.dense(1.0)),
+ (0.0, 2.0, Vectors.sparse(1, [], [])),
+ (2.0, 2.0, Vectors.dense(2.0)),
+ (2.0, 2.0, Vectors.dense(1.9))],
+ ["label", "weight", "features"])
+ lr = LogisticRegression(maxIter=5, regParam=0.01, weightCol="weight", fitIntercept=False)
+ model = lr.fit(df)
+ self.assertTrue(model.hasSummary)
+ s = model.summary
+ # test that api is callable and returns expected types
+ self.assertTrue(isinstance(s.predictions, DataFrame))
+ self.assertEqual(s.probabilityCol, "probability")
+ self.assertEqual(s.labelCol, "label")
+ self.assertEqual(s.featuresCol, "features")
+ self.assertEqual(s.predictionCol, "prediction")
+ objHist = s.objectiveHistory
+ self.assertTrue(isinstance(objHist, list) and isinstance(objHist[0], float))
+ self.assertGreater(s.totalIterations, 0)
+ self.assertTrue(isinstance(s.labels, list))
+ self.assertTrue(isinstance(s.truePositiveRateByLabel, list))
+ self.assertTrue(isinstance(s.falsePositiveRateByLabel, list))
+ self.assertTrue(isinstance(s.precisionByLabel, list))
+ self.assertTrue(isinstance(s.recallByLabel, list))
+ self.assertTrue(isinstance(s.fMeasureByLabel(), list))
+ self.assertTrue(isinstance(s.fMeasureByLabel(1.0), list))
+ self.assertAlmostEqual(s.accuracy, 0.75, 2)
+ self.assertAlmostEqual(s.weightedTruePositiveRate, 0.75, 2)
+ self.assertAlmostEqual(s.weightedFalsePositiveRate, 0.25, 2)
+ self.assertAlmostEqual(s.weightedRecall, 0.75, 2)
+ self.assertAlmostEqual(s.weightedPrecision, 0.583, 2)
+ self.assertAlmostEqual(s.weightedFMeasure(), 0.65, 2)
+ self.assertAlmostEqual(s.weightedFMeasure(1.0), 0.65, 2)
+ # test evaluation (with training dataset) produces a summary with same values
+ # one check is enough to verify a summary is returned, Scala version runs full test
+ sameSummary = model.evaluate(df)
+ self.assertAlmostEqual(sameSummary.accuracy, s.accuracy)
+
+ def test_gaussian_mixture_summary(self):
+ data = [(Vectors.dense(1.0),), (Vectors.dense(5.0),), (Vectors.dense(10.0),),
+ (Vectors.sparse(1, [], []),)]
+ df = self.spark.createDataFrame(data, ["features"])
+ gmm = GaussianMixture(k=2)
+ model = gmm.fit(df)
+ self.assertTrue(model.hasSummary)
+ s = model.summary
+ self.assertTrue(isinstance(s.predictions, DataFrame))
+ self.assertEqual(s.probabilityCol, "probability")
+ self.assertTrue(isinstance(s.probability, DataFrame))
+ self.assertEqual(s.featuresCol, "features")
+ self.assertEqual(s.predictionCol, "prediction")
+ self.assertTrue(isinstance(s.cluster, DataFrame))
+ self.assertEqual(len(s.clusterSizes), 2)
+ self.assertEqual(s.k, 2)
+ self.assertEqual(s.numIter, 3)
+
+ def test_bisecting_kmeans_summary(self):
+ data = [(Vectors.dense(1.0),), (Vectors.dense(5.0),), (Vectors.dense(10.0),),
+ (Vectors.sparse(1, [], []),)]
+ df = self.spark.createDataFrame(data, ["features"])
+ bkm = BisectingKMeans(k=2)
+ model = bkm.fit(df)
+ self.assertTrue(model.hasSummary)
+ s = model.summary
+ self.assertTrue(isinstance(s.predictions, DataFrame))
+ self.assertEqual(s.featuresCol, "features")
+ self.assertEqual(s.predictionCol, "prediction")
+ self.assertTrue(isinstance(s.cluster, DataFrame))
+ self.assertEqual(len(s.clusterSizes), 2)
+ self.assertEqual(s.k, 2)
+ self.assertEqual(s.numIter, 20)
+
+ def test_kmeans_summary(self):
+ data = [(Vectors.dense([0.0, 0.0]),), (Vectors.dense([1.0, 1.0]),),
+ (Vectors.dense([9.0, 8.0]),), (Vectors.dense([8.0, 9.0]),)]
+ df = self.spark.createDataFrame(data, ["features"])
+ kmeans = KMeans(k=2, seed=1)
+ model = kmeans.fit(df)
+ self.assertTrue(model.hasSummary)
+ s = model.summary
+ self.assertTrue(isinstance(s.predictions, DataFrame))
+ self.assertEqual(s.featuresCol, "features")
+ self.assertEqual(s.predictionCol, "prediction")
+ self.assertTrue(isinstance(s.cluster, DataFrame))
+ self.assertEqual(len(s.clusterSizes), 2)
+ self.assertEqual(s.k, 2)
+ self.assertEqual(s.numIter, 1)
+
+
+if __name__ == "__main__":
+ from pyspark.ml.tests.test_training_summary import *
+
+ try:
+ import xmlrunner
+ testRunner = xmlrunner.XMLTestRunner(output='target/test-reports')
+ except ImportError:
+ testRunner = None
+ unittest.main(testRunner=testRunner, verbosity=2)
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/test_tuning.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/test_tuning.py b/python/pyspark/ml/tests/test_tuning.py
new file mode 100644
index 0000000..af00d1d
--- /dev/null
+++ b/python/pyspark/ml/tests/test_tuning.py
@@ -0,0 +1,552 @@
+#
+# 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 sys
+import tempfile
+if sys.version_info[:2] <= (2, 6):
+ try:
+ import unittest2 as unittest
+ except ImportError:
+ sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+ sys.exit(1)
+else:
+ import unittest
+
+from pyspark.ml import Estimator, Model
+from pyspark.ml.classification import LogisticRegression, LogisticRegressionModel, OneVsRest
+from pyspark.ml.evaluation import BinaryClassificationEvaluator, \
+ MulticlassClassificationEvaluator, RegressionEvaluator
+from pyspark.ml.linalg import Vectors
+from pyspark.ml.param import Param, Params
+from pyspark.ml.tuning import CrossValidator, CrossValidatorModel, ParamGridBuilder, \
+ TrainValidationSplit, TrainValidationSplitModel
+from pyspark.sql.functions import rand
+from pyspark.testing.mlutils import SparkSessionTestCase
+
+
+class HasInducedError(Params):
+
+ def __init__(self):
+ super(HasInducedError, self).__init__()
+ self.inducedError = Param(self, "inducedError",
+ "Uniformly-distributed error added to feature")
+
+ def getInducedError(self):
+ return self.getOrDefault(self.inducedError)
+
+
+class InducedErrorModel(Model, HasInducedError):
+
+ def __init__(self):
+ super(InducedErrorModel, self).__init__()
+
+ def _transform(self, dataset):
+ return dataset.withColumn("prediction",
+ dataset.feature + (rand(0) * self.getInducedError()))
+
+
+class InducedErrorEstimator(Estimator, HasInducedError):
+
+ def __init__(self, inducedError=1.0):
+ super(InducedErrorEstimator, self).__init__()
+ self._set(inducedError=inducedError)
+
+ def _fit(self, dataset):
+ model = InducedErrorModel()
+ self._copyValues(model)
+ return model
+
+
+class CrossValidatorTests(SparkSessionTestCase):
+
+ def test_copy(self):
+ dataset = self.spark.createDataFrame([
+ (10, 10.0),
+ (50, 50.0),
+ (100, 100.0),
+ (500, 500.0)] * 10,
+ ["feature", "label"])
+
+ iee = InducedErrorEstimator()
+ evaluator = RegressionEvaluator(metricName="rmse")
+
+ grid = (ParamGridBuilder()
+ .addGrid(iee.inducedError, [100.0, 0.0, 10000.0])
+ .build())
+ cv = CrossValidator(estimator=iee, estimatorParamMaps=grid, evaluator=evaluator)
+ cvCopied = cv.copy()
+ self.assertEqual(cv.getEstimator().uid, cvCopied.getEstimator().uid)
+
+ cvModel = cv.fit(dataset)
+ cvModelCopied = cvModel.copy()
+ for index in range(len(cvModel.avgMetrics)):
+ self.assertTrue(abs(cvModel.avgMetrics[index] - cvModelCopied.avgMetrics[index])
+ < 0.0001)
+
+ def test_fit_minimize_metric(self):
+ dataset = self.spark.createDataFrame([
+ (10, 10.0),
+ (50, 50.0),
+ (100, 100.0),
+ (500, 500.0)] * 10,
+ ["feature", "label"])
+
+ iee = InducedErrorEstimator()
+ evaluator = RegressionEvaluator(metricName="rmse")
+
+ grid = (ParamGridBuilder()
+ .addGrid(iee.inducedError, [100.0, 0.0, 10000.0])
+ .build())
+ cv = CrossValidator(estimator=iee, estimatorParamMaps=grid, evaluator=evaluator)
+ cvModel = cv.fit(dataset)
+ bestModel = cvModel.bestModel
+ bestModelMetric = evaluator.evaluate(bestModel.transform(dataset))
+
+ self.assertEqual(0.0, bestModel.getOrDefault('inducedError'),
+ "Best model should have zero induced error")
+ self.assertEqual(0.0, bestModelMetric, "Best model has RMSE of 0")
+
+ def test_fit_maximize_metric(self):
+ dataset = self.spark.createDataFrame([
+ (10, 10.0),
+ (50, 50.0),
+ (100, 100.0),
+ (500, 500.0)] * 10,
+ ["feature", "label"])
+
+ iee = InducedErrorEstimator()
+ evaluator = RegressionEvaluator(metricName="r2")
+
+ grid = (ParamGridBuilder()
+ .addGrid(iee.inducedError, [100.0, 0.0, 10000.0])
+ .build())
+ cv = CrossValidator(estimator=iee, estimatorParamMaps=grid, evaluator=evaluator)
+ cvModel = cv.fit(dataset)
+ bestModel = cvModel.bestModel
+ bestModelMetric = evaluator.evaluate(bestModel.transform(dataset))
+
+ self.assertEqual(0.0, bestModel.getOrDefault('inducedError'),
+ "Best model should have zero induced error")
+ self.assertEqual(1.0, bestModelMetric, "Best model has R-squared of 1")
+
+ def test_param_grid_type_coercion(self):
+ lr = LogisticRegression(maxIter=10)
+ paramGrid = ParamGridBuilder().addGrid(lr.regParam, [0.5, 1]).build()
+ for param in paramGrid:
+ for v in param.values():
+ assert(type(v) == float)
+
+ def test_save_load_trained_model(self):
+ # This tests saving and loading the trained model only.
+ # Save/load for CrossValidator will be added later: SPARK-13786
+ temp_path = tempfile.mkdtemp()
+ dataset = self.spark.createDataFrame(
+ [(Vectors.dense([0.0]), 0.0),
+ (Vectors.dense([0.4]), 1.0),
+ (Vectors.dense([0.5]), 0.0),
+ (Vectors.dense([0.6]), 1.0),
+ (Vectors.dense([1.0]), 1.0)] * 10,
+ ["features", "label"])
+ lr = LogisticRegression()
+ grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
+ evaluator = BinaryClassificationEvaluator()
+ cv = CrossValidator(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator)
+ cvModel = cv.fit(dataset)
+ lrModel = cvModel.bestModel
+
+ cvModelPath = temp_path + "/cvModel"
+ lrModel.save(cvModelPath)
+ loadedLrModel = LogisticRegressionModel.load(cvModelPath)
+ self.assertEqual(loadedLrModel.uid, lrModel.uid)
+ self.assertEqual(loadedLrModel.intercept, lrModel.intercept)
+
+ def test_save_load_simple_estimator(self):
+ temp_path = tempfile.mkdtemp()
+ dataset = self.spark.createDataFrame(
+ [(Vectors.dense([0.0]), 0.0),
+ (Vectors.dense([0.4]), 1.0),
+ (Vectors.dense([0.5]), 0.0),
+ (Vectors.dense([0.6]), 1.0),
+ (Vectors.dense([1.0]), 1.0)] * 10,
+ ["features", "label"])
+
+ lr = LogisticRegression()
+ grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
+ evaluator = BinaryClassificationEvaluator()
+
+ # test save/load of CrossValidator
+ cv = CrossValidator(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator)
+ cvModel = cv.fit(dataset)
+ cvPath = temp_path + "/cv"
+ cv.save(cvPath)
+ loadedCV = CrossValidator.load(cvPath)
+ self.assertEqual(loadedCV.getEstimator().uid, cv.getEstimator().uid)
+ self.assertEqual(loadedCV.getEvaluator().uid, cv.getEvaluator().uid)
+ self.assertEqual(loadedCV.getEstimatorParamMaps(), cv.getEstimatorParamMaps())
+
+ # test save/load of CrossValidatorModel
+ cvModelPath = temp_path + "/cvModel"
+ cvModel.save(cvModelPath)
+ loadedModel = CrossValidatorModel.load(cvModelPath)
+ self.assertEqual(loadedModel.bestModel.uid, cvModel.bestModel.uid)
+
+ def test_parallel_evaluation(self):
+ dataset = self.spark.createDataFrame(
+ [(Vectors.dense([0.0]), 0.0),
+ (Vectors.dense([0.4]), 1.0),
+ (Vectors.dense([0.5]), 0.0),
+ (Vectors.dense([0.6]), 1.0),
+ (Vectors.dense([1.0]), 1.0)] * 10,
+ ["features", "label"])
+
+ lr = LogisticRegression()
+ grid = ParamGridBuilder().addGrid(lr.maxIter, [5, 6]).build()
+ evaluator = BinaryClassificationEvaluator()
+
+ # test save/load of CrossValidator
+ cv = CrossValidator(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator)
+ cv.setParallelism(1)
+ cvSerialModel = cv.fit(dataset)
+ cv.setParallelism(2)
+ cvParallelModel = cv.fit(dataset)
+ self.assertEqual(cvSerialModel.avgMetrics, cvParallelModel.avgMetrics)
+
+ def test_expose_sub_models(self):
+ temp_path = tempfile.mkdtemp()
+ dataset = self.spark.createDataFrame(
+ [(Vectors.dense([0.0]), 0.0),
+ (Vectors.dense([0.4]), 1.0),
+ (Vectors.dense([0.5]), 0.0),
+ (Vectors.dense([0.6]), 1.0),
+ (Vectors.dense([1.0]), 1.0)] * 10,
+ ["features", "label"])
+
+ lr = LogisticRegression()
+ grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
+ evaluator = BinaryClassificationEvaluator()
+
+ numFolds = 3
+ cv = CrossValidator(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator,
+ numFolds=numFolds, collectSubModels=True)
+
+ def checkSubModels(subModels):
+ self.assertEqual(len(subModels), numFolds)
+ for i in range(numFolds):
+ self.assertEqual(len(subModels[i]), len(grid))
+
+ cvModel = cv.fit(dataset)
+ checkSubModels(cvModel.subModels)
+
+ # Test the default value for option "persistSubModel" to be "true"
+ testSubPath = temp_path + "/testCrossValidatorSubModels"
+ savingPathWithSubModels = testSubPath + "cvModel3"
+ cvModel.save(savingPathWithSubModels)
+ cvModel3 = CrossValidatorModel.load(savingPathWithSubModels)
+ checkSubModels(cvModel3.subModels)
+ cvModel4 = cvModel3.copy()
+ checkSubModels(cvModel4.subModels)
+
+ savingPathWithoutSubModels = testSubPath + "cvModel2"
+ cvModel.write().option("persistSubModels", "false").save(savingPathWithoutSubModels)
+ cvModel2 = CrossValidatorModel.load(savingPathWithoutSubModels)
+ self.assertEqual(cvModel2.subModels, None)
+
+ for i in range(numFolds):
+ for j in range(len(grid)):
+ self.assertEqual(cvModel.subModels[i][j].uid, cvModel3.subModels[i][j].uid)
+
+ def test_save_load_nested_estimator(self):
+ temp_path = tempfile.mkdtemp()
+ dataset = self.spark.createDataFrame(
+ [(Vectors.dense([0.0]), 0.0),
+ (Vectors.dense([0.4]), 1.0),
+ (Vectors.dense([0.5]), 0.0),
+ (Vectors.dense([0.6]), 1.0),
+ (Vectors.dense([1.0]), 1.0)] * 10,
+ ["features", "label"])
+
+ ova = OneVsRest(classifier=LogisticRegression())
+ lr1 = LogisticRegression().setMaxIter(100)
+ lr2 = LogisticRegression().setMaxIter(150)
+ grid = ParamGridBuilder().addGrid(ova.classifier, [lr1, lr2]).build()
+ evaluator = MulticlassClassificationEvaluator()
+
+ # test save/load of CrossValidator
+ cv = CrossValidator(estimator=ova, estimatorParamMaps=grid, evaluator=evaluator)
+ cvModel = cv.fit(dataset)
+ cvPath = temp_path + "/cv"
+ cv.save(cvPath)
+ loadedCV = CrossValidator.load(cvPath)
+ self.assertEqual(loadedCV.getEstimator().uid, cv.getEstimator().uid)
+ self.assertEqual(loadedCV.getEvaluator().uid, cv.getEvaluator().uid)
+
+ originalParamMap = cv.getEstimatorParamMaps()
+ loadedParamMap = loadedCV.getEstimatorParamMaps()
+ for i, param in enumerate(loadedParamMap):
+ for p in param:
+ if p.name == "classifier":
+ self.assertEqual(param[p].uid, originalParamMap[i][p].uid)
+ else:
+ self.assertEqual(param[p], originalParamMap[i][p])
+
+ # test save/load of CrossValidatorModel
+ cvModelPath = temp_path + "/cvModel"
+ cvModel.save(cvModelPath)
+ loadedModel = CrossValidatorModel.load(cvModelPath)
+ self.assertEqual(loadedModel.bestModel.uid, cvModel.bestModel.uid)
+
+
+class TrainValidationSplitTests(SparkSessionTestCase):
+
+ def test_fit_minimize_metric(self):
+ dataset = self.spark.createDataFrame([
+ (10, 10.0),
+ (50, 50.0),
+ (100, 100.0),
+ (500, 500.0)] * 10,
+ ["feature", "label"])
+
+ iee = InducedErrorEstimator()
+ evaluator = RegressionEvaluator(metricName="rmse")
+
+ grid = ParamGridBuilder() \
+ .addGrid(iee.inducedError, [100.0, 0.0, 10000.0]) \
+ .build()
+ tvs = TrainValidationSplit(estimator=iee, estimatorParamMaps=grid, evaluator=evaluator)
+ tvsModel = tvs.fit(dataset)
+ bestModel = tvsModel.bestModel
+ bestModelMetric = evaluator.evaluate(bestModel.transform(dataset))
+ validationMetrics = tvsModel.validationMetrics
+
+ self.assertEqual(0.0, bestModel.getOrDefault('inducedError'),
+ "Best model should have zero induced error")
+ self.assertEqual(0.0, bestModelMetric, "Best model has RMSE of 0")
+ self.assertEqual(len(grid), len(validationMetrics),
+ "validationMetrics has the same size of grid parameter")
+ self.assertEqual(0.0, min(validationMetrics))
+
+ def test_fit_maximize_metric(self):
+ dataset = self.spark.createDataFrame([
+ (10, 10.0),
+ (50, 50.0),
+ (100, 100.0),
+ (500, 500.0)] * 10,
+ ["feature", "label"])
+
+ iee = InducedErrorEstimator()
+ evaluator = RegressionEvaluator(metricName="r2")
+
+ grid = ParamGridBuilder() \
+ .addGrid(iee.inducedError, [100.0, 0.0, 10000.0]) \
+ .build()
+ tvs = TrainValidationSplit(estimator=iee, estimatorParamMaps=grid, evaluator=evaluator)
+ tvsModel = tvs.fit(dataset)
+ bestModel = tvsModel.bestModel
+ bestModelMetric = evaluator.evaluate(bestModel.transform(dataset))
+ validationMetrics = tvsModel.validationMetrics
+
+ self.assertEqual(0.0, bestModel.getOrDefault('inducedError'),
+ "Best model should have zero induced error")
+ self.assertEqual(1.0, bestModelMetric, "Best model has R-squared of 1")
+ self.assertEqual(len(grid), len(validationMetrics),
+ "validationMetrics has the same size of grid parameter")
+ self.assertEqual(1.0, max(validationMetrics))
+
+ def test_save_load_trained_model(self):
+ # This tests saving and loading the trained model only.
+ # Save/load for TrainValidationSplit will be added later: SPARK-13786
+ temp_path = tempfile.mkdtemp()
+ dataset = self.spark.createDataFrame(
+ [(Vectors.dense([0.0]), 0.0),
+ (Vectors.dense([0.4]), 1.0),
+ (Vectors.dense([0.5]), 0.0),
+ (Vectors.dense([0.6]), 1.0),
+ (Vectors.dense([1.0]), 1.0)] * 10,
+ ["features", "label"])
+ lr = LogisticRegression()
+ grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
+ evaluator = BinaryClassificationEvaluator()
+ tvs = TrainValidationSplit(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator)
+ tvsModel = tvs.fit(dataset)
+ lrModel = tvsModel.bestModel
+
+ tvsModelPath = temp_path + "/tvsModel"
+ lrModel.save(tvsModelPath)
+ loadedLrModel = LogisticRegressionModel.load(tvsModelPath)
+ self.assertEqual(loadedLrModel.uid, lrModel.uid)
+ self.assertEqual(loadedLrModel.intercept, lrModel.intercept)
+
+ def test_save_load_simple_estimator(self):
+ # This tests saving and loading the trained model only.
+ # Save/load for TrainValidationSplit will be added later: SPARK-13786
+ temp_path = tempfile.mkdtemp()
+ dataset = self.spark.createDataFrame(
+ [(Vectors.dense([0.0]), 0.0),
+ (Vectors.dense([0.4]), 1.0),
+ (Vectors.dense([0.5]), 0.0),
+ (Vectors.dense([0.6]), 1.0),
+ (Vectors.dense([1.0]), 1.0)] * 10,
+ ["features", "label"])
+ lr = LogisticRegression()
+ grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
+ evaluator = BinaryClassificationEvaluator()
+ tvs = TrainValidationSplit(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator)
+ tvsModel = tvs.fit(dataset)
+
+ tvsPath = temp_path + "/tvs"
+ tvs.save(tvsPath)
+ loadedTvs = TrainValidationSplit.load(tvsPath)
+ self.assertEqual(loadedTvs.getEstimator().uid, tvs.getEstimator().uid)
+ self.assertEqual(loadedTvs.getEvaluator().uid, tvs.getEvaluator().uid)
+ self.assertEqual(loadedTvs.getEstimatorParamMaps(), tvs.getEstimatorParamMaps())
+
+ tvsModelPath = temp_path + "/tvsModel"
+ tvsModel.save(tvsModelPath)
+ loadedModel = TrainValidationSplitModel.load(tvsModelPath)
+ self.assertEqual(loadedModel.bestModel.uid, tvsModel.bestModel.uid)
+
+ def test_parallel_evaluation(self):
+ dataset = self.spark.createDataFrame(
+ [(Vectors.dense([0.0]), 0.0),
+ (Vectors.dense([0.4]), 1.0),
+ (Vectors.dense([0.5]), 0.0),
+ (Vectors.dense([0.6]), 1.0),
+ (Vectors.dense([1.0]), 1.0)] * 10,
+ ["features", "label"])
+ lr = LogisticRegression()
+ grid = ParamGridBuilder().addGrid(lr.maxIter, [5, 6]).build()
+ evaluator = BinaryClassificationEvaluator()
+ tvs = TrainValidationSplit(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator)
+ tvs.setParallelism(1)
+ tvsSerialModel = tvs.fit(dataset)
+ tvs.setParallelism(2)
+ tvsParallelModel = tvs.fit(dataset)
+ self.assertEqual(tvsSerialModel.validationMetrics, tvsParallelModel.validationMetrics)
+
+ def test_expose_sub_models(self):
+ temp_path = tempfile.mkdtemp()
+ dataset = self.spark.createDataFrame(
+ [(Vectors.dense([0.0]), 0.0),
+ (Vectors.dense([0.4]), 1.0),
+ (Vectors.dense([0.5]), 0.0),
+ (Vectors.dense([0.6]), 1.0),
+ (Vectors.dense([1.0]), 1.0)] * 10,
+ ["features", "label"])
+ lr = LogisticRegression()
+ grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
+ evaluator = BinaryClassificationEvaluator()
+ tvs = TrainValidationSplit(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator,
+ collectSubModels=True)
+ tvsModel = tvs.fit(dataset)
+ self.assertEqual(len(tvsModel.subModels), len(grid))
+
+ # Test the default value for option "persistSubModel" to be "true"
+ testSubPath = temp_path + "/testTrainValidationSplitSubModels"
+ savingPathWithSubModels = testSubPath + "cvModel3"
+ tvsModel.save(savingPathWithSubModels)
+ tvsModel3 = TrainValidationSplitModel.load(savingPathWithSubModels)
+ self.assertEqual(len(tvsModel3.subModels), len(grid))
+ tvsModel4 = tvsModel3.copy()
+ self.assertEqual(len(tvsModel4.subModels), len(grid))
+
+ savingPathWithoutSubModels = testSubPath + "cvModel2"
+ tvsModel.write().option("persistSubModels", "false").save(savingPathWithoutSubModels)
+ tvsModel2 = TrainValidationSplitModel.load(savingPathWithoutSubModels)
+ self.assertEqual(tvsModel2.subModels, None)
+
+ for i in range(len(grid)):
+ self.assertEqual(tvsModel.subModels[i].uid, tvsModel3.subModels[i].uid)
+
+ def test_save_load_nested_estimator(self):
+ # This tests saving and loading the trained model only.
+ # Save/load for TrainValidationSplit will be added later: SPARK-13786
+ temp_path = tempfile.mkdtemp()
+ dataset = self.spark.createDataFrame(
+ [(Vectors.dense([0.0]), 0.0),
+ (Vectors.dense([0.4]), 1.0),
+ (Vectors.dense([0.5]), 0.0),
+ (Vectors.dense([0.6]), 1.0),
+ (Vectors.dense([1.0]), 1.0)] * 10,
+ ["features", "label"])
+ ova = OneVsRest(classifier=LogisticRegression())
+ lr1 = LogisticRegression().setMaxIter(100)
+ lr2 = LogisticRegression().setMaxIter(150)
+ grid = ParamGridBuilder().addGrid(ova.classifier, [lr1, lr2]).build()
+ evaluator = MulticlassClassificationEvaluator()
+
+ tvs = TrainValidationSplit(estimator=ova, estimatorParamMaps=grid, evaluator=evaluator)
+ tvsModel = tvs.fit(dataset)
+ tvsPath = temp_path + "/tvs"
+ tvs.save(tvsPath)
+ loadedTvs = TrainValidationSplit.load(tvsPath)
+ self.assertEqual(loadedTvs.getEstimator().uid, tvs.getEstimator().uid)
+ self.assertEqual(loadedTvs.getEvaluator().uid, tvs.getEvaluator().uid)
+
+ originalParamMap = tvs.getEstimatorParamMaps()
+ loadedParamMap = loadedTvs.getEstimatorParamMaps()
+ for i, param in enumerate(loadedParamMap):
+ for p in param:
+ if p.name == "classifier":
+ self.assertEqual(param[p].uid, originalParamMap[i][p].uid)
+ else:
+ self.assertEqual(param[p], originalParamMap[i][p])
+
+ tvsModelPath = temp_path + "/tvsModel"
+ tvsModel.save(tvsModelPath)
+ loadedModel = TrainValidationSplitModel.load(tvsModelPath)
+ self.assertEqual(loadedModel.bestModel.uid, tvsModel.bestModel.uid)
+
+ def test_copy(self):
+ dataset = self.spark.createDataFrame([
+ (10, 10.0),
+ (50, 50.0),
+ (100, 100.0),
+ (500, 500.0)] * 10,
+ ["feature", "label"])
+
+ iee = InducedErrorEstimator()
+ evaluator = RegressionEvaluator(metricName="r2")
+
+ grid = ParamGridBuilder() \
+ .addGrid(iee.inducedError, [100.0, 0.0, 10000.0]) \
+ .build()
+ tvs = TrainValidationSplit(estimator=iee, estimatorParamMaps=grid, evaluator=evaluator)
+ tvsModel = tvs.fit(dataset)
+ tvsCopied = tvs.copy()
+ tvsModelCopied = tvsModel.copy()
+
+ self.assertEqual(tvs.getEstimator().uid, tvsCopied.getEstimator().uid,
+ "Copied TrainValidationSplit has the same uid of Estimator")
+
+ self.assertEqual(tvsModel.bestModel.uid, tvsModelCopied.bestModel.uid)
+ self.assertEqual(len(tvsModel.validationMetrics),
+ len(tvsModelCopied.validationMetrics),
+ "Copied validationMetrics has the same size of the original")
+ for index in range(len(tvsModel.validationMetrics)):
+ self.assertEqual(tvsModel.validationMetrics[index],
+ tvsModelCopied.validationMetrics[index])
+
+
+if __name__ == "__main__":
+ from pyspark.ml.tests.test_tuning import *
+
+ try:
+ import xmlrunner
+ testRunner = xmlrunner.XMLTestRunner(output='target/test-reports')
+ except ImportError:
+ testRunner = None
+ unittest.main(testRunner=testRunner, verbosity=2)
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/test_wrapper.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/test_wrapper.py b/python/pyspark/ml/tests/test_wrapper.py
new file mode 100644
index 0000000..4326d8e
--- /dev/null
+++ b/python/pyspark/ml/tests/test_wrapper.py
@@ -0,0 +1,120 @@
+#
+# 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 sys
+if sys.version_info[:2] <= (2, 6):
+ try:
+ import unittest2 as unittest
+ except ImportError:
+ sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+ sys.exit(1)
+else:
+ import unittest
+
+import py4j
+
+from pyspark.ml.linalg import DenseVector, Vectors
+from pyspark.ml.regression import LinearRegression
+from pyspark.ml.wrapper import _java2py, _py2java, JavaParams, JavaWrapper
+from pyspark.testing.mllibutils import MLlibTestCase
+from pyspark.testing.mlutils import SparkSessionTestCase
+
+
+class JavaWrapperMemoryTests(SparkSessionTestCase):
+
+ def test_java_object_gets_detached(self):
+ df = self.spark.createDataFrame([(1.0, 2.0, Vectors.dense(1.0)),
+ (0.0, 2.0, Vectors.sparse(1, [], []))],
+ ["label", "weight", "features"])
+ lr = LinearRegression(maxIter=1, regParam=0.0, solver="normal", weightCol="weight",
+ fitIntercept=False)
+
+ model = lr.fit(df)
+ summary = model.summary
+
+ self.assertIsInstance(model, JavaWrapper)
+ self.assertIsInstance(summary, JavaWrapper)
+ self.assertIsInstance(model, JavaParams)
+ self.assertNotIsInstance(summary, JavaParams)
+
+ error_no_object = 'Target Object ID does not exist for this gateway'
+
+ self.assertIn("LinearRegression_", model._java_obj.toString())
+ self.assertIn("LinearRegressionTrainingSummary", summary._java_obj.toString())
+
+ model.__del__()
+
+ with self.assertRaisesRegexp(py4j.protocol.Py4JError, error_no_object):
+ model._java_obj.toString()
+ self.assertIn("LinearRegressionTrainingSummary", summary._java_obj.toString())
+
+ try:
+ summary.__del__()
+ except:
+ pass
+
+ with self.assertRaisesRegexp(py4j.protocol.Py4JError, error_no_object):
+ model._java_obj.toString()
+ with self.assertRaisesRegexp(py4j.protocol.Py4JError, error_no_object):
+ summary._java_obj.toString()
+
+
+class WrapperTests(MLlibTestCase):
+
+ def test_new_java_array(self):
+ # test array of strings
+ str_list = ["a", "b", "c"]
+ java_class = self.sc._gateway.jvm.java.lang.String
+ java_array = JavaWrapper._new_java_array(str_list, java_class)
+ self.assertEqual(_java2py(self.sc, java_array), str_list)
+ # test array of integers
+ int_list = [1, 2, 3]
+ java_class = self.sc._gateway.jvm.java.lang.Integer
+ java_array = JavaWrapper._new_java_array(int_list, java_class)
+ self.assertEqual(_java2py(self.sc, java_array), int_list)
+ # test array of floats
+ float_list = [0.1, 0.2, 0.3]
+ java_class = self.sc._gateway.jvm.java.lang.Double
+ java_array = JavaWrapper._new_java_array(float_list, java_class)
+ self.assertEqual(_java2py(self.sc, java_array), float_list)
+ # test array of bools
+ bool_list = [False, True, True]
+ java_class = self.sc._gateway.jvm.java.lang.Boolean
+ java_array = JavaWrapper._new_java_array(bool_list, java_class)
+ self.assertEqual(_java2py(self.sc, java_array), bool_list)
+ # test array of Java DenseVectors
+ v1 = DenseVector([0.0, 1.0])
+ v2 = DenseVector([1.0, 0.0])
+ vec_java_list = [_py2java(self.sc, v1), _py2java(self.sc, v2)]
+ java_class = self.sc._gateway.jvm.org.apache.spark.ml.linalg.DenseVector
+ java_array = JavaWrapper._new_java_array(vec_java_list, java_class)
+ self.assertEqual(_java2py(self.sc, java_array), [v1, v2])
+ # test empty array
+ java_class = self.sc._gateway.jvm.java.lang.Integer
+ java_array = JavaWrapper._new_java_array([], java_class)
+ self.assertEqual(_java2py(self.sc, java_array), [])
+
+
+if __name__ == "__main__":
+ from pyspark.ml.tests.test_wrapper import *
+
+ try:
+ import xmlrunner
+ testRunner = xmlrunner.XMLTestRunner(output='target/test-reports')
+ except ImportError:
+ testRunner = None
+ unittest.main(testRunner=testRunner, verbosity=2)
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/testing/mlutils.py
----------------------------------------------------------------------
diff --git a/python/pyspark/testing/mlutils.py b/python/pyspark/testing/mlutils.py
new file mode 100644
index 0000000..12bf650
--- /dev/null
+++ b/python/pyspark/testing/mlutils.py
@@ -0,0 +1,161 @@
+#
+# 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.ml import Estimator, Model, Transformer, UnaryTransformer
+from pyspark.ml.param import Param, Params, TypeConverters
+from pyspark.ml.util import DefaultParamsReadable, DefaultParamsWritable
+from pyspark.ml.wrapper import _java2py
+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 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)
+ if not hasattr(py_stage, "_to_java"):
+ return
+ java_stage = py_stage._to_java()
+ if java_stage is None:
+ return
+ test_self.assertEqual(py_stage.uid, java_stage.uid(), msg=py_stage_str)
+ 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(), "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
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org
[4/4] spark git commit: [SPARK-26033][PYTHON][TESTS] Break large
ml/tests.py file into smaller files
Posted by gu...@apache.org.
[SPARK-26033][PYTHON][TESTS] Break large ml/tests.py file into smaller files
## What changes were proposed in this pull request?
This PR breaks down the large ml/tests.py file that contains all Python ML unit tests into several smaller test files to be easier to read and maintain.
The tests are broken down as follows:
```
pyspark
├── __init__.py
...
├── ml
│ ├── __init__.py
...
│ ├── tests
│ │ ├── __init__.py
│ │ ├── test_algorithms.py
│ │ ├── test_base.py
│ │ ├── test_evaluation.py
│ │ ├── test_feature.py
│ │ ├── test_image.py
│ │ ├── test_linalg.py
│ │ ├── test_param.py
│ │ ├── test_persistence.py
│ │ ├── test_pipeline.py
│ │ ├── test_stat.py
│ │ ├── test_training_summary.py
│ │ ├── test_tuning.py
│ │ └── test_wrapper.py
...
├── testing
...
│ ├── mlutils.py
...
```
## How was this patch tested?
Ran tests manually by module to ensure test count was the same, and ran `python/run-tests --modules=pyspark-ml` to verify all passing with Python 2.7 and Python 3.6.
Closes #23063 from BryanCutler/python-test-breakup-ml-SPARK-26033.
Authored-by: Bryan Cutler <cu...@gmail.com>
Signed-off-by: hyukjinkwon <gu...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/034ae305
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/034ae305
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/034ae305
Branch: refs/heads/master
Commit: 034ae305c33b1990b3c1a284044002874c343b4d
Parents: e00cac9
Author: Bryan Cutler <cu...@gmail.com>
Authored: Sun Nov 18 16:02:15 2018 +0800
Committer: hyukjinkwon <gu...@apache.org>
Committed: Sun Nov 18 16:02:15 2018 +0800
----------------------------------------------------------------------
dev/sparktestsupport/modules.py | 16 +-
python/pyspark/ml/tests.py | 2762 ------------------
python/pyspark/ml/tests/__init__.py | 16 +
python/pyspark/ml/tests/test_algorithms.py | 349 +++
python/pyspark/ml/tests/test_base.py | 85 +
python/pyspark/ml/tests/test_evaluation.py | 71 +
python/pyspark/ml/tests/test_feature.py | 318 ++
python/pyspark/ml/tests/test_image.py | 118 +
python/pyspark/ml/tests/test_linalg.py | 392 +++
python/pyspark/ml/tests/test_param.py | 372 +++
python/pyspark/ml/tests/test_persistence.py | 369 +++
python/pyspark/ml/tests/test_pipeline.py | 77 +
python/pyspark/ml/tests/test_stat.py | 58 +
.../pyspark/ml/tests/test_training_summary.py | 258 ++
python/pyspark/ml/tests/test_tuning.py | 552 ++++
python/pyspark/ml/tests/test_wrapper.py | 120 +
python/pyspark/testing/mlutils.py | 161 +
17 files changed, 3331 insertions(+), 2763 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/dev/sparktestsupport/modules.py
----------------------------------------------------------------------
diff --git a/dev/sparktestsupport/modules.py b/dev/sparktestsupport/modules.py
index 547635a..eef7f25 100644
--- a/dev/sparktestsupport/modules.py
+++ b/dev/sparktestsupport/modules.py
@@ -452,6 +452,7 @@ pyspark_ml = Module(
"python/pyspark/ml/"
],
python_test_goals=[
+ # doctests
"pyspark.ml.classification",
"pyspark.ml.clustering",
"pyspark.ml.evaluation",
@@ -463,7 +464,20 @@ pyspark_ml = Module(
"pyspark.ml.regression",
"pyspark.ml.stat",
"pyspark.ml.tuning",
- "pyspark.ml.tests",
+ # unittests
+ "pyspark.ml.tests.test_algorithms",
+ "pyspark.ml.tests.test_base",
+ "pyspark.ml.tests.test_evaluation",
+ "pyspark.ml.tests.test_feature",
+ "pyspark.ml.tests.test_image",
+ "pyspark.ml.tests.test_linalg",
+ "pyspark.ml.tests.test_param",
+ "pyspark.ml.tests.test_persistence",
+ "pyspark.ml.tests.test_pipeline",
+ "pyspark.ml.tests.test_stat",
+ "pyspark.ml.tests.test_training_summary",
+ "pyspark.ml.tests.test_tuning",
+ "pyspark.ml.tests.test_wrapper",
],
blacklisted_python_implementations=[
"PyPy" # Skip these tests under PyPy since they require numpy and it isn't available there
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org
[2/4] spark git commit: [SPARK-26033][PYTHON][TESTS] Break large
ml/tests.py file into smaller files
Posted by gu...@apache.org.
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/__init__.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/__init__.py b/python/pyspark/ml/tests/__init__.py
new file mode 100644
index 0000000..cce3aca
--- /dev/null
+++ b/python/pyspark/ml/tests/__init__.py
@@ -0,0 +1,16 @@
+#
+# 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.
+#
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/test_algorithms.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/test_algorithms.py b/python/pyspark/ml/tests/test_algorithms.py
new file mode 100644
index 0000000..1a72e12
--- /dev/null
+++ b/python/pyspark/ml/tests/test_algorithms.py
@@ -0,0 +1,349 @@
+#
+# 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.
+#
+
+from shutil import rmtree
+import sys
+import tempfile
+
+if sys.version_info[:2] <= (2, 6):
+ try:
+ import unittest2 as unittest
+ except ImportError:
+ sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+ sys.exit(1)
+else:
+ import unittest
+
+import numpy as np
+
+from pyspark.ml.classification import DecisionTreeClassifier, LogisticRegression, \
+ MultilayerPerceptronClassifier, OneVsRest
+from pyspark.ml.clustering import DistributedLDAModel, KMeans, LocalLDAModel, LDA, LDAModel
+from pyspark.ml.fpm import FPGrowth
+from pyspark.ml.linalg import Matrices, Vectors
+from pyspark.ml.recommendation import ALS
+from pyspark.ml.regression import GeneralizedLinearRegression, LinearRegression
+from pyspark.sql import Row
+from pyspark.testing.mlutils import SparkSessionTestCase
+
+
+class LogisticRegressionTest(SparkSessionTestCase):
+
+ def test_binomial_logistic_regression_with_bound(self):
+
+ df = self.spark.createDataFrame(
+ [(1.0, 1.0, Vectors.dense(0.0, 5.0)),
+ (0.0, 2.0, Vectors.dense(1.0, 2.0)),
+ (1.0, 3.0, Vectors.dense(2.0, 1.0)),
+ (0.0, 4.0, Vectors.dense(3.0, 3.0)), ], ["label", "weight", "features"])
+
+ lor = LogisticRegression(regParam=0.01, weightCol="weight",
+ lowerBoundsOnCoefficients=Matrices.dense(1, 2, [-1.0, -1.0]),
+ upperBoundsOnIntercepts=Vectors.dense(0.0))
+ model = lor.fit(df)
+ self.assertTrue(
+ np.allclose(model.coefficients.toArray(), [-0.2944, -0.0484], atol=1E-4))
+ self.assertTrue(np.isclose(model.intercept, 0.0, atol=1E-4))
+
+ def test_multinomial_logistic_regression_with_bound(self):
+
+ data_path = "data/mllib/sample_multiclass_classification_data.txt"
+ df = self.spark.read.format("libsvm").load(data_path)
+
+ lor = LogisticRegression(regParam=0.01,
+ lowerBoundsOnCoefficients=Matrices.dense(3, 4, range(12)),
+ upperBoundsOnIntercepts=Vectors.dense(0.0, 0.0, 0.0))
+ model = lor.fit(df)
+ expected = [[4.593, 4.5516, 9.0099, 12.2904],
+ [1.0, 8.1093, 7.0, 10.0],
+ [3.041, 5.0, 8.0, 11.0]]
+ for i in range(0, len(expected)):
+ self.assertTrue(
+ np.allclose(model.coefficientMatrix.toArray()[i], expected[i], atol=1E-4))
+ self.assertTrue(
+ np.allclose(model.interceptVector.toArray(), [-0.9057, -1.1392, -0.0033], atol=1E-4))
+
+
+class MultilayerPerceptronClassifierTest(SparkSessionTestCase):
+
+ def test_raw_and_probability_prediction(self):
+
+ data_path = "data/mllib/sample_multiclass_classification_data.txt"
+ df = self.spark.read.format("libsvm").load(data_path)
+
+ mlp = MultilayerPerceptronClassifier(maxIter=100, layers=[4, 5, 4, 3],
+ blockSize=128, seed=123)
+ model = mlp.fit(df)
+ test = self.sc.parallelize([Row(features=Vectors.dense(0.1, 0.1, 0.25, 0.25))]).toDF()
+ result = model.transform(test).head()
+ expected_prediction = 2.0
+ expected_probability = [0.0, 0.0, 1.0]
+ expected_rawPrediction = [57.3955, -124.5462, 67.9943]
+ self.assertTrue(result.prediction, expected_prediction)
+ self.assertTrue(np.allclose(result.probability, expected_probability, atol=1E-4))
+ self.assertTrue(np.allclose(result.rawPrediction, expected_rawPrediction, atol=1E-4))
+
+
+class OneVsRestTests(SparkSessionTestCase):
+
+ def test_copy(self):
+ df = self.spark.createDataFrame([(0.0, Vectors.dense(1.0, 0.8)),
+ (1.0, Vectors.sparse(2, [], [])),
+ (2.0, Vectors.dense(0.5, 0.5))],
+ ["label", "features"])
+ lr = LogisticRegression(maxIter=5, regParam=0.01)
+ ovr = OneVsRest(classifier=lr)
+ ovr1 = ovr.copy({lr.maxIter: 10})
+ self.assertEqual(ovr.getClassifier().getMaxIter(), 5)
+ self.assertEqual(ovr1.getClassifier().getMaxIter(), 10)
+ model = ovr.fit(df)
+ model1 = model.copy({model.predictionCol: "indexed"})
+ self.assertEqual(model1.getPredictionCol(), "indexed")
+
+ def test_output_columns(self):
+ df = self.spark.createDataFrame([(0.0, Vectors.dense(1.0, 0.8)),
+ (1.0, Vectors.sparse(2, [], [])),
+ (2.0, Vectors.dense(0.5, 0.5))],
+ ["label", "features"])
+ lr = LogisticRegression(maxIter=5, regParam=0.01)
+ ovr = OneVsRest(classifier=lr, parallelism=1)
+ model = ovr.fit(df)
+ output = model.transform(df)
+ self.assertEqual(output.columns, ["label", "features", "prediction"])
+
+ def test_parallelism_doesnt_change_output(self):
+ df = self.spark.createDataFrame([(0.0, Vectors.dense(1.0, 0.8)),
+ (1.0, Vectors.sparse(2, [], [])),
+ (2.0, Vectors.dense(0.5, 0.5))],
+ ["label", "features"])
+ ovrPar1 = OneVsRest(classifier=LogisticRegression(maxIter=5, regParam=.01), parallelism=1)
+ modelPar1 = ovrPar1.fit(df)
+ ovrPar2 = OneVsRest(classifier=LogisticRegression(maxIter=5, regParam=.01), parallelism=2)
+ modelPar2 = ovrPar2.fit(df)
+ for i, model in enumerate(modelPar1.models):
+ self.assertTrue(np.allclose(model.coefficients.toArray(),
+ modelPar2.models[i].coefficients.toArray(), atol=1E-4))
+ self.assertTrue(np.allclose(model.intercept, modelPar2.models[i].intercept, atol=1E-4))
+
+ def test_support_for_weightCol(self):
+ df = self.spark.createDataFrame([(0.0, Vectors.dense(1.0, 0.8), 1.0),
+ (1.0, Vectors.sparse(2, [], []), 1.0),
+ (2.0, Vectors.dense(0.5, 0.5), 1.0)],
+ ["label", "features", "weight"])
+ # classifier inherits hasWeightCol
+ lr = LogisticRegression(maxIter=5, regParam=0.01)
+ ovr = OneVsRest(classifier=lr, weightCol="weight")
+ self.assertIsNotNone(ovr.fit(df))
+ # classifier doesn't inherit hasWeightCol
+ dt = DecisionTreeClassifier()
+ ovr2 = OneVsRest(classifier=dt, weightCol="weight")
+ self.assertIsNotNone(ovr2.fit(df))
+
+
+class KMeansTests(SparkSessionTestCase):
+
+ def test_kmeans_cosine_distance(self):
+ data = [(Vectors.dense([1.0, 1.0]),), (Vectors.dense([10.0, 10.0]),),
+ (Vectors.dense([1.0, 0.5]),), (Vectors.dense([10.0, 4.4]),),
+ (Vectors.dense([-1.0, 1.0]),), (Vectors.dense([-100.0, 90.0]),)]
+ df = self.spark.createDataFrame(data, ["features"])
+ kmeans = KMeans(k=3, seed=1, distanceMeasure="cosine")
+ model = kmeans.fit(df)
+ result = model.transform(df).collect()
+ self.assertTrue(result[0].prediction == result[1].prediction)
+ self.assertTrue(result[2].prediction == result[3].prediction)
+ self.assertTrue(result[4].prediction == result[5].prediction)
+
+
+class LDATest(SparkSessionTestCase):
+
+ def _compare(self, m1, m2):
+ """
+ Temp method for comparing instances.
+ TODO: Replace with generic implementation once SPARK-14706 is merged.
+ """
+ self.assertEqual(m1.uid, m2.uid)
+ self.assertEqual(type(m1), type(m2))
+ self.assertEqual(len(m1.params), len(m2.params))
+ for p in m1.params:
+ if m1.isDefined(p):
+ self.assertEqual(m1.getOrDefault(p), m2.getOrDefault(p))
+ self.assertEqual(p.parent, m2.getParam(p.name).parent)
+ if isinstance(m1, LDAModel):
+ self.assertEqual(m1.vocabSize(), m2.vocabSize())
+ self.assertEqual(m1.topicsMatrix(), m2.topicsMatrix())
+
+ def test_persistence(self):
+ # Test save/load for LDA, LocalLDAModel, DistributedLDAModel.
+ df = self.spark.createDataFrame([
+ [1, Vectors.dense([0.0, 1.0])],
+ [2, Vectors.sparse(2, {0: 1.0})],
+ ], ["id", "features"])
+ # Fit model
+ lda = LDA(k=2, seed=1, optimizer="em")
+ distributedModel = lda.fit(df)
+ self.assertTrue(distributedModel.isDistributed())
+ localModel = distributedModel.toLocal()
+ self.assertFalse(localModel.isDistributed())
+ # Define paths
+ path = tempfile.mkdtemp()
+ lda_path = path + "/lda"
+ dist_model_path = path + "/distLDAModel"
+ local_model_path = path + "/localLDAModel"
+ # Test LDA
+ lda.save(lda_path)
+ lda2 = LDA.load(lda_path)
+ self._compare(lda, lda2)
+ # Test DistributedLDAModel
+ distributedModel.save(dist_model_path)
+ distributedModel2 = DistributedLDAModel.load(dist_model_path)
+ self._compare(distributedModel, distributedModel2)
+ # Test LocalLDAModel
+ localModel.save(local_model_path)
+ localModel2 = LocalLDAModel.load(local_model_path)
+ self._compare(localModel, localModel2)
+ # Clean up
+ try:
+ rmtree(path)
+ except OSError:
+ pass
+
+
+class FPGrowthTests(SparkSessionTestCase):
+ def setUp(self):
+ super(FPGrowthTests, self).setUp()
+ self.data = self.spark.createDataFrame(
+ [([1, 2], ), ([1, 2], ), ([1, 2, 3], ), ([1, 3], )],
+ ["items"])
+
+ def test_association_rules(self):
+ fp = FPGrowth()
+ fpm = fp.fit(self.data)
+
+ expected_association_rules = self.spark.createDataFrame(
+ [([3], [1], 1.0, 1.0), ([2], [1], 1.0, 1.0)],
+ ["antecedent", "consequent", "confidence", "lift"]
+ )
+ actual_association_rules = fpm.associationRules
+
+ self.assertEqual(actual_association_rules.subtract(expected_association_rules).count(), 0)
+ self.assertEqual(expected_association_rules.subtract(actual_association_rules).count(), 0)
+
+ def test_freq_itemsets(self):
+ fp = FPGrowth()
+ fpm = fp.fit(self.data)
+
+ expected_freq_itemsets = self.spark.createDataFrame(
+ [([1], 4), ([2], 3), ([2, 1], 3), ([3], 2), ([3, 1], 2)],
+ ["items", "freq"]
+ )
+ actual_freq_itemsets = fpm.freqItemsets
+
+ self.assertEqual(actual_freq_itemsets.subtract(expected_freq_itemsets).count(), 0)
+ self.assertEqual(expected_freq_itemsets.subtract(actual_freq_itemsets).count(), 0)
+
+ def tearDown(self):
+ del self.data
+
+
+class ALSTest(SparkSessionTestCase):
+
+ def test_storage_levels(self):
+ df = self.spark.createDataFrame(
+ [(0, 0, 4.0), (0, 1, 2.0), (1, 1, 3.0), (1, 2, 4.0), (2, 1, 1.0), (2, 2, 5.0)],
+ ["user", "item", "rating"])
+ als = ALS().setMaxIter(1).setRank(1)
+ # test default params
+ als.fit(df)
+ self.assertEqual(als.getIntermediateStorageLevel(), "MEMORY_AND_DISK")
+ self.assertEqual(als._java_obj.getIntermediateStorageLevel(), "MEMORY_AND_DISK")
+ self.assertEqual(als.getFinalStorageLevel(), "MEMORY_AND_DISK")
+ self.assertEqual(als._java_obj.getFinalStorageLevel(), "MEMORY_AND_DISK")
+ # test non-default params
+ als.setIntermediateStorageLevel("MEMORY_ONLY_2")
+ als.setFinalStorageLevel("DISK_ONLY")
+ als.fit(df)
+ self.assertEqual(als.getIntermediateStorageLevel(), "MEMORY_ONLY_2")
+ self.assertEqual(als._java_obj.getIntermediateStorageLevel(), "MEMORY_ONLY_2")
+ self.assertEqual(als.getFinalStorageLevel(), "DISK_ONLY")
+ self.assertEqual(als._java_obj.getFinalStorageLevel(), "DISK_ONLY")
+
+
+class GeneralizedLinearRegressionTest(SparkSessionTestCase):
+
+ def test_tweedie_distribution(self):
+
+ df = self.spark.createDataFrame(
+ [(1.0, Vectors.dense(0.0, 0.0)),
+ (1.0, Vectors.dense(1.0, 2.0)),
+ (2.0, Vectors.dense(0.0, 0.0)),
+ (2.0, Vectors.dense(1.0, 1.0)), ], ["label", "features"])
+
+ glr = GeneralizedLinearRegression(family="tweedie", variancePower=1.6)
+ model = glr.fit(df)
+ self.assertTrue(np.allclose(model.coefficients.toArray(), [-0.4645, 0.3402], atol=1E-4))
+ self.assertTrue(np.isclose(model.intercept, 0.7841, atol=1E-4))
+
+ model2 = glr.setLinkPower(-1.0).fit(df)
+ self.assertTrue(np.allclose(model2.coefficients.toArray(), [-0.6667, 0.5], atol=1E-4))
+ self.assertTrue(np.isclose(model2.intercept, 0.6667, atol=1E-4))
+
+ def test_offset(self):
+
+ df = self.spark.createDataFrame(
+ [(0.2, 1.0, 2.0, Vectors.dense(0.0, 5.0)),
+ (0.5, 2.1, 0.5, Vectors.dense(1.0, 2.0)),
+ (0.9, 0.4, 1.0, Vectors.dense(2.0, 1.0)),
+ (0.7, 0.7, 0.0, Vectors.dense(3.0, 3.0))], ["label", "weight", "offset", "features"])
+
+ glr = GeneralizedLinearRegression(family="poisson", weightCol="weight", offsetCol="offset")
+ model = glr.fit(df)
+ self.assertTrue(np.allclose(model.coefficients.toArray(), [0.664647, -0.3192581],
+ atol=1E-4))
+ self.assertTrue(np.isclose(model.intercept, -1.561613, atol=1E-4))
+
+
+class LinearRegressionTest(SparkSessionTestCase):
+
+ def test_linear_regression_with_huber_loss(self):
+
+ data_path = "data/mllib/sample_linear_regression_data.txt"
+ df = self.spark.read.format("libsvm").load(data_path)
+
+ lir = LinearRegression(loss="huber", epsilon=2.0)
+ model = lir.fit(df)
+
+ expectedCoefficients = [0.136, 0.7648, -0.7761, 2.4236, 0.537,
+ 1.2612, -0.333, -0.5694, -0.6311, 0.6053]
+ expectedIntercept = 0.1607
+ expectedScale = 9.758
+
+ self.assertTrue(
+ np.allclose(model.coefficients.toArray(), expectedCoefficients, atol=1E-3))
+ self.assertTrue(np.isclose(model.intercept, expectedIntercept, atol=1E-3))
+ self.assertTrue(np.isclose(model.scale, expectedScale, atol=1E-3))
+
+
+if __name__ == "__main__":
+ from pyspark.ml.tests.test_algorithms import *
+
+ try:
+ import xmlrunner
+ testRunner = xmlrunner.XMLTestRunner(output='target/test-reports')
+ except ImportError:
+ testRunner = None
+ unittest.main(testRunner=testRunner, verbosity=2)
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/test_base.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/test_base.py b/python/pyspark/ml/tests/test_base.py
new file mode 100644
index 0000000..59c45f6
--- /dev/null
+++ b/python/pyspark/ml/tests/test_base.py
@@ -0,0 +1,85 @@
+#
+# 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 sys
+if sys.version_info[:2] <= (2, 6):
+ try:
+ import unittest2 as unittest
+ except ImportError:
+ sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+ sys.exit(1)
+else:
+ import unittest
+
+from pyspark.sql.types import DoubleType, IntegerType
+from pyspark.testing.mlutils import MockDataset, MockEstimator, MockUnaryTransformer, \
+ SparkSessionTestCase
+
+
+class UnaryTransformerTests(SparkSessionTestCase):
+
+ def test_unary_transformer_validate_input_type(self):
+ shiftVal = 3
+ transformer = MockUnaryTransformer(shiftVal=shiftVal) \
+ .setInputCol("input").setOutputCol("output")
+
+ # should not raise any errors
+ transformer.validateInputType(DoubleType())
+
+ with self.assertRaises(TypeError):
+ # passing the wrong input type should raise an error
+ transformer.validateInputType(IntegerType())
+
+ def test_unary_transformer_transform(self):
+ shiftVal = 3
+ transformer = MockUnaryTransformer(shiftVal=shiftVal) \
+ .setInputCol("input").setOutputCol("output")
+
+ df = self.spark.range(0, 10).toDF('input')
+ df = df.withColumn("input", df.input.cast(dataType="double"))
+
+ transformed_df = transformer.transform(df)
+ results = transformed_df.select("input", "output").collect()
+
+ for res in results:
+ self.assertEqual(res.input + shiftVal, res.output)
+
+
+class EstimatorTest(unittest.TestCase):
+
+ def testDefaultFitMultiple(self):
+ N = 4
+ data = MockDataset()
+ estimator = MockEstimator()
+ params = [{estimator.fake: i} for i in range(N)]
+ modelIter = estimator.fitMultiple(data, params)
+ indexList = []
+ for index, model in modelIter:
+ self.assertEqual(model.getFake(), index)
+ indexList.append(index)
+ self.assertEqual(sorted(indexList), list(range(N)))
+
+
+if __name__ == "__main__":
+ from pyspark.ml.tests.test_base import *
+
+ try:
+ import xmlrunner
+ testRunner = xmlrunner.XMLTestRunner(output='target/test-reports')
+ except ImportError:
+ testRunner = None
+ unittest.main(testRunner=testRunner, verbosity=2)
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/test_evaluation.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/test_evaluation.py b/python/pyspark/ml/tests/test_evaluation.py
new file mode 100644
index 0000000..6c3e5c6
--- /dev/null
+++ b/python/pyspark/ml/tests/test_evaluation.py
@@ -0,0 +1,71 @@
+#
+# 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 sys
+if sys.version_info[:2] <= (2, 6):
+ try:
+ import unittest2 as unittest
+ except ImportError:
+ sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+ sys.exit(1)
+else:
+ import unittest
+
+import numpy as np
+
+from pyspark.ml.evaluation import ClusteringEvaluator, RegressionEvaluator
+from pyspark.ml.linalg import Vectors
+from pyspark.sql import Row
+from pyspark.testing.mlutils import SparkSessionTestCase
+
+
+class EvaluatorTests(SparkSessionTestCase):
+
+ def test_java_params(self):
+ """
+ This tests a bug fixed by SPARK-18274 which causes multiple copies
+ of a Params instance in Python to be linked to the same Java instance.
+ """
+ evaluator = RegressionEvaluator(metricName="r2")
+ df = self.spark.createDataFrame([Row(label=1.0, prediction=1.1)])
+ evaluator.evaluate(df)
+ self.assertEqual(evaluator._java_obj.getMetricName(), "r2")
+ evaluatorCopy = evaluator.copy({evaluator.metricName: "mae"})
+ evaluator.evaluate(df)
+ evaluatorCopy.evaluate(df)
+ self.assertEqual(evaluator._java_obj.getMetricName(), "r2")
+ self.assertEqual(evaluatorCopy._java_obj.getMetricName(), "mae")
+
+ def test_clustering_evaluator_with_cosine_distance(self):
+ featureAndPredictions = map(lambda x: (Vectors.dense(x[0]), x[1]),
+ [([1.0, 1.0], 1.0), ([10.0, 10.0], 1.0), ([1.0, 0.5], 2.0),
+ ([10.0, 4.4], 2.0), ([-1.0, 1.0], 3.0), ([-100.0, 90.0], 3.0)])
+ dataset = self.spark.createDataFrame(featureAndPredictions, ["features", "prediction"])
+ evaluator = ClusteringEvaluator(predictionCol="prediction", distanceMeasure="cosine")
+ self.assertEqual(evaluator.getDistanceMeasure(), "cosine")
+ self.assertTrue(np.isclose(evaluator.evaluate(dataset), 0.992671213, atol=1e-5))
+
+
+if __name__ == "__main__":
+ from pyspark.ml.tests.test_evaluation import *
+
+ try:
+ import xmlrunner
+ testRunner = xmlrunner.XMLTestRunner(output='target/test-reports')
+ except ImportError:
+ testRunner = None
+ unittest.main(testRunner=testRunner, verbosity=2)
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/test_feature.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/test_feature.py b/python/pyspark/ml/tests/test_feature.py
new file mode 100644
index 0000000..23f66e7
--- /dev/null
+++ b/python/pyspark/ml/tests/test_feature.py
@@ -0,0 +1,318 @@
+# -*- 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 sys
+if sys.version_info[:2] <= (2, 6):
+ try:
+ import unittest2 as unittest
+ except ImportError:
+ sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+ sys.exit(1)
+else:
+ import unittest
+
+if sys.version > '3':
+ basestring = str
+
+from pyspark.ml.feature import Binarizer, CountVectorizer, CountVectorizerModel, HashingTF, IDF, \
+ NGram, RFormula, StopWordsRemover, StringIndexer, StringIndexerModel, VectorSizeHint
+from pyspark.ml.linalg import DenseVector, SparseVector, Vectors
+from pyspark.sql import Row
+from pyspark.testing.utils import QuietTest
+from pyspark.testing.mlutils import check_params, SparkSessionTestCase
+
+
+class FeatureTests(SparkSessionTestCase):
+
+ def test_binarizer(self):
+ b0 = Binarizer()
+ self.assertListEqual(b0.params, [b0.inputCol, b0.outputCol, b0.threshold])
+ self.assertTrue(all([~b0.isSet(p) for p in b0.params]))
+ self.assertTrue(b0.hasDefault(b0.threshold))
+ self.assertEqual(b0.getThreshold(), 0.0)
+ b0.setParams(inputCol="input", outputCol="output").setThreshold(1.0)
+ self.assertTrue(all([b0.isSet(p) for p in b0.params]))
+ self.assertEqual(b0.getThreshold(), 1.0)
+ self.assertEqual(b0.getInputCol(), "input")
+ self.assertEqual(b0.getOutputCol(), "output")
+
+ b0c = b0.copy({b0.threshold: 2.0})
+ self.assertEqual(b0c.uid, b0.uid)
+ self.assertListEqual(b0c.params, b0.params)
+ self.assertEqual(b0c.getThreshold(), 2.0)
+
+ b1 = Binarizer(threshold=2.0, inputCol="input", outputCol="output")
+ self.assertNotEqual(b1.uid, b0.uid)
+ self.assertEqual(b1.getThreshold(), 2.0)
+ self.assertEqual(b1.getInputCol(), "input")
+ self.assertEqual(b1.getOutputCol(), "output")
+
+ def test_idf(self):
+ dataset = self.spark.createDataFrame([
+ (DenseVector([1.0, 2.0]),),
+ (DenseVector([0.0, 1.0]),),
+ (DenseVector([3.0, 0.2]),)], ["tf"])
+ idf0 = IDF(inputCol="tf")
+ self.assertListEqual(idf0.params, [idf0.inputCol, idf0.minDocFreq, idf0.outputCol])
+ idf0m = idf0.fit(dataset, {idf0.outputCol: "idf"})
+ self.assertEqual(idf0m.uid, idf0.uid,
+ "Model should inherit the UID from its parent estimator.")
+ output = idf0m.transform(dataset)
+ self.assertIsNotNone(output.head().idf)
+ # Test that parameters transferred to Python Model
+ check_params(self, idf0m)
+
+ def test_ngram(self):
+ dataset = self.spark.createDataFrame([
+ Row(input=["a", "b", "c", "d", "e"])])
+ ngram0 = NGram(n=4, inputCol="input", outputCol="output")
+ self.assertEqual(ngram0.getN(), 4)
+ self.assertEqual(ngram0.getInputCol(), "input")
+ self.assertEqual(ngram0.getOutputCol(), "output")
+ transformedDF = ngram0.transform(dataset)
+ self.assertEqual(transformedDF.head().output, ["a b c d", "b c d e"])
+
+ def test_stopwordsremover(self):
+ dataset = self.spark.createDataFrame([Row(input=["a", "panda"])])
+ stopWordRemover = StopWordsRemover(inputCol="input", outputCol="output")
+ # Default
+ self.assertEqual(stopWordRemover.getInputCol(), "input")
+ transformedDF = stopWordRemover.transform(dataset)
+ self.assertEqual(transformedDF.head().output, ["panda"])
+ self.assertEqual(type(stopWordRemover.getStopWords()), list)
+ self.assertTrue(isinstance(stopWordRemover.getStopWords()[0], basestring))
+ # Custom
+ stopwords = ["panda"]
+ stopWordRemover.setStopWords(stopwords)
+ self.assertEqual(stopWordRemover.getInputCol(), "input")
+ self.assertEqual(stopWordRemover.getStopWords(), stopwords)
+ transformedDF = stopWordRemover.transform(dataset)
+ self.assertEqual(transformedDF.head().output, ["a"])
+ # with language selection
+ stopwords = StopWordsRemover.loadDefaultStopWords("turkish")
+ dataset = self.spark.createDataFrame([Row(input=["acaba", "ama", "biri"])])
+ stopWordRemover.setStopWords(stopwords)
+ self.assertEqual(stopWordRemover.getStopWords(), stopwords)
+ transformedDF = stopWordRemover.transform(dataset)
+ self.assertEqual(transformedDF.head().output, [])
+ # with locale
+ stopwords = ["BELKÄ°"]
+ dataset = self.spark.createDataFrame([Row(input=["belki"])])
+ stopWordRemover.setStopWords(stopwords).setLocale("tr")
+ self.assertEqual(stopWordRemover.getStopWords(), stopwords)
+ transformedDF = stopWordRemover.transform(dataset)
+ self.assertEqual(transformedDF.head().output, [])
+
+ def test_count_vectorizer_with_binary(self):
+ dataset = self.spark.createDataFrame([
+ (0, "a a a b b c".split(' '), SparseVector(3, {0: 1.0, 1: 1.0, 2: 1.0}),),
+ (1, "a a".split(' '), SparseVector(3, {0: 1.0}),),
+ (2, "a b".split(' '), SparseVector(3, {0: 1.0, 1: 1.0}),),
+ (3, "c".split(' '), SparseVector(3, {2: 1.0}),)], ["id", "words", "expected"])
+ cv = CountVectorizer(binary=True, inputCol="words", outputCol="features")
+ model = cv.fit(dataset)
+
+ transformedList = model.transform(dataset).select("features", "expected").collect()
+
+ for r in transformedList:
+ feature, expected = r
+ self.assertEqual(feature, expected)
+
+ def test_count_vectorizer_with_maxDF(self):
+ dataset = self.spark.createDataFrame([
+ (0, "a b c d".split(' '), SparseVector(3, {0: 1.0, 1: 1.0, 2: 1.0}),),
+ (1, "a b c".split(' '), SparseVector(3, {0: 1.0, 1: 1.0}),),
+ (2, "a b".split(' '), SparseVector(3, {0: 1.0}),),
+ (3, "a".split(' '), SparseVector(3, {}),)], ["id", "words", "expected"])
+ cv = CountVectorizer(inputCol="words", outputCol="features")
+ model1 = cv.setMaxDF(3).fit(dataset)
+ self.assertEqual(model1.vocabulary, ['b', 'c', 'd'])
+
+ transformedList1 = model1.transform(dataset).select("features", "expected").collect()
+
+ for r in transformedList1:
+ feature, expected = r
+ self.assertEqual(feature, expected)
+
+ model2 = cv.setMaxDF(0.75).fit(dataset)
+ self.assertEqual(model2.vocabulary, ['b', 'c', 'd'])
+
+ transformedList2 = model2.transform(dataset).select("features", "expected").collect()
+
+ for r in transformedList2:
+ feature, expected = r
+ self.assertEqual(feature, expected)
+
+ def test_count_vectorizer_from_vocab(self):
+ model = CountVectorizerModel.from_vocabulary(["a", "b", "c"], inputCol="words",
+ outputCol="features", minTF=2)
+ self.assertEqual(model.vocabulary, ["a", "b", "c"])
+ self.assertEqual(model.getMinTF(), 2)
+
+ dataset = self.spark.createDataFrame([
+ (0, "a a a b b c".split(' '), SparseVector(3, {0: 3.0, 1: 2.0}),),
+ (1, "a a".split(' '), SparseVector(3, {0: 2.0}),),
+ (2, "a b".split(' '), SparseVector(3, {}),)], ["id", "words", "expected"])
+
+ transformed_list = model.transform(dataset).select("features", "expected").collect()
+
+ for r in transformed_list:
+ feature, expected = r
+ self.assertEqual(feature, expected)
+
+ # Test an empty vocabulary
+ with QuietTest(self.sc):
+ with self.assertRaisesRegexp(Exception, "vocabSize.*invalid.*0"):
+ CountVectorizerModel.from_vocabulary([], inputCol="words")
+
+ # Test model with default settings can transform
+ model_default = CountVectorizerModel.from_vocabulary(["a", "b", "c"], inputCol="words")
+ transformed_list = model_default.transform(dataset) \
+ .select(model_default.getOrDefault(model_default.outputCol)).collect()
+ self.assertEqual(len(transformed_list), 3)
+
+ def test_rformula_force_index_label(self):
+ df = self.spark.createDataFrame([
+ (1.0, 1.0, "a"),
+ (0.0, 2.0, "b"),
+ (1.0, 0.0, "a")], ["y", "x", "s"])
+ # Does not index label by default since it's numeric type.
+ rf = RFormula(formula="y ~ x + s")
+ model = rf.fit(df)
+ transformedDF = model.transform(df)
+ self.assertEqual(transformedDF.head().label, 1.0)
+ # Force to index label.
+ rf2 = RFormula(formula="y ~ x + s").setForceIndexLabel(True)
+ model2 = rf2.fit(df)
+ transformedDF2 = model2.transform(df)
+ self.assertEqual(transformedDF2.head().label, 0.0)
+
+ def test_rformula_string_indexer_order_type(self):
+ df = self.spark.createDataFrame([
+ (1.0, 1.0, "a"),
+ (0.0, 2.0, "b"),
+ (1.0, 0.0, "a")], ["y", "x", "s"])
+ rf = RFormula(formula="y ~ x + s", stringIndexerOrderType="alphabetDesc")
+ self.assertEqual(rf.getStringIndexerOrderType(), 'alphabetDesc')
+ transformedDF = rf.fit(df).transform(df)
+ observed = transformedDF.select("features").collect()
+ expected = [[1.0, 0.0], [2.0, 1.0], [0.0, 0.0]]
+ for i in range(0, len(expected)):
+ self.assertTrue(all(observed[i]["features"].toArray() == expected[i]))
+
+ def test_string_indexer_handle_invalid(self):
+ df = self.spark.createDataFrame([
+ (0, "a"),
+ (1, "d"),
+ (2, None)], ["id", "label"])
+
+ si1 = StringIndexer(inputCol="label", outputCol="indexed", handleInvalid="keep",
+ stringOrderType="alphabetAsc")
+ model1 = si1.fit(df)
+ td1 = model1.transform(df)
+ actual1 = td1.select("id", "indexed").collect()
+ expected1 = [Row(id=0, indexed=0.0), Row(id=1, indexed=1.0), Row(id=2, indexed=2.0)]
+ self.assertEqual(actual1, expected1)
+
+ si2 = si1.setHandleInvalid("skip")
+ model2 = si2.fit(df)
+ td2 = model2.transform(df)
+ actual2 = td2.select("id", "indexed").collect()
+ expected2 = [Row(id=0, indexed=0.0), Row(id=1, indexed=1.0)]
+ self.assertEqual(actual2, expected2)
+
+ def test_string_indexer_from_labels(self):
+ model = StringIndexerModel.from_labels(["a", "b", "c"], inputCol="label",
+ outputCol="indexed", handleInvalid="keep")
+ self.assertEqual(model.labels, ["a", "b", "c"])
+
+ df1 = self.spark.createDataFrame([
+ (0, "a"),
+ (1, "c"),
+ (2, None),
+ (3, "b"),
+ (4, "b")], ["id", "label"])
+
+ result1 = model.transform(df1)
+ actual1 = result1.select("id", "indexed").collect()
+ expected1 = [Row(id=0, indexed=0.0), Row(id=1, indexed=2.0), Row(id=2, indexed=3.0),
+ Row(id=3, indexed=1.0), Row(id=4, indexed=1.0)]
+ self.assertEqual(actual1, expected1)
+
+ model_empty_labels = StringIndexerModel.from_labels(
+ [], inputCol="label", outputCol="indexed", handleInvalid="keep")
+ actual2 = model_empty_labels.transform(df1).select("id", "indexed").collect()
+ expected2 = [Row(id=0, indexed=0.0), Row(id=1, indexed=0.0), Row(id=2, indexed=0.0),
+ Row(id=3, indexed=0.0), Row(id=4, indexed=0.0)]
+ self.assertEqual(actual2, expected2)
+
+ # Test model with default settings can transform
+ model_default = StringIndexerModel.from_labels(["a", "b", "c"], inputCol="label")
+ df2 = self.spark.createDataFrame([
+ (0, "a"),
+ (1, "c"),
+ (2, "b"),
+ (3, "b"),
+ (4, "b")], ["id", "label"])
+ transformed_list = model_default.transform(df2) \
+ .select(model_default.getOrDefault(model_default.outputCol)).collect()
+ self.assertEqual(len(transformed_list), 5)
+
+ def test_vector_size_hint(self):
+ df = self.spark.createDataFrame(
+ [(0, Vectors.dense([0.0, 10.0, 0.5])),
+ (1, Vectors.dense([1.0, 11.0, 0.5, 0.6])),
+ (2, Vectors.dense([2.0, 12.0]))],
+ ["id", "vector"])
+
+ sizeHint = VectorSizeHint(
+ inputCol="vector",
+ handleInvalid="skip")
+ sizeHint.setSize(3)
+ self.assertEqual(sizeHint.getSize(), 3)
+
+ output = sizeHint.transform(df).head().vector
+ expected = DenseVector([0.0, 10.0, 0.5])
+ self.assertEqual(output, expected)
+
+
+class HashingTFTest(SparkSessionTestCase):
+
+ def test_apply_binary_term_freqs(self):
+
+ df = self.spark.createDataFrame([(0, ["a", "a", "b", "c", "c", "c"])], ["id", "words"])
+ n = 10
+ hashingTF = HashingTF()
+ hashingTF.setInputCol("words").setOutputCol("features").setNumFeatures(n).setBinary(True)
+ output = hashingTF.transform(df)
+ features = output.select("features").first().features.toArray()
+ expected = Vectors.dense([1.0, 1.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]).toArray()
+ for i in range(0, n):
+ self.assertAlmostEqual(features[i], expected[i], 14, "Error at " + str(i) +
+ ": expected " + str(expected[i]) + ", got " + str(features[i]))
+
+
+if __name__ == "__main__":
+ from pyspark.ml.tests.test_feature import *
+
+ try:
+ import xmlrunner
+ testRunner = xmlrunner.XMLTestRunner(output='target/test-reports')
+ except ImportError:
+ testRunner = None
+ unittest.main(testRunner=testRunner, verbosity=2)
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/test_image.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/test_image.py b/python/pyspark/ml/tests/test_image.py
new file mode 100644
index 0000000..dcc7a32
--- /dev/null
+++ b/python/pyspark/ml/tests/test_image.py
@@ -0,0 +1,118 @@
+#
+# 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 sys
+if sys.version_info[:2] <= (2, 6):
+ try:
+ import unittest2 as unittest
+ except ImportError:
+ sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+ sys.exit(1)
+else:
+ import unittest
+
+import py4j
+
+from pyspark.ml.image import ImageSchema
+from pyspark.testing.mlutils import PySparkTestCase, SparkSessionTestCase
+from pyspark.sql import HiveContext, Row
+from pyspark.testing.utils import QuietTest
+
+
+class ImageReaderTest(SparkSessionTestCase):
+
+ def test_read_images(self):
+ data_path = 'data/mllib/images/origin/kittens'
+ df = ImageSchema.readImages(data_path, recursive=True, dropImageFailures=True)
+ self.assertEqual(df.count(), 4)
+ first_row = df.take(1)[0][0]
+ array = ImageSchema.toNDArray(first_row)
+ self.assertEqual(len(array), first_row[1])
+ self.assertEqual(ImageSchema.toImage(array, origin=first_row[0]), first_row)
+ self.assertEqual(df.schema, ImageSchema.imageSchema)
+ self.assertEqual(df.schema["image"].dataType, ImageSchema.columnSchema)
+ expected = {'CV_8UC3': 16, 'Undefined': -1, 'CV_8U': 0, 'CV_8UC1': 0, 'CV_8UC4': 24}
+ self.assertEqual(ImageSchema.ocvTypes, expected)
+ expected = ['origin', 'height', 'width', 'nChannels', 'mode', 'data']
+ self.assertEqual(ImageSchema.imageFields, expected)
+ self.assertEqual(ImageSchema.undefinedImageType, "Undefined")
+
+ with QuietTest(self.sc):
+ self.assertRaisesRegexp(
+ TypeError,
+ "image argument should be pyspark.sql.types.Row; however",
+ lambda: ImageSchema.toNDArray("a"))
+
+ with QuietTest(self.sc):
+ self.assertRaisesRegexp(
+ ValueError,
+ "image argument should have attributes specified in",
+ lambda: ImageSchema.toNDArray(Row(a=1)))
+
+ with QuietTest(self.sc):
+ self.assertRaisesRegexp(
+ TypeError,
+ "array argument should be numpy.ndarray; however, it got",
+ lambda: ImageSchema.toImage("a"))
+
+
+class ImageReaderTest2(PySparkTestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ super(ImageReaderTest2, cls).setUpClass()
+ cls.hive_available = True
+ # Note that here we enable Hive's support.
+ cls.spark = None
+ try:
+ cls.sc._jvm.org.apache.hadoop.hive.conf.HiveConf()
+ except py4j.protocol.Py4JError:
+ cls.tearDownClass()
+ cls.hive_available = False
+ except TypeError:
+ cls.tearDownClass()
+ cls.hive_available = False
+ if cls.hive_available:
+ cls.spark = HiveContext._createForTesting(cls.sc)
+
+ def setUp(self):
+ if not self.hive_available:
+ self.skipTest("Hive is not available.")
+
+ @classmethod
+ def tearDownClass(cls):
+ super(ImageReaderTest2, cls).tearDownClass()
+ if cls.spark is not None:
+ cls.spark.sparkSession.stop()
+ cls.spark = None
+
+ def test_read_images_multiple_times(self):
+ # This test case is to check if `ImageSchema.readImages` tries to
+ # initiate Hive client multiple times. See SPARK-22651.
+ data_path = 'data/mllib/images/origin/kittens'
+ ImageSchema.readImages(data_path, recursive=True, dropImageFailures=True)
+ ImageSchema.readImages(data_path, recursive=True, dropImageFailures=True)
+
+
+if __name__ == "__main__":
+ from pyspark.ml.tests.test_image import *
+
+ try:
+ import xmlrunner
+ testRunner = xmlrunner.XMLTestRunner(output='target/test-reports')
+ except ImportError:
+ testRunner = None
+ unittest.main(testRunner=testRunner, verbosity=2)
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/test_linalg.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/test_linalg.py b/python/pyspark/ml/tests/test_linalg.py
new file mode 100644
index 0000000..76e5386
--- /dev/null
+++ b/python/pyspark/ml/tests/test_linalg.py
@@ -0,0 +1,392 @@
+#
+# 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 sys
+if sys.version_info[:2] <= (2, 6):
+ try:
+ import unittest2 as unittest
+ except ImportError:
+ sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+ sys.exit(1)
+else:
+ import unittest
+
+import array as pyarray
+from numpy import arange, array, array_equal, inf, ones, tile, zeros
+
+from pyspark.ml.linalg import DenseMatrix, DenseVector, MatrixUDT, SparseMatrix, SparseVector, \
+ Vector, VectorUDT, Vectors
+from pyspark.testing.mllibutils import make_serializer, MLlibTestCase
+from pyspark.sql import Row
+
+
+ser = make_serializer()
+
+
+def _squared_distance(a, b):
+ if isinstance(a, Vector):
+ return a.squared_distance(b)
+ else:
+ return b.squared_distance(a)
+
+
+class VectorTests(MLlibTestCase):
+
+ def _test_serialize(self, v):
+ self.assertEqual(v, ser.loads(ser.dumps(v)))
+ jvec = self.sc._jvm.org.apache.spark.ml.python.MLSerDe.loads(bytearray(ser.dumps(v)))
+ nv = ser.loads(bytes(self.sc._jvm.org.apache.spark.ml.python.MLSerDe.dumps(jvec)))
+ self.assertEqual(v, nv)
+ vs = [v] * 100
+ jvecs = self.sc._jvm.org.apache.spark.ml.python.MLSerDe.loads(bytearray(ser.dumps(vs)))
+ nvs = ser.loads(bytes(self.sc._jvm.org.apache.spark.ml.python.MLSerDe.dumps(jvecs)))
+ self.assertEqual(vs, nvs)
+
+ def test_serialize(self):
+ self._test_serialize(DenseVector(range(10)))
+ self._test_serialize(DenseVector(array([1., 2., 3., 4.])))
+ self._test_serialize(DenseVector(pyarray.array('d', range(10))))
+ self._test_serialize(SparseVector(4, {1: 1, 3: 2}))
+ self._test_serialize(SparseVector(3, {}))
+ self._test_serialize(DenseMatrix(2, 3, range(6)))
+ sm1 = SparseMatrix(
+ 3, 4, [0, 2, 2, 4, 4], [1, 2, 1, 2], [1.0, 2.0, 4.0, 5.0])
+ self._test_serialize(sm1)
+
+ def test_dot(self):
+ sv = SparseVector(4, {1: 1, 3: 2})
+ dv = DenseVector(array([1., 2., 3., 4.]))
+ lst = DenseVector([1, 2, 3, 4])
+ mat = array([[1., 2., 3., 4.],
+ [1., 2., 3., 4.],
+ [1., 2., 3., 4.],
+ [1., 2., 3., 4.]])
+ arr = pyarray.array('d', [0, 1, 2, 3])
+ self.assertEqual(10.0, sv.dot(dv))
+ self.assertTrue(array_equal(array([3., 6., 9., 12.]), sv.dot(mat)))
+ self.assertEqual(30.0, dv.dot(dv))
+ self.assertTrue(array_equal(array([10., 20., 30., 40.]), dv.dot(mat)))
+ self.assertEqual(30.0, lst.dot(dv))
+ self.assertTrue(array_equal(array([10., 20., 30., 40.]), lst.dot(mat)))
+ self.assertEqual(7.0, sv.dot(arr))
+
+ def test_squared_distance(self):
+ sv = SparseVector(4, {1: 1, 3: 2})
+ dv = DenseVector(array([1., 2., 3., 4.]))
+ lst = DenseVector([4, 3, 2, 1])
+ lst1 = [4, 3, 2, 1]
+ arr = pyarray.array('d', [0, 2, 1, 3])
+ narr = array([0, 2, 1, 3])
+ self.assertEqual(15.0, _squared_distance(sv, dv))
+ self.assertEqual(25.0, _squared_distance(sv, lst))
+ self.assertEqual(20.0, _squared_distance(dv, lst))
+ self.assertEqual(15.0, _squared_distance(dv, sv))
+ self.assertEqual(25.0, _squared_distance(lst, sv))
+ self.assertEqual(20.0, _squared_distance(lst, dv))
+ self.assertEqual(0.0, _squared_distance(sv, sv))
+ self.assertEqual(0.0, _squared_distance(dv, dv))
+ self.assertEqual(0.0, _squared_distance(lst, lst))
+ self.assertEqual(25.0, _squared_distance(sv, lst1))
+ self.assertEqual(3.0, _squared_distance(sv, arr))
+ self.assertEqual(3.0, _squared_distance(sv, narr))
+
+ def test_hash(self):
+ v1 = DenseVector([0.0, 1.0, 0.0, 5.5])
+ v2 = SparseVector(4, [(1, 1.0), (3, 5.5)])
+ v3 = DenseVector([0.0, 1.0, 0.0, 5.5])
+ v4 = SparseVector(4, [(1, 1.0), (3, 2.5)])
+ self.assertEqual(hash(v1), hash(v2))
+ self.assertEqual(hash(v1), hash(v3))
+ self.assertEqual(hash(v2), hash(v3))
+ self.assertFalse(hash(v1) == hash(v4))
+ self.assertFalse(hash(v2) == hash(v4))
+
+ def test_eq(self):
+ v1 = DenseVector([0.0, 1.0, 0.0, 5.5])
+ v2 = SparseVector(4, [(1, 1.0), (3, 5.5)])
+ v3 = DenseVector([0.0, 1.0, 0.0, 5.5])
+ v4 = SparseVector(6, [(1, 1.0), (3, 5.5)])
+ v5 = DenseVector([0.0, 1.0, 0.0, 2.5])
+ v6 = SparseVector(4, [(1, 1.0), (3, 2.5)])
+ self.assertEqual(v1, v2)
+ self.assertEqual(v1, v3)
+ self.assertFalse(v2 == v4)
+ self.assertFalse(v1 == v5)
+ self.assertFalse(v1 == v6)
+
+ def test_equals(self):
+ indices = [1, 2, 4]
+ values = [1., 3., 2.]
+ self.assertTrue(Vectors._equals(indices, values, list(range(5)), [0., 1., 3., 0., 2.]))
+ self.assertFalse(Vectors._equals(indices, values, list(range(5)), [0., 3., 1., 0., 2.]))
+ self.assertFalse(Vectors._equals(indices, values, list(range(5)), [0., 3., 0., 2.]))
+ self.assertFalse(Vectors._equals(indices, values, list(range(5)), [0., 1., 3., 2., 2.]))
+
+ def test_conversion(self):
+ # numpy arrays should be automatically upcast to float64
+ # tests for fix of [SPARK-5089]
+ v = array([1, 2, 3, 4], dtype='float64')
+ dv = DenseVector(v)
+ self.assertTrue(dv.array.dtype == 'float64')
+ v = array([1, 2, 3, 4], dtype='float32')
+ dv = DenseVector(v)
+ self.assertTrue(dv.array.dtype == 'float64')
+
+ def test_sparse_vector_indexing(self):
+ sv = SparseVector(5, {1: 1, 3: 2})
+ self.assertEqual(sv[0], 0.)
+ self.assertEqual(sv[3], 2.)
+ self.assertEqual(sv[1], 1.)
+ self.assertEqual(sv[2], 0.)
+ self.assertEqual(sv[4], 0.)
+ self.assertEqual(sv[-1], 0.)
+ self.assertEqual(sv[-2], 2.)
+ self.assertEqual(sv[-3], 0.)
+ self.assertEqual(sv[-5], 0.)
+ for ind in [5, -6]:
+ self.assertRaises(IndexError, sv.__getitem__, ind)
+ for ind in [7.8, '1']:
+ self.assertRaises(TypeError, sv.__getitem__, ind)
+
+ zeros = SparseVector(4, {})
+ self.assertEqual(zeros[0], 0.0)
+ self.assertEqual(zeros[3], 0.0)
+ for ind in [4, -5]:
+ self.assertRaises(IndexError, zeros.__getitem__, ind)
+
+ empty = SparseVector(0, {})
+ for ind in [-1, 0, 1]:
+ self.assertRaises(IndexError, empty.__getitem__, ind)
+
+ def test_sparse_vector_iteration(self):
+ self.assertListEqual(list(SparseVector(3, [], [])), [0.0, 0.0, 0.0])
+ self.assertListEqual(list(SparseVector(5, [0, 3], [1.0, 2.0])), [1.0, 0.0, 0.0, 2.0, 0.0])
+
+ def test_matrix_indexing(self):
+ mat = DenseMatrix(3, 2, [0, 1, 4, 6, 8, 10])
+ expected = [[0, 6], [1, 8], [4, 10]]
+ for i in range(3):
+ for j in range(2):
+ self.assertEqual(mat[i, j], expected[i][j])
+
+ for i, j in [(-1, 0), (4, 1), (3, 4)]:
+ self.assertRaises(IndexError, mat.__getitem__, (i, j))
+
+ def test_repr_dense_matrix(self):
+ mat = DenseMatrix(3, 2, [0, 1, 4, 6, 8, 10])
+ self.assertTrue(
+ repr(mat),
+ 'DenseMatrix(3, 2, [0.0, 1.0, 4.0, 6.0, 8.0, 10.0], False)')
+
+ mat = DenseMatrix(3, 2, [0, 1, 4, 6, 8, 10], True)
+ self.assertTrue(
+ repr(mat),
+ 'DenseMatrix(3, 2, [0.0, 1.0, 4.0, 6.0, 8.0, 10.0], False)')
+
+ mat = DenseMatrix(6, 3, zeros(18))
+ self.assertTrue(
+ repr(mat),
+ 'DenseMatrix(6, 3, [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, ..., \
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], False)')
+
+ def test_repr_sparse_matrix(self):
+ sm1t = SparseMatrix(
+ 3, 4, [0, 2, 3, 5], [0, 1, 2, 0, 2], [3.0, 2.0, 4.0, 9.0, 8.0],
+ isTransposed=True)
+ self.assertTrue(
+ repr(sm1t),
+ 'SparseMatrix(3, 4, [0, 2, 3, 5], [0, 1, 2, 0, 2], [3.0, 2.0, 4.0, 9.0, 8.0], True)')
+
+ indices = tile(arange(6), 3)
+ values = ones(18)
+ sm = SparseMatrix(6, 3, [0, 6, 12, 18], indices, values)
+ self.assertTrue(
+ repr(sm), "SparseMatrix(6, 3, [0, 6, 12, 18], \
+ [0, 1, 2, 3, 4, 5, 0, 1, ..., 4, 5, 0, 1, 2, 3, 4, 5], \
+ [1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, ..., \
+ 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0], False)")
+
+ self.assertTrue(
+ str(sm),
+ "6 X 3 CSCMatrix\n\
+ (0,0) 1.0\n(1,0) 1.0\n(2,0) 1.0\n(3,0) 1.0\n(4,0) 1.0\n(5,0) 1.0\n\
+ (0,1) 1.0\n(1,1) 1.0\n(2,1) 1.0\n(3,1) 1.0\n(4,1) 1.0\n(5,1) 1.0\n\
+ (0,2) 1.0\n(1,2) 1.0\n(2,2) 1.0\n(3,2) 1.0\n..\n..")
+
+ sm = SparseMatrix(1, 18, zeros(19), [], [])
+ self.assertTrue(
+ repr(sm),
+ 'SparseMatrix(1, 18, \
+ [0, 0, 0, 0, 0, 0, 0, 0, ..., 0, 0, 0, 0, 0, 0, 0, 0], [], [], False)')
+
+ def test_sparse_matrix(self):
+ # Test sparse matrix creation.
+ sm1 = SparseMatrix(
+ 3, 4, [0, 2, 2, 4, 4], [1, 2, 1, 2], [1.0, 2.0, 4.0, 5.0])
+ self.assertEqual(sm1.numRows, 3)
+ self.assertEqual(sm1.numCols, 4)
+ self.assertEqual(sm1.colPtrs.tolist(), [0, 2, 2, 4, 4])
+ self.assertEqual(sm1.rowIndices.tolist(), [1, 2, 1, 2])
+ self.assertEqual(sm1.values.tolist(), [1.0, 2.0, 4.0, 5.0])
+ self.assertTrue(
+ repr(sm1),
+ 'SparseMatrix(3, 4, [0, 2, 2, 4, 4], [1, 2, 1, 2], [1.0, 2.0, 4.0, 5.0], False)')
+
+ # Test indexing
+ expected = [
+ [0, 0, 0, 0],
+ [1, 0, 4, 0],
+ [2, 0, 5, 0]]
+
+ for i in range(3):
+ for j in range(4):
+ self.assertEqual(expected[i][j], sm1[i, j])
+ self.assertTrue(array_equal(sm1.toArray(), expected))
+
+ for i, j in [(-1, 1), (4, 3), (3, 5)]:
+ self.assertRaises(IndexError, sm1.__getitem__, (i, j))
+
+ # Test conversion to dense and sparse.
+ smnew = sm1.toDense().toSparse()
+ self.assertEqual(sm1.numRows, smnew.numRows)
+ self.assertEqual(sm1.numCols, smnew.numCols)
+ self.assertTrue(array_equal(sm1.colPtrs, smnew.colPtrs))
+ self.assertTrue(array_equal(sm1.rowIndices, smnew.rowIndices))
+ self.assertTrue(array_equal(sm1.values, smnew.values))
+
+ sm1t = SparseMatrix(
+ 3, 4, [0, 2, 3, 5], [0, 1, 2, 0, 2], [3.0, 2.0, 4.0, 9.0, 8.0],
+ isTransposed=True)
+ self.assertEqual(sm1t.numRows, 3)
+ self.assertEqual(sm1t.numCols, 4)
+ self.assertEqual(sm1t.colPtrs.tolist(), [0, 2, 3, 5])
+ self.assertEqual(sm1t.rowIndices.tolist(), [0, 1, 2, 0, 2])
+ self.assertEqual(sm1t.values.tolist(), [3.0, 2.0, 4.0, 9.0, 8.0])
+
+ expected = [
+ [3, 2, 0, 0],
+ [0, 0, 4, 0],
+ [9, 0, 8, 0]]
+
+ for i in range(3):
+ for j in range(4):
+ self.assertEqual(expected[i][j], sm1t[i, j])
+ self.assertTrue(array_equal(sm1t.toArray(), expected))
+
+ def test_dense_matrix_is_transposed(self):
+ mat1 = DenseMatrix(3, 2, [0, 4, 1, 6, 3, 9], isTransposed=True)
+ mat = DenseMatrix(3, 2, [0, 1, 3, 4, 6, 9])
+ self.assertEqual(mat1, mat)
+
+ expected = [[0, 4], [1, 6], [3, 9]]
+ for i in range(3):
+ for j in range(2):
+ self.assertEqual(mat1[i, j], expected[i][j])
+ self.assertTrue(array_equal(mat1.toArray(), expected))
+
+ sm = mat1.toSparse()
+ self.assertTrue(array_equal(sm.rowIndices, [1, 2, 0, 1, 2]))
+ self.assertTrue(array_equal(sm.colPtrs, [0, 2, 5]))
+ self.assertTrue(array_equal(sm.values, [1, 3, 4, 6, 9]))
+
+ def test_norms(self):
+ a = DenseVector([0, 2, 3, -1])
+ self.assertAlmostEqual(a.norm(2), 3.742, 3)
+ self.assertTrue(a.norm(1), 6)
+ self.assertTrue(a.norm(inf), 3)
+ a = SparseVector(4, [0, 2], [3, -4])
+ self.assertAlmostEqual(a.norm(2), 5)
+ self.assertTrue(a.norm(1), 7)
+ self.assertTrue(a.norm(inf), 4)
+
+ tmp = SparseVector(4, [0, 2], [3, 0])
+ self.assertEqual(tmp.numNonzeros(), 1)
+
+
+class VectorUDTTests(MLlibTestCase):
+
+ dv0 = DenseVector([])
+ dv1 = DenseVector([1.0, 2.0])
+ sv0 = SparseVector(2, [], [])
+ sv1 = SparseVector(2, [1], [2.0])
+ udt = VectorUDT()
+
+ def test_json_schema(self):
+ self.assertEqual(VectorUDT.fromJson(self.udt.jsonValue()), self.udt)
+
+ def test_serialization(self):
+ for v in [self.dv0, self.dv1, self.sv0, self.sv1]:
+ self.assertEqual(v, self.udt.deserialize(self.udt.serialize(v)))
+
+ def test_infer_schema(self):
+ rdd = self.sc.parallelize([Row(label=1.0, features=self.dv1),
+ Row(label=0.0, features=self.sv1)])
+ df = rdd.toDF()
+ schema = df.schema
+ field = [f for f in schema.fields if f.name == "features"][0]
+ self.assertEqual(field.dataType, self.udt)
+ vectors = df.rdd.map(lambda p: p.features).collect()
+ self.assertEqual(len(vectors), 2)
+ for v in vectors:
+ if isinstance(v, SparseVector):
+ self.assertEqual(v, self.sv1)
+ elif isinstance(v, DenseVector):
+ self.assertEqual(v, self.dv1)
+ else:
+ raise TypeError("expecting a vector but got %r of type %r" % (v, type(v)))
+
+
+class MatrixUDTTests(MLlibTestCase):
+
+ dm1 = DenseMatrix(3, 2, [0, 1, 4, 5, 9, 10])
+ dm2 = DenseMatrix(3, 2, [0, 1, 4, 5, 9, 10], isTransposed=True)
+ sm1 = SparseMatrix(1, 1, [0, 1], [0], [2.0])
+ sm2 = SparseMatrix(2, 1, [0, 0, 1], [0], [5.0], isTransposed=True)
+ udt = MatrixUDT()
+
+ def test_json_schema(self):
+ self.assertEqual(MatrixUDT.fromJson(self.udt.jsonValue()), self.udt)
+
+ def test_serialization(self):
+ for m in [self.dm1, self.dm2, self.sm1, self.sm2]:
+ self.assertEqual(m, self.udt.deserialize(self.udt.serialize(m)))
+
+ def test_infer_schema(self):
+ rdd = self.sc.parallelize([("dense", self.dm1), ("sparse", self.sm1)])
+ df = rdd.toDF()
+ schema = df.schema
+ self.assertTrue(schema.fields[1].dataType, self.udt)
+ matrices = df.rdd.map(lambda x: x._2).collect()
+ self.assertEqual(len(matrices), 2)
+ for m in matrices:
+ if isinstance(m, DenseMatrix):
+ self.assertTrue(m, self.dm1)
+ elif isinstance(m, SparseMatrix):
+ self.assertTrue(m, self.sm1)
+ else:
+ raise ValueError("Expected a matrix but got type %r" % type(m))
+
+
+if __name__ == "__main__":
+ from pyspark.ml.tests.test_linalg import *
+
+ try:
+ import xmlrunner
+ testRunner = xmlrunner.XMLTestRunner(output='target/test-reports')
+ except ImportError:
+ testRunner = None
+ unittest.main(testRunner=testRunner, verbosity=2)
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests/test_param.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests/test_param.py b/python/pyspark/ml/tests/test_param.py
new file mode 100644
index 0000000..1f36d45
--- /dev/null
+++ b/python/pyspark/ml/tests/test_param.py
@@ -0,0 +1,372 @@
+# -*- 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 inspect
+import sys
+import array as pyarray
+if sys.version_info[:2] <= (2, 6):
+ try:
+ import unittest2 as unittest
+ except ImportError:
+ sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+ sys.exit(1)
+else:
+ import unittest
+
+if sys.version > '3':
+ xrange = range
+
+import numpy as np
+
+from pyspark import keyword_only
+from pyspark.ml.classification import LogisticRegression
+from pyspark.ml.clustering import KMeans
+from pyspark.ml.feature import Binarizer, Bucketizer, ElementwiseProduct, IndexToString, \
+ VectorSlicer, Word2Vec
+from pyspark.ml.linalg import DenseVector, SparseVector
+from pyspark.ml.param import Param, Params, TypeConverters
+from pyspark.ml.param.shared import HasInputCol, HasMaxIter, HasSeed
+from pyspark.ml.wrapper import JavaParams
+from pyspark.testing.mlutils import check_params, PySparkTestCase, SparkSessionTestCase
+
+
+class ParamTypeConversionTests(PySparkTestCase):
+ """
+ Test that param type conversion happens.
+ """
+
+ def test_int(self):
+ lr = LogisticRegression(maxIter=5.0)
+ self.assertEqual(lr.getMaxIter(), 5)
+ self.assertTrue(type(lr.getMaxIter()) == int)
+ self.assertRaises(TypeError, lambda: LogisticRegression(maxIter="notAnInt"))
+ self.assertRaises(TypeError, lambda: LogisticRegression(maxIter=5.1))
+
+ def test_float(self):
+ lr = LogisticRegression(tol=1)
+ self.assertEqual(lr.getTol(), 1.0)
+ self.assertTrue(type(lr.getTol()) == float)
+ self.assertRaises(TypeError, lambda: LogisticRegression(tol="notAFloat"))
+
+ def test_vector(self):
+ ewp = ElementwiseProduct(scalingVec=[1, 3])
+ self.assertEqual(ewp.getScalingVec(), DenseVector([1.0, 3.0]))
+ ewp = ElementwiseProduct(scalingVec=np.array([1.2, 3.4]))
+ self.assertEqual(ewp.getScalingVec(), DenseVector([1.2, 3.4]))
+ self.assertRaises(TypeError, lambda: ElementwiseProduct(scalingVec=["a", "b"]))
+
+ def test_list(self):
+ l = [0, 1]
+ for lst_like in [l, np.array(l), DenseVector(l), SparseVector(len(l), range(len(l)), l),
+ pyarray.array('l', l), xrange(2), tuple(l)]:
+ converted = TypeConverters.toList(lst_like)
+ self.assertEqual(type(converted), list)
+ self.assertListEqual(converted, l)
+
+ def test_list_int(self):
+ for indices in [[1.0, 2.0], np.array([1.0, 2.0]), DenseVector([1.0, 2.0]),
+ SparseVector(2, {0: 1.0, 1: 2.0}), xrange(1, 3), (1.0, 2.0),
+ pyarray.array('d', [1.0, 2.0])]:
+ vs = VectorSlicer(indices=indices)
+ self.assertListEqual(vs.getIndices(), [1, 2])
+ self.assertTrue(all([type(v) == int for v in vs.getIndices()]))
+ self.assertRaises(TypeError, lambda: VectorSlicer(indices=["a", "b"]))
+
+ def test_list_float(self):
+ b = Bucketizer(splits=[1, 4])
+ self.assertEqual(b.getSplits(), [1.0, 4.0])
+ self.assertTrue(all([type(v) == float for v in b.getSplits()]))
+ self.assertRaises(TypeError, lambda: Bucketizer(splits=["a", 1.0]))
+
+ def test_list_string(self):
+ for labels in [np.array(['a', u'b']), ['a', u'b'], np.array(['a', 'b'])]:
+ idx_to_string = IndexToString(labels=labels)
+ self.assertListEqual(idx_to_string.getLabels(), ['a', 'b'])
+ self.assertRaises(TypeError, lambda: IndexToString(labels=['a', 2]))
+
+ def test_string(self):
+ lr = LogisticRegression()
+ for col in ['features', u'features', np.str_('features')]:
+ lr.setFeaturesCol(col)
+ self.assertEqual(lr.getFeaturesCol(), 'features')
+ self.assertRaises(TypeError, lambda: LogisticRegression(featuresCol=2.3))
+
+ def test_bool(self):
+ self.assertRaises(TypeError, lambda: LogisticRegression(fitIntercept=1))
+ self.assertRaises(TypeError, lambda: LogisticRegression(fitIntercept="false"))
+
+
+class TestParams(HasMaxIter, HasInputCol, HasSeed):
+ """
+ A subclass of Params mixed with HasMaxIter, HasInputCol and HasSeed.
+ """
+ @keyword_only
+ def __init__(self, seed=None):
+ super(TestParams, self).__init__()
+ self._setDefault(maxIter=10)
+ kwargs = self._input_kwargs
+ self.setParams(**kwargs)
+
+ @keyword_only
+ def setParams(self, seed=None):
+ """
+ setParams(self, seed=None)
+ Sets params for this test.
+ """
+ kwargs = self._input_kwargs
+ return self._set(**kwargs)
+
+
+class OtherTestParams(HasMaxIter, HasInputCol, HasSeed):
+ """
+ A subclass of Params mixed with HasMaxIter, HasInputCol and HasSeed.
+ """
+ @keyword_only
+ def __init__(self, seed=None):
+ super(OtherTestParams, self).__init__()
+ self._setDefault(maxIter=10)
+ kwargs = self._input_kwargs
+ self.setParams(**kwargs)
+
+ @keyword_only
+ def setParams(self, seed=None):
+ """
+ setParams(self, seed=None)
+ Sets params for this test.
+ """
+ kwargs = self._input_kwargs
+ return self._set(**kwargs)
+
+
+class HasThrowableProperty(Params):
+
+ def __init__(self):
+ super(HasThrowableProperty, self).__init__()
+ self.p = Param(self, "none", "empty param")
+
+ @property
+ def test_property(self):
+ raise RuntimeError("Test property to raise error when invoked")
+
+
+class ParamTests(SparkSessionTestCase):
+
+ def test_copy_new_parent(self):
+ testParams = TestParams()
+ # Copying an instantiated param should fail
+ with self.assertRaises(ValueError):
+ testParams.maxIter._copy_new_parent(testParams)
+ # Copying a dummy param should succeed
+ TestParams.maxIter._copy_new_parent(testParams)
+ maxIter = testParams.maxIter
+ self.assertEqual(maxIter.name, "maxIter")
+ self.assertEqual(maxIter.doc, "max number of iterations (>= 0).")
+ self.assertTrue(maxIter.parent == testParams.uid)
+
+ def test_param(self):
+ testParams = TestParams()
+ maxIter = testParams.maxIter
+ self.assertEqual(maxIter.name, "maxIter")
+ self.assertEqual(maxIter.doc, "max number of iterations (>= 0).")
+ self.assertTrue(maxIter.parent == testParams.uid)
+
+ def test_hasparam(self):
+ testParams = TestParams()
+ self.assertTrue(all([testParams.hasParam(p.name) for p in testParams.params]))
+ self.assertFalse(testParams.hasParam("notAParameter"))
+ self.assertTrue(testParams.hasParam(u"maxIter"))
+
+ def test_resolveparam(self):
+ testParams = TestParams()
+ self.assertEqual(testParams._resolveParam(testParams.maxIter), testParams.maxIter)
+ self.assertEqual(testParams._resolveParam("maxIter"), testParams.maxIter)
+
+ self.assertEqual(testParams._resolveParam(u"maxIter"), testParams.maxIter)
+ if sys.version_info[0] >= 3:
+ # In Python 3, it is allowed to get/set attributes with non-ascii characters.
+ e_cls = AttributeError
+ else:
+ e_cls = UnicodeEncodeError
+ self.assertRaises(e_cls, lambda: testParams._resolveParam(u"ì•„"))
+
+ def test_params(self):
+ testParams = TestParams()
+ maxIter = testParams.maxIter
+ inputCol = testParams.inputCol
+ seed = testParams.seed
+
+ params = testParams.params
+ self.assertEqual(params, [inputCol, maxIter, seed])
+
+ self.assertTrue(testParams.hasParam(maxIter.name))
+ self.assertTrue(testParams.hasDefault(maxIter))
+ self.assertFalse(testParams.isSet(maxIter))
+ self.assertTrue(testParams.isDefined(maxIter))
+ self.assertEqual(testParams.getMaxIter(), 10)
+ testParams.setMaxIter(100)
+ self.assertTrue(testParams.isSet(maxIter))
+ self.assertEqual(testParams.getMaxIter(), 100)
+
+ self.assertTrue(testParams.hasParam(inputCol.name))
+ self.assertFalse(testParams.hasDefault(inputCol))
+ self.assertFalse(testParams.isSet(inputCol))
+ self.assertFalse(testParams.isDefined(inputCol))
+ with self.assertRaises(KeyError):
+ testParams.getInputCol()
+
+ otherParam = Param(Params._dummy(), "otherParam", "Parameter used to test that " +
+ "set raises an error for a non-member parameter.",
+ typeConverter=TypeConverters.toString)
+ with self.assertRaises(ValueError):
+ testParams.set(otherParam, "value")
+
+ # Since the default is normally random, set it to a known number for debug str
+ testParams._setDefault(seed=41)
+ testParams.setSeed(43)
+
+ self.assertEqual(
+ testParams.explainParams(),
+ "\n".join(["inputCol: input column name. (undefined)",
+ "maxIter: max number of iterations (>= 0). (default: 10, current: 100)",
+ "seed: random seed. (default: 41, current: 43)"]))
+
+ def test_kmeans_param(self):
+ algo = KMeans()
+ self.assertEqual(algo.getInitMode(), "k-means||")
+ algo.setK(10)
+ self.assertEqual(algo.getK(), 10)
+ algo.setInitSteps(10)
+ self.assertEqual(algo.getInitSteps(), 10)
+ self.assertEqual(algo.getDistanceMeasure(), "euclidean")
+ algo.setDistanceMeasure("cosine")
+ self.assertEqual(algo.getDistanceMeasure(), "cosine")
+
+ def test_hasseed(self):
+ noSeedSpecd = TestParams()
+ withSeedSpecd = TestParams(seed=42)
+ other = OtherTestParams()
+ # Check that we no longer use 42 as the magic number
+ self.assertNotEqual(noSeedSpecd.getSeed(), 42)
+ origSeed = noSeedSpecd.getSeed()
+ # Check that we only compute the seed once
+ self.assertEqual(noSeedSpecd.getSeed(), origSeed)
+ # Check that a specified seed is honored
+ self.assertEqual(withSeedSpecd.getSeed(), 42)
+ # Check that a different class has a different seed
+ self.assertNotEqual(other.getSeed(), noSeedSpecd.getSeed())
+
+ def test_param_property_error(self):
+ param_store = HasThrowableProperty()
+ self.assertRaises(RuntimeError, lambda: param_store.test_property)
+ params = param_store.params # should not invoke the property 'test_property'
+ self.assertEqual(len(params), 1)
+
+ def test_word2vec_param(self):
+ model = Word2Vec().setWindowSize(6)
+ # Check windowSize is set properly
+ self.assertEqual(model.getWindowSize(), 6)
+
+ def test_copy_param_extras(self):
+ tp = TestParams(seed=42)
+ extra = {tp.getParam(TestParams.inputCol.name): "copy_input"}
+ tp_copy = tp.copy(extra=extra)
+ self.assertEqual(tp.uid, tp_copy.uid)
+ self.assertEqual(tp.params, tp_copy.params)
+ for k, v in extra.items():
+ self.assertTrue(tp_copy.isDefined(k))
+ self.assertEqual(tp_copy.getOrDefault(k), v)
+ copied_no_extra = {}
+ for k, v in tp_copy._paramMap.items():
+ if k not in extra:
+ copied_no_extra[k] = v
+ self.assertEqual(tp._paramMap, copied_no_extra)
+ self.assertEqual(tp._defaultParamMap, tp_copy._defaultParamMap)
+
+ def test_logistic_regression_check_thresholds(self):
+ self.assertIsInstance(
+ LogisticRegression(threshold=0.5, thresholds=[0.5, 0.5]),
+ LogisticRegression
+ )
+
+ self.assertRaisesRegexp(
+ ValueError,
+ "Logistic Regression getThreshold found inconsistent.*$",
+ LogisticRegression, threshold=0.42, thresholds=[0.5, 0.5]
+ )
+
+ def test_preserve_set_state(self):
+ dataset = self.spark.createDataFrame([(0.5,)], ["data"])
+ binarizer = Binarizer(inputCol="data")
+ self.assertFalse(binarizer.isSet("threshold"))
+ binarizer.transform(dataset)
+ binarizer._transfer_params_from_java()
+ self.assertFalse(binarizer.isSet("threshold"),
+ "Params not explicitly set should remain unset after transform")
+
+ def test_default_params_transferred(self):
+ dataset = self.spark.createDataFrame([(0.5,)], ["data"])
+ binarizer = Binarizer(inputCol="data")
+ # intentionally change the pyspark default, but don't set it
+ binarizer._defaultParamMap[binarizer.outputCol] = "my_default"
+ result = binarizer.transform(dataset).select("my_default").collect()
+ self.assertFalse(binarizer.isSet(binarizer.outputCol))
+ self.assertEqual(result[0][0], 1.0)
+
+
+class DefaultValuesTests(PySparkTestCase):
+ """
+ Test :py:class:`JavaParams` classes to see if their default Param values match
+ those in their Scala counterparts.
+ """
+
+ def test_java_params(self):
+ import pyspark.ml.feature
+ import pyspark.ml.classification
+ import pyspark.ml.clustering
+ import pyspark.ml.evaluation
+ import pyspark.ml.pipeline
+ import pyspark.ml.recommendation
+ import pyspark.ml.regression
+
+ modules = [pyspark.ml.feature, pyspark.ml.classification, pyspark.ml.clustering,
+ pyspark.ml.evaluation, pyspark.ml.pipeline, pyspark.ml.recommendation,
+ pyspark.ml.regression]
+ for module in modules:
+ for name, cls in inspect.getmembers(module, inspect.isclass):
+ if not name.endswith('Model') and not name.endswith('Params') \
+ and issubclass(cls, JavaParams) and not inspect.isabstract(cls):
+ # NOTE: disable check_params_exist until there is parity with Scala API
+ check_params(self, cls(), check_params_exist=False)
+
+ # Additional classes that need explicit construction
+ from pyspark.ml.feature import CountVectorizerModel, StringIndexerModel
+ check_params(self, CountVectorizerModel.from_vocabulary(['a'], 'input'),
+ check_params_exist=False)
+ check_params(self, StringIndexerModel.from_labels(['a', 'b'], 'input'),
+ check_params_exist=False)
+
+
+if __name__ == "__main__":
+ from pyspark.ml.tests.test_param import *
+
+ try:
+ import xmlrunner
+ testRunner = xmlrunner.XMLTestRunner(output='target/test-reports')
+ except ImportError:
+ testRunner = None
+ unittest.main(testRunner=testRunner, verbosity=2)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org
[3/4] spark git commit: [SPARK-26033][PYTHON][TESTS] Break large
ml/tests.py file into smaller files
Posted by gu...@apache.org.
http://git-wip-us.apache.org/repos/asf/spark/blob/034ae305/python/pyspark/ml/tests.py
----------------------------------------------------------------------
diff --git a/python/pyspark/ml/tests.py b/python/pyspark/ml/tests.py
deleted file mode 100755
index 2b4b731..0000000
--- a/python/pyspark/ml/tests.py
+++ /dev/null
@@ -1,2762 +0,0 @@
-# -*- 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.
-#
-
-"""
-Unit tests for MLlib Python DataFrame-based APIs.
-"""
-import sys
-if sys.version > '3':
- xrange = range
- basestring = str
-
-try:
- import xmlrunner
-except ImportError:
- xmlrunner = None
-
-if sys.version_info[:2] <= (2, 6):
- try:
- import unittest2 as unittest
- except ImportError:
- sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
- sys.exit(1)
-else:
- import unittest
-
-from shutil import rmtree
-import tempfile
-import array as pyarray
-import numpy as np
-from numpy import abs, all, arange, array, array_equal, inf, ones, tile, zeros
-import inspect
-import py4j
-
-from pyspark import keyword_only, SparkContext
-from pyspark.ml import Estimator, Model, Pipeline, PipelineModel, Transformer, UnaryTransformer
-from pyspark.ml.classification import *
-from pyspark.ml.clustering import *
-from pyspark.ml.common import _java2py, _py2java
-from pyspark.ml.evaluation import BinaryClassificationEvaluator, ClusteringEvaluator, \
- MulticlassClassificationEvaluator, RegressionEvaluator
-from pyspark.ml.feature import *
-from pyspark.ml.fpm import FPGrowth, FPGrowthModel
-from pyspark.ml.image import ImageSchema
-from pyspark.ml.linalg import DenseMatrix, DenseMatrix, DenseVector, Matrices, MatrixUDT, \
- SparseMatrix, SparseVector, Vector, VectorUDT, Vectors
-from pyspark.ml.param import Param, Params, TypeConverters
-from pyspark.ml.param.shared import HasInputCol, HasMaxIter, HasSeed
-from pyspark.ml.recommendation import ALS
-from pyspark.ml.regression import DecisionTreeRegressor, GeneralizedLinearRegression, \
- LinearRegression
-from pyspark.ml.stat import ChiSquareTest
-from pyspark.ml.tuning import *
-from pyspark.ml.util import *
-from pyspark.ml.wrapper import JavaParams, JavaWrapper
-from pyspark.serializers import PickleSerializer
-from pyspark.sql import DataFrame, Row, SparkSession, HiveContext
-from pyspark.sql.functions import rand
-from pyspark.sql.types import DoubleType, IntegerType
-from pyspark.storagelevel import *
-from pyspark.testing.utils import QuietTest, ReusedPySparkTestCase as PySparkTestCase
-
-ser = PickleSerializer()
-
-
-class MLlibTestCase(unittest.TestCase):
- def setUp(self):
- self.sc = SparkContext('local[4]', "MLlib tests")
- self.spark = SparkSession(self.sc)
-
- def tearDown(self):
- self.spark.stop()
-
-
-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(), "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 JavaWrapperMemoryTests(SparkSessionTestCase):
-
- def test_java_object_gets_detached(self):
- df = self.spark.createDataFrame([(1.0, 2.0, Vectors.dense(1.0)),
- (0.0, 2.0, Vectors.sparse(1, [], []))],
- ["label", "weight", "features"])
- lr = LinearRegression(maxIter=1, regParam=0.0, solver="normal", weightCol="weight",
- fitIntercept=False)
-
- model = lr.fit(df)
- summary = model.summary
-
- self.assertIsInstance(model, JavaWrapper)
- self.assertIsInstance(summary, JavaWrapper)
- self.assertIsInstance(model, JavaParams)
- self.assertNotIsInstance(summary, JavaParams)
-
- error_no_object = 'Target Object ID does not exist for this gateway'
-
- self.assertIn("LinearRegression_", model._java_obj.toString())
- self.assertIn("LinearRegressionTrainingSummary", summary._java_obj.toString())
-
- model.__del__()
-
- with self.assertRaisesRegexp(py4j.protocol.Py4JError, error_no_object):
- model._java_obj.toString()
- self.assertIn("LinearRegressionTrainingSummary", summary._java_obj.toString())
-
- try:
- summary.__del__()
- except:
- pass
-
- with self.assertRaisesRegexp(py4j.protocol.Py4JError, error_no_object):
- model._java_obj.toString()
- with self.assertRaisesRegexp(py4j.protocol.Py4JError, error_no_object):
- summary._java_obj.toString()
-
-
-class ParamTypeConversionTests(PySparkTestCase):
- """
- Test that param type conversion happens.
- """
-
- def test_int(self):
- lr = LogisticRegression(maxIter=5.0)
- self.assertEqual(lr.getMaxIter(), 5)
- self.assertTrue(type(lr.getMaxIter()) == int)
- self.assertRaises(TypeError, lambda: LogisticRegression(maxIter="notAnInt"))
- self.assertRaises(TypeError, lambda: LogisticRegression(maxIter=5.1))
-
- def test_float(self):
- lr = LogisticRegression(tol=1)
- self.assertEqual(lr.getTol(), 1.0)
- self.assertTrue(type(lr.getTol()) == float)
- self.assertRaises(TypeError, lambda: LogisticRegression(tol="notAFloat"))
-
- def test_vector(self):
- ewp = ElementwiseProduct(scalingVec=[1, 3])
- self.assertEqual(ewp.getScalingVec(), DenseVector([1.0, 3.0]))
- ewp = ElementwiseProduct(scalingVec=np.array([1.2, 3.4]))
- self.assertEqual(ewp.getScalingVec(), DenseVector([1.2, 3.4]))
- self.assertRaises(TypeError, lambda: ElementwiseProduct(scalingVec=["a", "b"]))
-
- def test_list(self):
- l = [0, 1]
- for lst_like in [l, np.array(l), DenseVector(l), SparseVector(len(l),
- range(len(l)), l), pyarray.array('l', l), xrange(2), tuple(l)]:
- converted = TypeConverters.toList(lst_like)
- self.assertEqual(type(converted), list)
- self.assertListEqual(converted, l)
-
- def test_list_int(self):
- for indices in [[1.0, 2.0], np.array([1.0, 2.0]), DenseVector([1.0, 2.0]),
- SparseVector(2, {0: 1.0, 1: 2.0}), xrange(1, 3), (1.0, 2.0),
- pyarray.array('d', [1.0, 2.0])]:
- vs = VectorSlicer(indices=indices)
- self.assertListEqual(vs.getIndices(), [1, 2])
- self.assertTrue(all([type(v) == int for v in vs.getIndices()]))
- self.assertRaises(TypeError, lambda: VectorSlicer(indices=["a", "b"]))
-
- def test_list_float(self):
- b = Bucketizer(splits=[1, 4])
- self.assertEqual(b.getSplits(), [1.0, 4.0])
- self.assertTrue(all([type(v) == float for v in b.getSplits()]))
- self.assertRaises(TypeError, lambda: Bucketizer(splits=["a", 1.0]))
-
- def test_list_string(self):
- for labels in [np.array(['a', u'b']), ['a', u'b'], np.array(['a', 'b'])]:
- idx_to_string = IndexToString(labels=labels)
- self.assertListEqual(idx_to_string.getLabels(), ['a', 'b'])
- self.assertRaises(TypeError, lambda: IndexToString(labels=['a', 2]))
-
- def test_string(self):
- lr = LogisticRegression()
- for col in ['features', u'features', np.str_('features')]:
- lr.setFeaturesCol(col)
- self.assertEqual(lr.getFeaturesCol(), 'features')
- self.assertRaises(TypeError, lambda: LogisticRegression(featuresCol=2.3))
-
- def test_bool(self):
- self.assertRaises(TypeError, lambda: LogisticRegression(fitIntercept=1))
- self.assertRaises(TypeError, lambda: LogisticRegression(fitIntercept="false"))
-
-
-class PipelineTests(PySparkTestCase):
-
- def test_pipeline(self):
- dataset = MockDataset()
- estimator0 = MockEstimator()
- transformer1 = MockTransformer()
- estimator2 = MockEstimator()
- transformer3 = MockTransformer()
- pipeline = Pipeline(stages=[estimator0, transformer1, estimator2, transformer3])
- pipeline_model = pipeline.fit(dataset, {estimator0.fake: 0, transformer1.fake: 1})
- model0, transformer1, model2, transformer3 = pipeline_model.stages
- self.assertEqual(0, model0.dataset_index)
- self.assertEqual(0, model0.getFake())
- self.assertEqual(1, transformer1.dataset_index)
- self.assertEqual(1, transformer1.getFake())
- self.assertEqual(2, dataset.index)
- self.assertIsNone(model2.dataset_index, "The last model shouldn't be called in fit.")
- self.assertIsNone(transformer3.dataset_index,
- "The last transformer shouldn't be called in fit.")
- dataset = pipeline_model.transform(dataset)
- self.assertEqual(2, model0.dataset_index)
- self.assertEqual(3, transformer1.dataset_index)
- self.assertEqual(4, model2.dataset_index)
- self.assertEqual(5, transformer3.dataset_index)
- self.assertEqual(6, dataset.index)
-
- def test_identity_pipeline(self):
- dataset = MockDataset()
-
- def doTransform(pipeline):
- pipeline_model = pipeline.fit(dataset)
- return pipeline_model.transform(dataset)
- # check that empty pipeline did not perform any transformation
- self.assertEqual(dataset.index, doTransform(Pipeline(stages=[])).index)
- # check that failure to set stages param will raise KeyError for missing param
- self.assertRaises(KeyError, lambda: doTransform(Pipeline()))
-
-
-class TestParams(HasMaxIter, HasInputCol, HasSeed):
- """
- A subclass of Params mixed with HasMaxIter, HasInputCol and HasSeed.
- """
- @keyword_only
- def __init__(self, seed=None):
- super(TestParams, self).__init__()
- self._setDefault(maxIter=10)
- kwargs = self._input_kwargs
- self.setParams(**kwargs)
-
- @keyword_only
- def setParams(self, seed=None):
- """
- setParams(self, seed=None)
- Sets params for this test.
- """
- kwargs = self._input_kwargs
- return self._set(**kwargs)
-
-
-class OtherTestParams(HasMaxIter, HasInputCol, HasSeed):
- """
- A subclass of Params mixed with HasMaxIter, HasInputCol and HasSeed.
- """
- @keyword_only
- def __init__(self, seed=None):
- super(OtherTestParams, self).__init__()
- self._setDefault(maxIter=10)
- kwargs = self._input_kwargs
- self.setParams(**kwargs)
-
- @keyword_only
- def setParams(self, seed=None):
- """
- setParams(self, seed=None)
- Sets params for this test.
- """
- kwargs = self._input_kwargs
- return self._set(**kwargs)
-
-
-class HasThrowableProperty(Params):
-
- def __init__(self):
- super(HasThrowableProperty, self).__init__()
- self.p = Param(self, "none", "empty param")
-
- @property
- def test_property(self):
- raise RuntimeError("Test property to raise error when invoked")
-
-
-class ParamTests(SparkSessionTestCase):
-
- def test_copy_new_parent(self):
- testParams = TestParams()
- # Copying an instantiated param should fail
- with self.assertRaises(ValueError):
- testParams.maxIter._copy_new_parent(testParams)
- # Copying a dummy param should succeed
- TestParams.maxIter._copy_new_parent(testParams)
- maxIter = testParams.maxIter
- self.assertEqual(maxIter.name, "maxIter")
- self.assertEqual(maxIter.doc, "max number of iterations (>= 0).")
- self.assertTrue(maxIter.parent == testParams.uid)
-
- def test_param(self):
- testParams = TestParams()
- maxIter = testParams.maxIter
- self.assertEqual(maxIter.name, "maxIter")
- self.assertEqual(maxIter.doc, "max number of iterations (>= 0).")
- self.assertTrue(maxIter.parent == testParams.uid)
-
- def test_hasparam(self):
- testParams = TestParams()
- self.assertTrue(all([testParams.hasParam(p.name) for p in testParams.params]))
- self.assertFalse(testParams.hasParam("notAParameter"))
- self.assertTrue(testParams.hasParam(u"maxIter"))
-
- def test_resolveparam(self):
- testParams = TestParams()
- self.assertEqual(testParams._resolveParam(testParams.maxIter), testParams.maxIter)
- self.assertEqual(testParams._resolveParam("maxIter"), testParams.maxIter)
-
- self.assertEqual(testParams._resolveParam(u"maxIter"), testParams.maxIter)
- if sys.version_info[0] >= 3:
- # In Python 3, it is allowed to get/set attributes with non-ascii characters.
- e_cls = AttributeError
- else:
- e_cls = UnicodeEncodeError
- self.assertRaises(e_cls, lambda: testParams._resolveParam(u"ì•„"))
-
- def test_params(self):
- testParams = TestParams()
- maxIter = testParams.maxIter
- inputCol = testParams.inputCol
- seed = testParams.seed
-
- params = testParams.params
- self.assertEqual(params, [inputCol, maxIter, seed])
-
- self.assertTrue(testParams.hasParam(maxIter.name))
- self.assertTrue(testParams.hasDefault(maxIter))
- self.assertFalse(testParams.isSet(maxIter))
- self.assertTrue(testParams.isDefined(maxIter))
- self.assertEqual(testParams.getMaxIter(), 10)
- testParams.setMaxIter(100)
- self.assertTrue(testParams.isSet(maxIter))
- self.assertEqual(testParams.getMaxIter(), 100)
-
- self.assertTrue(testParams.hasParam(inputCol.name))
- self.assertFalse(testParams.hasDefault(inputCol))
- self.assertFalse(testParams.isSet(inputCol))
- self.assertFalse(testParams.isDefined(inputCol))
- with self.assertRaises(KeyError):
- testParams.getInputCol()
-
- otherParam = Param(Params._dummy(), "otherParam", "Parameter used to test that " +
- "set raises an error for a non-member parameter.",
- typeConverter=TypeConverters.toString)
- with self.assertRaises(ValueError):
- testParams.set(otherParam, "value")
-
- # Since the default is normally random, set it to a known number for debug str
- testParams._setDefault(seed=41)
- testParams.setSeed(43)
-
- self.assertEqual(
- testParams.explainParams(),
- "\n".join(["inputCol: input column name. (undefined)",
- "maxIter: max number of iterations (>= 0). (default: 10, current: 100)",
- "seed: random seed. (default: 41, current: 43)"]))
-
- def test_kmeans_param(self):
- algo = KMeans()
- self.assertEqual(algo.getInitMode(), "k-means||")
- algo.setK(10)
- self.assertEqual(algo.getK(), 10)
- algo.setInitSteps(10)
- self.assertEqual(algo.getInitSteps(), 10)
- self.assertEqual(algo.getDistanceMeasure(), "euclidean")
- algo.setDistanceMeasure("cosine")
- self.assertEqual(algo.getDistanceMeasure(), "cosine")
-
- def test_hasseed(self):
- noSeedSpecd = TestParams()
- withSeedSpecd = TestParams(seed=42)
- other = OtherTestParams()
- # Check that we no longer use 42 as the magic number
- self.assertNotEqual(noSeedSpecd.getSeed(), 42)
- origSeed = noSeedSpecd.getSeed()
- # Check that we only compute the seed once
- self.assertEqual(noSeedSpecd.getSeed(), origSeed)
- # Check that a specified seed is honored
- self.assertEqual(withSeedSpecd.getSeed(), 42)
- # Check that a different class has a different seed
- self.assertNotEqual(other.getSeed(), noSeedSpecd.getSeed())
-
- def test_param_property_error(self):
- param_store = HasThrowableProperty()
- self.assertRaises(RuntimeError, lambda: param_store.test_property)
- params = param_store.params # should not invoke the property 'test_property'
- self.assertEqual(len(params), 1)
-
- def test_word2vec_param(self):
- model = Word2Vec().setWindowSize(6)
- # Check windowSize is set properly
- self.assertEqual(model.getWindowSize(), 6)
-
- def test_copy_param_extras(self):
- tp = TestParams(seed=42)
- extra = {tp.getParam(TestParams.inputCol.name): "copy_input"}
- tp_copy = tp.copy(extra=extra)
- self.assertEqual(tp.uid, tp_copy.uid)
- self.assertEqual(tp.params, tp_copy.params)
- for k, v in extra.items():
- self.assertTrue(tp_copy.isDefined(k))
- self.assertEqual(tp_copy.getOrDefault(k), v)
- copied_no_extra = {}
- for k, v in tp_copy._paramMap.items():
- if k not in extra:
- copied_no_extra[k] = v
- self.assertEqual(tp._paramMap, copied_no_extra)
- self.assertEqual(tp._defaultParamMap, tp_copy._defaultParamMap)
-
- def test_logistic_regression_check_thresholds(self):
- self.assertIsInstance(
- LogisticRegression(threshold=0.5, thresholds=[0.5, 0.5]),
- LogisticRegression
- )
-
- self.assertRaisesRegexp(
- ValueError,
- "Logistic Regression getThreshold found inconsistent.*$",
- LogisticRegression, threshold=0.42, thresholds=[0.5, 0.5]
- )
-
- def test_preserve_set_state(self):
- dataset = self.spark.createDataFrame([(0.5,)], ["data"])
- binarizer = Binarizer(inputCol="data")
- self.assertFalse(binarizer.isSet("threshold"))
- binarizer.transform(dataset)
- binarizer._transfer_params_from_java()
- self.assertFalse(binarizer.isSet("threshold"),
- "Params not explicitly set should remain unset after transform")
-
- def test_default_params_transferred(self):
- dataset = self.spark.createDataFrame([(0.5,)], ["data"])
- binarizer = Binarizer(inputCol="data")
- # intentionally change the pyspark default, but don't set it
- binarizer._defaultParamMap[binarizer.outputCol] = "my_default"
- result = binarizer.transform(dataset).select("my_default").collect()
- self.assertFalse(binarizer.isSet(binarizer.outputCol))
- self.assertEqual(result[0][0], 1.0)
-
- @staticmethod
- def check_params(test_self, py_stage, check_params_exist=True):
- """
- Checks common requirements for 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)
- if not hasattr(py_stage, "_to_java"):
- return
- java_stage = py_stage._to_java()
- if java_stage is None:
- return
- test_self.assertEqual(py_stage.uid, java_stage.uid(), msg=py_stage_str)
- 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 EvaluatorTests(SparkSessionTestCase):
-
- def test_java_params(self):
- """
- This tests a bug fixed by SPARK-18274 which causes multiple copies
- of a Params instance in Python to be linked to the same Java instance.
- """
- evaluator = RegressionEvaluator(metricName="r2")
- df = self.spark.createDataFrame([Row(label=1.0, prediction=1.1)])
- evaluator.evaluate(df)
- self.assertEqual(evaluator._java_obj.getMetricName(), "r2")
- evaluatorCopy = evaluator.copy({evaluator.metricName: "mae"})
- evaluator.evaluate(df)
- evaluatorCopy.evaluate(df)
- self.assertEqual(evaluator._java_obj.getMetricName(), "r2")
- self.assertEqual(evaluatorCopy._java_obj.getMetricName(), "mae")
-
- def test_clustering_evaluator_with_cosine_distance(self):
- featureAndPredictions = map(lambda x: (Vectors.dense(x[0]), x[1]),
- [([1.0, 1.0], 1.0), ([10.0, 10.0], 1.0), ([1.0, 0.5], 2.0),
- ([10.0, 4.4], 2.0), ([-1.0, 1.0], 3.0), ([-100.0, 90.0], 3.0)])
- dataset = self.spark.createDataFrame(featureAndPredictions, ["features", "prediction"])
- evaluator = ClusteringEvaluator(predictionCol="prediction", distanceMeasure="cosine")
- self.assertEqual(evaluator.getDistanceMeasure(), "cosine")
- self.assertTrue(np.isclose(evaluator.evaluate(dataset), 0.992671213, atol=1e-5))
-
-
-class FeatureTests(SparkSessionTestCase):
-
- def test_binarizer(self):
- b0 = Binarizer()
- self.assertListEqual(b0.params, [b0.inputCol, b0.outputCol, b0.threshold])
- self.assertTrue(all([~b0.isSet(p) for p in b0.params]))
- self.assertTrue(b0.hasDefault(b0.threshold))
- self.assertEqual(b0.getThreshold(), 0.0)
- b0.setParams(inputCol="input", outputCol="output").setThreshold(1.0)
- self.assertTrue(all([b0.isSet(p) for p in b0.params]))
- self.assertEqual(b0.getThreshold(), 1.0)
- self.assertEqual(b0.getInputCol(), "input")
- self.assertEqual(b0.getOutputCol(), "output")
-
- b0c = b0.copy({b0.threshold: 2.0})
- self.assertEqual(b0c.uid, b0.uid)
- self.assertListEqual(b0c.params, b0.params)
- self.assertEqual(b0c.getThreshold(), 2.0)
-
- b1 = Binarizer(threshold=2.0, inputCol="input", outputCol="output")
- self.assertNotEqual(b1.uid, b0.uid)
- self.assertEqual(b1.getThreshold(), 2.0)
- self.assertEqual(b1.getInputCol(), "input")
- self.assertEqual(b1.getOutputCol(), "output")
-
- def test_idf(self):
- dataset = self.spark.createDataFrame([
- (DenseVector([1.0, 2.0]),),
- (DenseVector([0.0, 1.0]),),
- (DenseVector([3.0, 0.2]),)], ["tf"])
- idf0 = IDF(inputCol="tf")
- self.assertListEqual(idf0.params, [idf0.inputCol, idf0.minDocFreq, idf0.outputCol])
- idf0m = idf0.fit(dataset, {idf0.outputCol: "idf"})
- self.assertEqual(idf0m.uid, idf0.uid,
- "Model should inherit the UID from its parent estimator.")
- output = idf0m.transform(dataset)
- self.assertIsNotNone(output.head().idf)
- # Test that parameters transferred to Python Model
- ParamTests.check_params(self, idf0m)
-
- def test_ngram(self):
- dataset = self.spark.createDataFrame([
- Row(input=["a", "b", "c", "d", "e"])])
- ngram0 = NGram(n=4, inputCol="input", outputCol="output")
- self.assertEqual(ngram0.getN(), 4)
- self.assertEqual(ngram0.getInputCol(), "input")
- self.assertEqual(ngram0.getOutputCol(), "output")
- transformedDF = ngram0.transform(dataset)
- self.assertEqual(transformedDF.head().output, ["a b c d", "b c d e"])
-
- def test_stopwordsremover(self):
- dataset = self.spark.createDataFrame([Row(input=["a", "panda"])])
- stopWordRemover = StopWordsRemover(inputCol="input", outputCol="output")
- # Default
- self.assertEqual(stopWordRemover.getInputCol(), "input")
- transformedDF = stopWordRemover.transform(dataset)
- self.assertEqual(transformedDF.head().output, ["panda"])
- self.assertEqual(type(stopWordRemover.getStopWords()), list)
- self.assertTrue(isinstance(stopWordRemover.getStopWords()[0], basestring))
- # Custom
- stopwords = ["panda"]
- stopWordRemover.setStopWords(stopwords)
- self.assertEqual(stopWordRemover.getInputCol(), "input")
- self.assertEqual(stopWordRemover.getStopWords(), stopwords)
- transformedDF = stopWordRemover.transform(dataset)
- self.assertEqual(transformedDF.head().output, ["a"])
- # with language selection
- stopwords = StopWordsRemover.loadDefaultStopWords("turkish")
- dataset = self.spark.createDataFrame([Row(input=["acaba", "ama", "biri"])])
- stopWordRemover.setStopWords(stopwords)
- self.assertEqual(stopWordRemover.getStopWords(), stopwords)
- transformedDF = stopWordRemover.transform(dataset)
- self.assertEqual(transformedDF.head().output, [])
- # with locale
- stopwords = ["BELKÄ°"]
- dataset = self.spark.createDataFrame([Row(input=["belki"])])
- stopWordRemover.setStopWords(stopwords).setLocale("tr")
- self.assertEqual(stopWordRemover.getStopWords(), stopwords)
- transformedDF = stopWordRemover.transform(dataset)
- self.assertEqual(transformedDF.head().output, [])
-
- def test_count_vectorizer_with_binary(self):
- dataset = self.spark.createDataFrame([
- (0, "a a a b b c".split(' '), SparseVector(3, {0: 1.0, 1: 1.0, 2: 1.0}),),
- (1, "a a".split(' '), SparseVector(3, {0: 1.0}),),
- (2, "a b".split(' '), SparseVector(3, {0: 1.0, 1: 1.0}),),
- (3, "c".split(' '), SparseVector(3, {2: 1.0}),)], ["id", "words", "expected"])
- cv = CountVectorizer(binary=True, inputCol="words", outputCol="features")
- model = cv.fit(dataset)
-
- transformedList = model.transform(dataset).select("features", "expected").collect()
-
- for r in transformedList:
- feature, expected = r
- self.assertEqual(feature, expected)
-
- def test_count_vectorizer_with_maxDF(self):
- dataset = self.spark.createDataFrame([
- (0, "a b c d".split(' '), SparseVector(3, {0: 1.0, 1: 1.0, 2: 1.0}),),
- (1, "a b c".split(' '), SparseVector(3, {0: 1.0, 1: 1.0}),),
- (2, "a b".split(' '), SparseVector(3, {0: 1.0}),),
- (3, "a".split(' '), SparseVector(3, {}),)], ["id", "words", "expected"])
- cv = CountVectorizer(inputCol="words", outputCol="features")
- model1 = cv.setMaxDF(3).fit(dataset)
- self.assertEqual(model1.vocabulary, ['b', 'c', 'd'])
-
- transformedList1 = model1.transform(dataset).select("features", "expected").collect()
-
- for r in transformedList1:
- feature, expected = r
- self.assertEqual(feature, expected)
-
- model2 = cv.setMaxDF(0.75).fit(dataset)
- self.assertEqual(model2.vocabulary, ['b', 'c', 'd'])
-
- transformedList2 = model2.transform(dataset).select("features", "expected").collect()
-
- for r in transformedList2:
- feature, expected = r
- self.assertEqual(feature, expected)
-
- def test_count_vectorizer_from_vocab(self):
- model = CountVectorizerModel.from_vocabulary(["a", "b", "c"], inputCol="words",
- outputCol="features", minTF=2)
- self.assertEqual(model.vocabulary, ["a", "b", "c"])
- self.assertEqual(model.getMinTF(), 2)
-
- dataset = self.spark.createDataFrame([
- (0, "a a a b b c".split(' '), SparseVector(3, {0: 3.0, 1: 2.0}),),
- (1, "a a".split(' '), SparseVector(3, {0: 2.0}),),
- (2, "a b".split(' '), SparseVector(3, {}),)], ["id", "words", "expected"])
-
- transformed_list = model.transform(dataset).select("features", "expected").collect()
-
- for r in transformed_list:
- feature, expected = r
- self.assertEqual(feature, expected)
-
- # Test an empty vocabulary
- with QuietTest(self.sc):
- with self.assertRaisesRegexp(Exception, "vocabSize.*invalid.*0"):
- CountVectorizerModel.from_vocabulary([], inputCol="words")
-
- # Test model with default settings can transform
- model_default = CountVectorizerModel.from_vocabulary(["a", "b", "c"], inputCol="words")
- transformed_list = model_default.transform(dataset)\
- .select(model_default.getOrDefault(model_default.outputCol)).collect()
- self.assertEqual(len(transformed_list), 3)
-
- def test_rformula_force_index_label(self):
- df = self.spark.createDataFrame([
- (1.0, 1.0, "a"),
- (0.0, 2.0, "b"),
- (1.0, 0.0, "a")], ["y", "x", "s"])
- # Does not index label by default since it's numeric type.
- rf = RFormula(formula="y ~ x + s")
- model = rf.fit(df)
- transformedDF = model.transform(df)
- self.assertEqual(transformedDF.head().label, 1.0)
- # Force to index label.
- rf2 = RFormula(formula="y ~ x + s").setForceIndexLabel(True)
- model2 = rf2.fit(df)
- transformedDF2 = model2.transform(df)
- self.assertEqual(transformedDF2.head().label, 0.0)
-
- def test_rformula_string_indexer_order_type(self):
- df = self.spark.createDataFrame([
- (1.0, 1.0, "a"),
- (0.0, 2.0, "b"),
- (1.0, 0.0, "a")], ["y", "x", "s"])
- rf = RFormula(formula="y ~ x + s", stringIndexerOrderType="alphabetDesc")
- self.assertEqual(rf.getStringIndexerOrderType(), 'alphabetDesc')
- transformedDF = rf.fit(df).transform(df)
- observed = transformedDF.select("features").collect()
- expected = [[1.0, 0.0], [2.0, 1.0], [0.0, 0.0]]
- for i in range(0, len(expected)):
- self.assertTrue(all(observed[i]["features"].toArray() == expected[i]))
-
- def test_string_indexer_handle_invalid(self):
- df = self.spark.createDataFrame([
- (0, "a"),
- (1, "d"),
- (2, None)], ["id", "label"])
-
- si1 = StringIndexer(inputCol="label", outputCol="indexed", handleInvalid="keep",
- stringOrderType="alphabetAsc")
- model1 = si1.fit(df)
- td1 = model1.transform(df)
- actual1 = td1.select("id", "indexed").collect()
- expected1 = [Row(id=0, indexed=0.0), Row(id=1, indexed=1.0), Row(id=2, indexed=2.0)]
- self.assertEqual(actual1, expected1)
-
- si2 = si1.setHandleInvalid("skip")
- model2 = si2.fit(df)
- td2 = model2.transform(df)
- actual2 = td2.select("id", "indexed").collect()
- expected2 = [Row(id=0, indexed=0.0), Row(id=1, indexed=1.0)]
- self.assertEqual(actual2, expected2)
-
- def test_string_indexer_from_labels(self):
- model = StringIndexerModel.from_labels(["a", "b", "c"], inputCol="label",
- outputCol="indexed", handleInvalid="keep")
- self.assertEqual(model.labels, ["a", "b", "c"])
-
- df1 = self.spark.createDataFrame([
- (0, "a"),
- (1, "c"),
- (2, None),
- (3, "b"),
- (4, "b")], ["id", "label"])
-
- result1 = model.transform(df1)
- actual1 = result1.select("id", "indexed").collect()
- expected1 = [Row(id=0, indexed=0.0), Row(id=1, indexed=2.0), Row(id=2, indexed=3.0),
- Row(id=3, indexed=1.0), Row(id=4, indexed=1.0)]
- self.assertEqual(actual1, expected1)
-
- model_empty_labels = StringIndexerModel.from_labels(
- [], inputCol="label", outputCol="indexed", handleInvalid="keep")
- actual2 = model_empty_labels.transform(df1).select("id", "indexed").collect()
- expected2 = [Row(id=0, indexed=0.0), Row(id=1, indexed=0.0), Row(id=2, indexed=0.0),
- Row(id=3, indexed=0.0), Row(id=4, indexed=0.0)]
- self.assertEqual(actual2, expected2)
-
- # Test model with default settings can transform
- model_default = StringIndexerModel.from_labels(["a", "b", "c"], inputCol="label")
- df2 = self.spark.createDataFrame([
- (0, "a"),
- (1, "c"),
- (2, "b"),
- (3, "b"),
- (4, "b")], ["id", "label"])
- transformed_list = model_default.transform(df2)\
- .select(model_default.getOrDefault(model_default.outputCol)).collect()
- self.assertEqual(len(transformed_list), 5)
-
- def test_vector_size_hint(self):
- df = self.spark.createDataFrame(
- [(0, Vectors.dense([0.0, 10.0, 0.5])),
- (1, Vectors.dense([1.0, 11.0, 0.5, 0.6])),
- (2, Vectors.dense([2.0, 12.0]))],
- ["id", "vector"])
-
- sizeHint = VectorSizeHint(
- inputCol="vector",
- handleInvalid="skip")
- sizeHint.setSize(3)
- self.assertEqual(sizeHint.getSize(), 3)
-
- output = sizeHint.transform(df).head().vector
- expected = DenseVector([0.0, 10.0, 0.5])
- self.assertEqual(output, expected)
-
-
-class HasInducedError(Params):
-
- def __init__(self):
- super(HasInducedError, self).__init__()
- self.inducedError = Param(self, "inducedError",
- "Uniformly-distributed error added to feature")
-
- def getInducedError(self):
- return self.getOrDefault(self.inducedError)
-
-
-class InducedErrorModel(Model, HasInducedError):
-
- def __init__(self):
- super(InducedErrorModel, self).__init__()
-
- def _transform(self, dataset):
- return dataset.withColumn("prediction",
- dataset.feature + (rand(0) * self.getInducedError()))
-
-
-class InducedErrorEstimator(Estimator, HasInducedError):
-
- def __init__(self, inducedError=1.0):
- super(InducedErrorEstimator, self).__init__()
- self._set(inducedError=inducedError)
-
- def _fit(self, dataset):
- model = InducedErrorModel()
- self._copyValues(model)
- return model
-
-
-class CrossValidatorTests(SparkSessionTestCase):
-
- def test_copy(self):
- dataset = self.spark.createDataFrame([
- (10, 10.0),
- (50, 50.0),
- (100, 100.0),
- (500, 500.0)] * 10,
- ["feature", "label"])
-
- iee = InducedErrorEstimator()
- evaluator = RegressionEvaluator(metricName="rmse")
-
- grid = (ParamGridBuilder()
- .addGrid(iee.inducedError, [100.0, 0.0, 10000.0])
- .build())
- cv = CrossValidator(estimator=iee, estimatorParamMaps=grid, evaluator=evaluator)
- cvCopied = cv.copy()
- self.assertEqual(cv.getEstimator().uid, cvCopied.getEstimator().uid)
-
- cvModel = cv.fit(dataset)
- cvModelCopied = cvModel.copy()
- for index in range(len(cvModel.avgMetrics)):
- self.assertTrue(abs(cvModel.avgMetrics[index] - cvModelCopied.avgMetrics[index])
- < 0.0001)
-
- def test_fit_minimize_metric(self):
- dataset = self.spark.createDataFrame([
- (10, 10.0),
- (50, 50.0),
- (100, 100.0),
- (500, 500.0)] * 10,
- ["feature", "label"])
-
- iee = InducedErrorEstimator()
- evaluator = RegressionEvaluator(metricName="rmse")
-
- grid = (ParamGridBuilder()
- .addGrid(iee.inducedError, [100.0, 0.0, 10000.0])
- .build())
- cv = CrossValidator(estimator=iee, estimatorParamMaps=grid, evaluator=evaluator)
- cvModel = cv.fit(dataset)
- bestModel = cvModel.bestModel
- bestModelMetric = evaluator.evaluate(bestModel.transform(dataset))
-
- self.assertEqual(0.0, bestModel.getOrDefault('inducedError'),
- "Best model should have zero induced error")
- self.assertEqual(0.0, bestModelMetric, "Best model has RMSE of 0")
-
- def test_fit_maximize_metric(self):
- dataset = self.spark.createDataFrame([
- (10, 10.0),
- (50, 50.0),
- (100, 100.0),
- (500, 500.0)] * 10,
- ["feature", "label"])
-
- iee = InducedErrorEstimator()
- evaluator = RegressionEvaluator(metricName="r2")
-
- grid = (ParamGridBuilder()
- .addGrid(iee.inducedError, [100.0, 0.0, 10000.0])
- .build())
- cv = CrossValidator(estimator=iee, estimatorParamMaps=grid, evaluator=evaluator)
- cvModel = cv.fit(dataset)
- bestModel = cvModel.bestModel
- bestModelMetric = evaluator.evaluate(bestModel.transform(dataset))
-
- self.assertEqual(0.0, bestModel.getOrDefault('inducedError'),
- "Best model should have zero induced error")
- self.assertEqual(1.0, bestModelMetric, "Best model has R-squared of 1")
-
- def test_param_grid_type_coercion(self):
- lr = LogisticRegression(maxIter=10)
- paramGrid = ParamGridBuilder().addGrid(lr.regParam, [0.5, 1]).build()
- for param in paramGrid:
- for v in param.values():
- assert(type(v) == float)
-
- def test_save_load_trained_model(self):
- # This tests saving and loading the trained model only.
- # Save/load for CrossValidator will be added later: SPARK-13786
- temp_path = tempfile.mkdtemp()
- dataset = self.spark.createDataFrame(
- [(Vectors.dense([0.0]), 0.0),
- (Vectors.dense([0.4]), 1.0),
- (Vectors.dense([0.5]), 0.0),
- (Vectors.dense([0.6]), 1.0),
- (Vectors.dense([1.0]), 1.0)] * 10,
- ["features", "label"])
- lr = LogisticRegression()
- grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
- evaluator = BinaryClassificationEvaluator()
- cv = CrossValidator(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator)
- cvModel = cv.fit(dataset)
- lrModel = cvModel.bestModel
-
- cvModelPath = temp_path + "/cvModel"
- lrModel.save(cvModelPath)
- loadedLrModel = LogisticRegressionModel.load(cvModelPath)
- self.assertEqual(loadedLrModel.uid, lrModel.uid)
- self.assertEqual(loadedLrModel.intercept, lrModel.intercept)
-
- def test_save_load_simple_estimator(self):
- temp_path = tempfile.mkdtemp()
- dataset = self.spark.createDataFrame(
- [(Vectors.dense([0.0]), 0.0),
- (Vectors.dense([0.4]), 1.0),
- (Vectors.dense([0.5]), 0.0),
- (Vectors.dense([0.6]), 1.0),
- (Vectors.dense([1.0]), 1.0)] * 10,
- ["features", "label"])
-
- lr = LogisticRegression()
- grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
- evaluator = BinaryClassificationEvaluator()
-
- # test save/load of CrossValidator
- cv = CrossValidator(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator)
- cvModel = cv.fit(dataset)
- cvPath = temp_path + "/cv"
- cv.save(cvPath)
- loadedCV = CrossValidator.load(cvPath)
- self.assertEqual(loadedCV.getEstimator().uid, cv.getEstimator().uid)
- self.assertEqual(loadedCV.getEvaluator().uid, cv.getEvaluator().uid)
- self.assertEqual(loadedCV.getEstimatorParamMaps(), cv.getEstimatorParamMaps())
-
- # test save/load of CrossValidatorModel
- cvModelPath = temp_path + "/cvModel"
- cvModel.save(cvModelPath)
- loadedModel = CrossValidatorModel.load(cvModelPath)
- self.assertEqual(loadedModel.bestModel.uid, cvModel.bestModel.uid)
-
- def test_parallel_evaluation(self):
- dataset = self.spark.createDataFrame(
- [(Vectors.dense([0.0]), 0.0),
- (Vectors.dense([0.4]), 1.0),
- (Vectors.dense([0.5]), 0.0),
- (Vectors.dense([0.6]), 1.0),
- (Vectors.dense([1.0]), 1.0)] * 10,
- ["features", "label"])
-
- lr = LogisticRegression()
- grid = ParamGridBuilder().addGrid(lr.maxIter, [5, 6]).build()
- evaluator = BinaryClassificationEvaluator()
-
- # test save/load of CrossValidator
- cv = CrossValidator(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator)
- cv.setParallelism(1)
- cvSerialModel = cv.fit(dataset)
- cv.setParallelism(2)
- cvParallelModel = cv.fit(dataset)
- self.assertEqual(cvSerialModel.avgMetrics, cvParallelModel.avgMetrics)
-
- def test_expose_sub_models(self):
- temp_path = tempfile.mkdtemp()
- dataset = self.spark.createDataFrame(
- [(Vectors.dense([0.0]), 0.0),
- (Vectors.dense([0.4]), 1.0),
- (Vectors.dense([0.5]), 0.0),
- (Vectors.dense([0.6]), 1.0),
- (Vectors.dense([1.0]), 1.0)] * 10,
- ["features", "label"])
-
- lr = LogisticRegression()
- grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
- evaluator = BinaryClassificationEvaluator()
-
- numFolds = 3
- cv = CrossValidator(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator,
- numFolds=numFolds, collectSubModels=True)
-
- def checkSubModels(subModels):
- self.assertEqual(len(subModels), numFolds)
- for i in range(numFolds):
- self.assertEqual(len(subModels[i]), len(grid))
-
- cvModel = cv.fit(dataset)
- checkSubModels(cvModel.subModels)
-
- # Test the default value for option "persistSubModel" to be "true"
- testSubPath = temp_path + "/testCrossValidatorSubModels"
- savingPathWithSubModels = testSubPath + "cvModel3"
- cvModel.save(savingPathWithSubModels)
- cvModel3 = CrossValidatorModel.load(savingPathWithSubModels)
- checkSubModels(cvModel3.subModels)
- cvModel4 = cvModel3.copy()
- checkSubModels(cvModel4.subModels)
-
- savingPathWithoutSubModels = testSubPath + "cvModel2"
- cvModel.write().option("persistSubModels", "false").save(savingPathWithoutSubModels)
- cvModel2 = CrossValidatorModel.load(savingPathWithoutSubModels)
- self.assertEqual(cvModel2.subModels, None)
-
- for i in range(numFolds):
- for j in range(len(grid)):
- self.assertEqual(cvModel.subModels[i][j].uid, cvModel3.subModels[i][j].uid)
-
- def test_save_load_nested_estimator(self):
- temp_path = tempfile.mkdtemp()
- dataset = self.spark.createDataFrame(
- [(Vectors.dense([0.0]), 0.0),
- (Vectors.dense([0.4]), 1.0),
- (Vectors.dense([0.5]), 0.0),
- (Vectors.dense([0.6]), 1.0),
- (Vectors.dense([1.0]), 1.0)] * 10,
- ["features", "label"])
-
- ova = OneVsRest(classifier=LogisticRegression())
- lr1 = LogisticRegression().setMaxIter(100)
- lr2 = LogisticRegression().setMaxIter(150)
- grid = ParamGridBuilder().addGrid(ova.classifier, [lr1, lr2]).build()
- evaluator = MulticlassClassificationEvaluator()
-
- # test save/load of CrossValidator
- cv = CrossValidator(estimator=ova, estimatorParamMaps=grid, evaluator=evaluator)
- cvModel = cv.fit(dataset)
- cvPath = temp_path + "/cv"
- cv.save(cvPath)
- loadedCV = CrossValidator.load(cvPath)
- self.assertEqual(loadedCV.getEstimator().uid, cv.getEstimator().uid)
- self.assertEqual(loadedCV.getEvaluator().uid, cv.getEvaluator().uid)
-
- originalParamMap = cv.getEstimatorParamMaps()
- loadedParamMap = loadedCV.getEstimatorParamMaps()
- for i, param in enumerate(loadedParamMap):
- for p in param:
- if p.name == "classifier":
- self.assertEqual(param[p].uid, originalParamMap[i][p].uid)
- else:
- self.assertEqual(param[p], originalParamMap[i][p])
-
- # test save/load of CrossValidatorModel
- cvModelPath = temp_path + "/cvModel"
- cvModel.save(cvModelPath)
- loadedModel = CrossValidatorModel.load(cvModelPath)
- self.assertEqual(loadedModel.bestModel.uid, cvModel.bestModel.uid)
-
-
-class TrainValidationSplitTests(SparkSessionTestCase):
-
- def test_fit_minimize_metric(self):
- dataset = self.spark.createDataFrame([
- (10, 10.0),
- (50, 50.0),
- (100, 100.0),
- (500, 500.0)] * 10,
- ["feature", "label"])
-
- iee = InducedErrorEstimator()
- evaluator = RegressionEvaluator(metricName="rmse")
-
- grid = ParamGridBuilder() \
- .addGrid(iee.inducedError, [100.0, 0.0, 10000.0]) \
- .build()
- tvs = TrainValidationSplit(estimator=iee, estimatorParamMaps=grid, evaluator=evaluator)
- tvsModel = tvs.fit(dataset)
- bestModel = tvsModel.bestModel
- bestModelMetric = evaluator.evaluate(bestModel.transform(dataset))
- validationMetrics = tvsModel.validationMetrics
-
- self.assertEqual(0.0, bestModel.getOrDefault('inducedError'),
- "Best model should have zero induced error")
- self.assertEqual(0.0, bestModelMetric, "Best model has RMSE of 0")
- self.assertEqual(len(grid), len(validationMetrics),
- "validationMetrics has the same size of grid parameter")
- self.assertEqual(0.0, min(validationMetrics))
-
- def test_fit_maximize_metric(self):
- dataset = self.spark.createDataFrame([
- (10, 10.0),
- (50, 50.0),
- (100, 100.0),
- (500, 500.0)] * 10,
- ["feature", "label"])
-
- iee = InducedErrorEstimator()
- evaluator = RegressionEvaluator(metricName="r2")
-
- grid = ParamGridBuilder() \
- .addGrid(iee.inducedError, [100.0, 0.0, 10000.0]) \
- .build()
- tvs = TrainValidationSplit(estimator=iee, estimatorParamMaps=grid, evaluator=evaluator)
- tvsModel = tvs.fit(dataset)
- bestModel = tvsModel.bestModel
- bestModelMetric = evaluator.evaluate(bestModel.transform(dataset))
- validationMetrics = tvsModel.validationMetrics
-
- self.assertEqual(0.0, bestModel.getOrDefault('inducedError'),
- "Best model should have zero induced error")
- self.assertEqual(1.0, bestModelMetric, "Best model has R-squared of 1")
- self.assertEqual(len(grid), len(validationMetrics),
- "validationMetrics has the same size of grid parameter")
- self.assertEqual(1.0, max(validationMetrics))
-
- def test_save_load_trained_model(self):
- # This tests saving and loading the trained model only.
- # Save/load for TrainValidationSplit will be added later: SPARK-13786
- temp_path = tempfile.mkdtemp()
- dataset = self.spark.createDataFrame(
- [(Vectors.dense([0.0]), 0.0),
- (Vectors.dense([0.4]), 1.0),
- (Vectors.dense([0.5]), 0.0),
- (Vectors.dense([0.6]), 1.0),
- (Vectors.dense([1.0]), 1.0)] * 10,
- ["features", "label"])
- lr = LogisticRegression()
- grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
- evaluator = BinaryClassificationEvaluator()
- tvs = TrainValidationSplit(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator)
- tvsModel = tvs.fit(dataset)
- lrModel = tvsModel.bestModel
-
- tvsModelPath = temp_path + "/tvsModel"
- lrModel.save(tvsModelPath)
- loadedLrModel = LogisticRegressionModel.load(tvsModelPath)
- self.assertEqual(loadedLrModel.uid, lrModel.uid)
- self.assertEqual(loadedLrModel.intercept, lrModel.intercept)
-
- def test_save_load_simple_estimator(self):
- # This tests saving and loading the trained model only.
- # Save/load for TrainValidationSplit will be added later: SPARK-13786
- temp_path = tempfile.mkdtemp()
- dataset = self.spark.createDataFrame(
- [(Vectors.dense([0.0]), 0.0),
- (Vectors.dense([0.4]), 1.0),
- (Vectors.dense([0.5]), 0.0),
- (Vectors.dense([0.6]), 1.0),
- (Vectors.dense([1.0]), 1.0)] * 10,
- ["features", "label"])
- lr = LogisticRegression()
- grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
- evaluator = BinaryClassificationEvaluator()
- tvs = TrainValidationSplit(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator)
- tvsModel = tvs.fit(dataset)
-
- tvsPath = temp_path + "/tvs"
- tvs.save(tvsPath)
- loadedTvs = TrainValidationSplit.load(tvsPath)
- self.assertEqual(loadedTvs.getEstimator().uid, tvs.getEstimator().uid)
- self.assertEqual(loadedTvs.getEvaluator().uid, tvs.getEvaluator().uid)
- self.assertEqual(loadedTvs.getEstimatorParamMaps(), tvs.getEstimatorParamMaps())
-
- tvsModelPath = temp_path + "/tvsModel"
- tvsModel.save(tvsModelPath)
- loadedModel = TrainValidationSplitModel.load(tvsModelPath)
- self.assertEqual(loadedModel.bestModel.uid, tvsModel.bestModel.uid)
-
- def test_parallel_evaluation(self):
- dataset = self.spark.createDataFrame(
- [(Vectors.dense([0.0]), 0.0),
- (Vectors.dense([0.4]), 1.0),
- (Vectors.dense([0.5]), 0.0),
- (Vectors.dense([0.6]), 1.0),
- (Vectors.dense([1.0]), 1.0)] * 10,
- ["features", "label"])
- lr = LogisticRegression()
- grid = ParamGridBuilder().addGrid(lr.maxIter, [5, 6]).build()
- evaluator = BinaryClassificationEvaluator()
- tvs = TrainValidationSplit(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator)
- tvs.setParallelism(1)
- tvsSerialModel = tvs.fit(dataset)
- tvs.setParallelism(2)
- tvsParallelModel = tvs.fit(dataset)
- self.assertEqual(tvsSerialModel.validationMetrics, tvsParallelModel.validationMetrics)
-
- def test_expose_sub_models(self):
- temp_path = tempfile.mkdtemp()
- dataset = self.spark.createDataFrame(
- [(Vectors.dense([0.0]), 0.0),
- (Vectors.dense([0.4]), 1.0),
- (Vectors.dense([0.5]), 0.0),
- (Vectors.dense([0.6]), 1.0),
- (Vectors.dense([1.0]), 1.0)] * 10,
- ["features", "label"])
- lr = LogisticRegression()
- grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
- evaluator = BinaryClassificationEvaluator()
- tvs = TrainValidationSplit(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator,
- collectSubModels=True)
- tvsModel = tvs.fit(dataset)
- self.assertEqual(len(tvsModel.subModels), len(grid))
-
- # Test the default value for option "persistSubModel" to be "true"
- testSubPath = temp_path + "/testTrainValidationSplitSubModels"
- savingPathWithSubModels = testSubPath + "cvModel3"
- tvsModel.save(savingPathWithSubModels)
- tvsModel3 = TrainValidationSplitModel.load(savingPathWithSubModels)
- self.assertEqual(len(tvsModel3.subModels), len(grid))
- tvsModel4 = tvsModel3.copy()
- self.assertEqual(len(tvsModel4.subModels), len(grid))
-
- savingPathWithoutSubModels = testSubPath + "cvModel2"
- tvsModel.write().option("persistSubModels", "false").save(savingPathWithoutSubModels)
- tvsModel2 = TrainValidationSplitModel.load(savingPathWithoutSubModels)
- self.assertEqual(tvsModel2.subModels, None)
-
- for i in range(len(grid)):
- self.assertEqual(tvsModel.subModels[i].uid, tvsModel3.subModels[i].uid)
-
- def test_save_load_nested_estimator(self):
- # This tests saving and loading the trained model only.
- # Save/load for TrainValidationSplit will be added later: SPARK-13786
- temp_path = tempfile.mkdtemp()
- dataset = self.spark.createDataFrame(
- [(Vectors.dense([0.0]), 0.0),
- (Vectors.dense([0.4]), 1.0),
- (Vectors.dense([0.5]), 0.0),
- (Vectors.dense([0.6]), 1.0),
- (Vectors.dense([1.0]), 1.0)] * 10,
- ["features", "label"])
- ova = OneVsRest(classifier=LogisticRegression())
- lr1 = LogisticRegression().setMaxIter(100)
- lr2 = LogisticRegression().setMaxIter(150)
- grid = ParamGridBuilder().addGrid(ova.classifier, [lr1, lr2]).build()
- evaluator = MulticlassClassificationEvaluator()
-
- tvs = TrainValidationSplit(estimator=ova, estimatorParamMaps=grid, evaluator=evaluator)
- tvsModel = tvs.fit(dataset)
- tvsPath = temp_path + "/tvs"
- tvs.save(tvsPath)
- loadedTvs = TrainValidationSplit.load(tvsPath)
- self.assertEqual(loadedTvs.getEstimator().uid, tvs.getEstimator().uid)
- self.assertEqual(loadedTvs.getEvaluator().uid, tvs.getEvaluator().uid)
-
- originalParamMap = tvs.getEstimatorParamMaps()
- loadedParamMap = loadedTvs.getEstimatorParamMaps()
- for i, param in enumerate(loadedParamMap):
- for p in param:
- if p.name == "classifier":
- self.assertEqual(param[p].uid, originalParamMap[i][p].uid)
- else:
- self.assertEqual(param[p], originalParamMap[i][p])
-
- tvsModelPath = temp_path + "/tvsModel"
- tvsModel.save(tvsModelPath)
- loadedModel = TrainValidationSplitModel.load(tvsModelPath)
- self.assertEqual(loadedModel.bestModel.uid, tvsModel.bestModel.uid)
-
- def test_copy(self):
- dataset = self.spark.createDataFrame([
- (10, 10.0),
- (50, 50.0),
- (100, 100.0),
- (500, 500.0)] * 10,
- ["feature", "label"])
-
- iee = InducedErrorEstimator()
- evaluator = RegressionEvaluator(metricName="r2")
-
- grid = ParamGridBuilder() \
- .addGrid(iee.inducedError, [100.0, 0.0, 10000.0]) \
- .build()
- tvs = TrainValidationSplit(estimator=iee, estimatorParamMaps=grid, evaluator=evaluator)
- tvsModel = tvs.fit(dataset)
- tvsCopied = tvs.copy()
- tvsModelCopied = tvsModel.copy()
-
- self.assertEqual(tvs.getEstimator().uid, tvsCopied.getEstimator().uid,
- "Copied TrainValidationSplit has the same uid of Estimator")
-
- self.assertEqual(tvsModel.bestModel.uid, tvsModelCopied.bestModel.uid)
- self.assertEqual(len(tvsModel.validationMetrics),
- len(tvsModelCopied.validationMetrics),
- "Copied validationMetrics has the same size of the original")
- for index in range(len(tvsModel.validationMetrics)):
- self.assertEqual(tvsModel.validationMetrics[index],
- tvsModelCopied.validationMetrics[index])
-
-
-class PersistenceTest(SparkSessionTestCase):
-
- def test_linear_regression(self):
- lr = LinearRegression(maxIter=1)
- path = tempfile.mkdtemp()
- lr_path = path + "/lr"
- lr.save(lr_path)
- lr2 = LinearRegression.load(lr_path)
- self.assertEqual(lr.uid, lr2.uid)
- self.assertEqual(type(lr.uid), type(lr2.uid))
- self.assertEqual(lr2.uid, lr2.maxIter.parent,
- "Loaded LinearRegression instance uid (%s) did not match Param's uid (%s)"
- % (lr2.uid, lr2.maxIter.parent))
- self.assertEqual(lr._defaultParamMap[lr.maxIter], lr2._defaultParamMap[lr2.maxIter],
- "Loaded LinearRegression instance default params did not match " +
- "original defaults")
- try:
- rmtree(path)
- except OSError:
- pass
-
- def test_linear_regression_pmml_basic(self):
- # Most of the validation is done in the Scala side, here we just check
- # that we output text rather than parquet (e.g. that the format flag
- # was respected).
- df = self.spark.createDataFrame([(1.0, 2.0, Vectors.dense(1.0)),
- (0.0, 2.0, Vectors.sparse(1, [], []))],
- ["label", "weight", "features"])
- lr = LinearRegression(maxIter=1)
- model = lr.fit(df)
- path = tempfile.mkdtemp()
- lr_path = path + "/lr-pmml"
- model.write().format("pmml").save(lr_path)
- pmml_text_list = self.sc.textFile(lr_path).collect()
- pmml_text = "\n".join(pmml_text_list)
- self.assertIn("Apache Spark", pmml_text)
- self.assertIn("PMML", pmml_text)
-
- def test_logistic_regression(self):
- lr = LogisticRegression(maxIter=1)
- path = tempfile.mkdtemp()
- lr_path = path + "/logreg"
- lr.save(lr_path)
- lr2 = LogisticRegression.load(lr_path)
- self.assertEqual(lr2.uid, lr2.maxIter.parent,
- "Loaded LogisticRegression instance uid (%s) "
- "did not match Param's uid (%s)"
- % (lr2.uid, lr2.maxIter.parent))
- self.assertEqual(lr._defaultParamMap[lr.maxIter], lr2._defaultParamMap[lr2.maxIter],
- "Loaded LogisticRegression instance default params did not match " +
- "original defaults")
- try:
- rmtree(path)
- except OSError:
- pass
-
- def _compare_params(self, m1, m2, param):
- """
- Compare 2 ML Params instances for the given param, and assert both have the same param value
- and parent. The param must be a parameter of m1.
- """
- # Prevent key not found error in case of some param in neither paramMap nor defaultParamMap.
- if m1.isDefined(param):
- paramValue1 = m1.getOrDefault(param)
- paramValue2 = m2.getOrDefault(m2.getParam(param.name))
- if isinstance(paramValue1, Params):
- self._compare_pipelines(paramValue1, paramValue2)
- else:
- self.assertEqual(paramValue1, paramValue2) # for general types param
- # Assert parents are equal
- self.assertEqual(param.parent, m2.getParam(param.name).parent)
- else:
- # If m1 is not defined param, then m2 should not, too. See SPARK-14931.
- self.assertFalse(m2.isDefined(m2.getParam(param.name)))
-
- def _compare_pipelines(self, m1, m2):
- """
- Compare 2 ML types, asserting that they are equivalent.
- This currently supports:
- - basic types
- - Pipeline, PipelineModel
- - OneVsRest, OneVsRestModel
- This checks:
- - uid
- - type
- - Param values and parents
- """
- self.assertEqual(m1.uid, m2.uid)
- self.assertEqual(type(m1), type(m2))
- if isinstance(m1, JavaParams) or isinstance(m1, Transformer):
- self.assertEqual(len(m1.params), len(m2.params))
- for p in m1.params:
- self._compare_params(m1, m2, p)
- elif isinstance(m1, Pipeline):
- self.assertEqual(len(m1.getStages()), len(m2.getStages()))
- for s1, s2 in zip(m1.getStages(), m2.getStages()):
- self._compare_pipelines(s1, s2)
- elif isinstance(m1, PipelineModel):
- self.assertEqual(len(m1.stages), len(m2.stages))
- for s1, s2 in zip(m1.stages, m2.stages):
- self._compare_pipelines(s1, s2)
- elif isinstance(m1, OneVsRest) or isinstance(m1, OneVsRestModel):
- for p in m1.params:
- self._compare_params(m1, m2, p)
- if isinstance(m1, OneVsRestModel):
- self.assertEqual(len(m1.models), len(m2.models))
- for x, y in zip(m1.models, m2.models):
- self._compare_pipelines(x, y)
- else:
- raise RuntimeError("_compare_pipelines does not yet support type: %s" % type(m1))
-
- def test_pipeline_persistence(self):
- """
- Pipeline[HashingTF, PCA]
- """
- temp_path = tempfile.mkdtemp()
-
- try:
- df = self.spark.createDataFrame([(["a", "b", "c"],), (["c", "d", "e"],)], ["words"])
- tf = HashingTF(numFeatures=10, inputCol="words", outputCol="features")
- pca = PCA(k=2, inputCol="features", outputCol="pca_features")
- pl = Pipeline(stages=[tf, pca])
- model = pl.fit(df)
-
- pipeline_path = temp_path + "/pipeline"
- pl.save(pipeline_path)
- loaded_pipeline = Pipeline.load(pipeline_path)
- self._compare_pipelines(pl, loaded_pipeline)
-
- model_path = temp_path + "/pipeline-model"
- model.save(model_path)
- loaded_model = PipelineModel.load(model_path)
- self._compare_pipelines(model, loaded_model)
- finally:
- try:
- rmtree(temp_path)
- except OSError:
- pass
-
- def test_nested_pipeline_persistence(self):
- """
- Pipeline[HashingTF, Pipeline[PCA]]
- """
- temp_path = tempfile.mkdtemp()
-
- try:
- df = self.spark.createDataFrame([(["a", "b", "c"],), (["c", "d", "e"],)], ["words"])
- tf = HashingTF(numFeatures=10, inputCol="words", outputCol="features")
- pca = PCA(k=2, inputCol="features", outputCol="pca_features")
- p0 = Pipeline(stages=[pca])
- pl = Pipeline(stages=[tf, p0])
- model = pl.fit(df)
-
- pipeline_path = temp_path + "/pipeline"
- pl.save(pipeline_path)
- loaded_pipeline = Pipeline.load(pipeline_path)
- self._compare_pipelines(pl, loaded_pipeline)
-
- model_path = temp_path + "/pipeline-model"
- model.save(model_path)
- loaded_model = PipelineModel.load(model_path)
- self._compare_pipelines(model, loaded_model)
- finally:
- try:
- rmtree(temp_path)
- except OSError:
- pass
-
- def test_python_transformer_pipeline_persistence(self):
- """
- Pipeline[MockUnaryTransformer, Binarizer]
- """
- temp_path = tempfile.mkdtemp()
-
- try:
- df = self.spark.range(0, 10).toDF('input')
- tf = MockUnaryTransformer(shiftVal=2)\
- .setInputCol("input").setOutputCol("shiftedInput")
- tf2 = Binarizer(threshold=6, inputCol="shiftedInput", outputCol="binarized")
- pl = Pipeline(stages=[tf, tf2])
- model = pl.fit(df)
-
- pipeline_path = temp_path + "/pipeline"
- pl.save(pipeline_path)
- loaded_pipeline = Pipeline.load(pipeline_path)
- self._compare_pipelines(pl, loaded_pipeline)
-
- model_path = temp_path + "/pipeline-model"
- model.save(model_path)
- loaded_model = PipelineModel.load(model_path)
- self._compare_pipelines(model, loaded_model)
- finally:
- try:
- rmtree(temp_path)
- except OSError:
- pass
-
- def test_onevsrest(self):
- temp_path = tempfile.mkdtemp()
- df = self.spark.createDataFrame([(0.0, Vectors.dense(1.0, 0.8)),
- (1.0, Vectors.sparse(2, [], [])),
- (2.0, Vectors.dense(0.5, 0.5))] * 10,
- ["label", "features"])
- lr = LogisticRegression(maxIter=5, regParam=0.01)
- ovr = OneVsRest(classifier=lr)
- model = ovr.fit(df)
- ovrPath = temp_path + "/ovr"
- ovr.save(ovrPath)
- loadedOvr = OneVsRest.load(ovrPath)
- self._compare_pipelines(ovr, loadedOvr)
- modelPath = temp_path + "/ovrModel"
- model.save(modelPath)
- loadedModel = OneVsRestModel.load(modelPath)
- self._compare_pipelines(model, loadedModel)
-
- def test_decisiontree_classifier(self):
- dt = DecisionTreeClassifier(maxDepth=1)
- path = tempfile.mkdtemp()
- dtc_path = path + "/dtc"
- dt.save(dtc_path)
- dt2 = DecisionTreeClassifier.load(dtc_path)
- self.assertEqual(dt2.uid, dt2.maxDepth.parent,
- "Loaded DecisionTreeClassifier instance uid (%s) "
- "did not match Param's uid (%s)"
- % (dt2.uid, dt2.maxDepth.parent))
- self.assertEqual(dt._defaultParamMap[dt.maxDepth], dt2._defaultParamMap[dt2.maxDepth],
- "Loaded DecisionTreeClassifier instance default params did not match " +
- "original defaults")
- try:
- rmtree(path)
- except OSError:
- pass
-
- def test_decisiontree_regressor(self):
- dt = DecisionTreeRegressor(maxDepth=1)
- path = tempfile.mkdtemp()
- dtr_path = path + "/dtr"
- dt.save(dtr_path)
- dt2 = DecisionTreeClassifier.load(dtr_path)
- self.assertEqual(dt2.uid, dt2.maxDepth.parent,
- "Loaded DecisionTreeRegressor instance uid (%s) "
- "did not match Param's uid (%s)"
- % (dt2.uid, dt2.maxDepth.parent))
- self.assertEqual(dt._defaultParamMap[dt.maxDepth], dt2._defaultParamMap[dt2.maxDepth],
- "Loaded DecisionTreeRegressor instance default params did not match " +
- "original defaults")
- try:
- rmtree(path)
- except OSError:
- pass
-
- def test_default_read_write(self):
- temp_path = tempfile.mkdtemp()
-
- lr = LogisticRegression()
- lr.setMaxIter(50)
- lr.setThreshold(.75)
- writer = DefaultParamsWriter(lr)
-
- savePath = temp_path + "/lr"
- writer.save(savePath)
-
- reader = DefaultParamsReadable.read()
- lr2 = reader.load(savePath)
-
- self.assertEqual(lr.uid, lr2.uid)
- self.assertEqual(lr.extractParamMap(), lr2.extractParamMap())
-
- # test overwrite
- lr.setThreshold(.8)
- writer.overwrite().save(savePath)
-
- reader = DefaultParamsReadable.read()
- lr3 = reader.load(savePath)
-
- self.assertEqual(lr.uid, lr3.uid)
- self.assertEqual(lr.extractParamMap(), lr3.extractParamMap())
-
- def test_default_read_write_default_params(self):
- lr = LogisticRegression()
- self.assertFalse(lr.isSet(lr.getParam("threshold")))
-
- lr.setMaxIter(50)
- lr.setThreshold(.75)
-
- # `threshold` is set by user, default param `predictionCol` is not set by user.
- self.assertTrue(lr.isSet(lr.getParam("threshold")))
- self.assertFalse(lr.isSet(lr.getParam("predictionCol")))
- self.assertTrue(lr.hasDefault(lr.getParam("predictionCol")))
-
- writer = DefaultParamsWriter(lr)
- metadata = json.loads(writer._get_metadata_to_save(lr, self.sc))
- self.assertTrue("defaultParamMap" in metadata)
-
- reader = DefaultParamsReadable.read()
- metadataStr = json.dumps(metadata, separators=[',', ':'])
- loadedMetadata = reader._parseMetaData(metadataStr, )
- reader.getAndSetParams(lr, loadedMetadata)
-
- self.assertTrue(lr.isSet(lr.getParam("threshold")))
- self.assertFalse(lr.isSet(lr.getParam("predictionCol")))
- self.assertTrue(lr.hasDefault(lr.getParam("predictionCol")))
-
- # manually create metadata without `defaultParamMap` section.
- del metadata['defaultParamMap']
- metadataStr = json.dumps(metadata, separators=[',', ':'])
- loadedMetadata = reader._parseMetaData(metadataStr, )
- with self.assertRaisesRegexp(AssertionError, "`defaultParamMap` section not found"):
- reader.getAndSetParams(lr, loadedMetadata)
-
- # Prior to 2.4.0, metadata doesn't have `defaultParamMap`.
- metadata['sparkVersion'] = '2.3.0'
- metadataStr = json.dumps(metadata, separators=[',', ':'])
- loadedMetadata = reader._parseMetaData(metadataStr, )
- reader.getAndSetParams(lr, loadedMetadata)
-
-
-class LDATest(SparkSessionTestCase):
-
- def _compare(self, m1, m2):
- """
- Temp method for comparing instances.
- TODO: Replace with generic implementation once SPARK-14706 is merged.
- """
- self.assertEqual(m1.uid, m2.uid)
- self.assertEqual(type(m1), type(m2))
- self.assertEqual(len(m1.params), len(m2.params))
- for p in m1.params:
- if m1.isDefined(p):
- self.assertEqual(m1.getOrDefault(p), m2.getOrDefault(p))
- self.assertEqual(p.parent, m2.getParam(p.name).parent)
- if isinstance(m1, LDAModel):
- self.assertEqual(m1.vocabSize(), m2.vocabSize())
- self.assertEqual(m1.topicsMatrix(), m2.topicsMatrix())
-
- def test_persistence(self):
- # Test save/load for LDA, LocalLDAModel, DistributedLDAModel.
- df = self.spark.createDataFrame([
- [1, Vectors.dense([0.0, 1.0])],
- [2, Vectors.sparse(2, {0: 1.0})],
- ], ["id", "features"])
- # Fit model
- lda = LDA(k=2, seed=1, optimizer="em")
- distributedModel = lda.fit(df)
- self.assertTrue(distributedModel.isDistributed())
- localModel = distributedModel.toLocal()
- self.assertFalse(localModel.isDistributed())
- # Define paths
- path = tempfile.mkdtemp()
- lda_path = path + "/lda"
- dist_model_path = path + "/distLDAModel"
- local_model_path = path + "/localLDAModel"
- # Test LDA
- lda.save(lda_path)
- lda2 = LDA.load(lda_path)
- self._compare(lda, lda2)
- # Test DistributedLDAModel
- distributedModel.save(dist_model_path)
- distributedModel2 = DistributedLDAModel.load(dist_model_path)
- self._compare(distributedModel, distributedModel2)
- # Test LocalLDAModel
- localModel.save(local_model_path)
- localModel2 = LocalLDAModel.load(local_model_path)
- self._compare(localModel, localModel2)
- # Clean up
- try:
- rmtree(path)
- except OSError:
- pass
-
-
-class TrainingSummaryTest(SparkSessionTestCase):
-
- def test_linear_regression_summary(self):
- df = self.spark.createDataFrame([(1.0, 2.0, Vectors.dense(1.0)),
- (0.0, 2.0, Vectors.sparse(1, [], []))],
- ["label", "weight", "features"])
- lr = LinearRegression(maxIter=5, regParam=0.0, solver="normal", weightCol="weight",
- fitIntercept=False)
- model = lr.fit(df)
- self.assertTrue(model.hasSummary)
- s = model.summary
- # test that api is callable and returns expected types
- self.assertGreater(s.totalIterations, 0)
- self.assertTrue(isinstance(s.predictions, DataFrame))
- self.assertEqual(s.predictionCol, "prediction")
- self.assertEqual(s.labelCol, "label")
- self.assertEqual(s.featuresCol, "features")
- objHist = s.objectiveHistory
- self.assertTrue(isinstance(objHist, list) and isinstance(objHist[0], float))
- self.assertAlmostEqual(s.explainedVariance, 0.25, 2)
- self.assertAlmostEqual(s.meanAbsoluteError, 0.0)
- self.assertAlmostEqual(s.meanSquaredError, 0.0)
- self.assertAlmostEqual(s.rootMeanSquaredError, 0.0)
- self.assertAlmostEqual(s.r2, 1.0, 2)
- self.assertAlmostEqual(s.r2adj, 1.0, 2)
- self.assertTrue(isinstance(s.residuals, DataFrame))
- self.assertEqual(s.numInstances, 2)
- self.assertEqual(s.degreesOfFreedom, 1)
- devResiduals = s.devianceResiduals
- self.assertTrue(isinstance(devResiduals, list) and isinstance(devResiduals[0], float))
- coefStdErr = s.coefficientStandardErrors
- self.assertTrue(isinstance(coefStdErr, list) and isinstance(coefStdErr[0], float))
- tValues = s.tValues
- self.assertTrue(isinstance(tValues, list) and isinstance(tValues[0], float))
- pValues = s.pValues
- self.assertTrue(isinstance(pValues, list) and isinstance(pValues[0], float))
- # test evaluation (with training dataset) produces a summary with same values
- # one check is enough to verify a summary is returned
- # The child class LinearRegressionTrainingSummary runs full test
- sameSummary = model.evaluate(df)
- self.assertAlmostEqual(sameSummary.explainedVariance, s.explainedVariance)
-
- def test_glr_summary(self):
- from pyspark.ml.linalg import Vectors
- df = self.spark.createDataFrame([(1.0, 2.0, Vectors.dense(1.0)),
- (0.0, 2.0, Vectors.sparse(1, [], []))],
- ["label", "weight", "features"])
- glr = GeneralizedLinearRegression(family="gaussian", link="identity", weightCol="weight",
- fitIntercept=False)
- model = glr.fit(df)
- self.assertTrue(model.hasSummary)
- s = model.summary
- # test that api is callable and returns expected types
- self.assertEqual(s.numIterations, 1) # this should default to a single iteration of WLS
- self.assertTrue(isinstance(s.predictions, DataFrame))
- self.assertEqual(s.predictionCol, "prediction")
- self.assertEqual(s.numInstances, 2)
- self.assertTrue(isinstance(s.residuals(), DataFrame))
- self.assertTrue(isinstance(s.residuals("pearson"), DataFrame))
- coefStdErr = s.coefficientStandardErrors
- self.assertTrue(isinstance(coefStdErr, list) and isinstance(coefStdErr[0], float))
- tValues = s.tValues
- self.assertTrue(isinstance(tValues, list) and isinstance(tValues[0], float))
- pValues = s.pValues
- self.assertTrue(isinstance(pValues, list) and isinstance(pValues[0], float))
- self.assertEqual(s.degreesOfFreedom, 1)
- self.assertEqual(s.residualDegreeOfFreedom, 1)
- self.assertEqual(s.residualDegreeOfFreedomNull, 2)
- self.assertEqual(s.rank, 1)
- self.assertTrue(isinstance(s.solver, basestring))
- self.assertTrue(isinstance(s.aic, float))
- self.assertTrue(isinstance(s.deviance, float))
- self.assertTrue(isinstance(s.nullDeviance, float))
- self.assertTrue(isinstance(s.dispersion, float))
- # test evaluation (with training dataset) produces a summary with same values
- # one check is enough to verify a summary is returned
- # The child class GeneralizedLinearRegressionTrainingSummary runs full test
- sameSummary = model.evaluate(df)
- self.assertAlmostEqual(sameSummary.deviance, s.deviance)
-
- def test_binary_logistic_regression_summary(self):
- df = self.spark.createDataFrame([(1.0, 2.0, Vectors.dense(1.0)),
- (0.0, 2.0, Vectors.sparse(1, [], []))],
- ["label", "weight", "features"])
- lr = LogisticRegression(maxIter=5, regParam=0.01, weightCol="weight", fitIntercept=False)
- model = lr.fit(df)
- self.assertTrue(model.hasSummary)
- s = model.summary
- # test that api is callable and returns expected types
- self.assertTrue(isinstance(s.predictions, DataFrame))
- self.assertEqual(s.probabilityCol, "probability")
- self.assertEqual(s.labelCol, "label")
- self.assertEqual(s.featuresCol, "features")
- self.assertEqual(s.predictionCol, "prediction")
- objHist = s.objectiveHistory
- self.assertTrue(isinstance(objHist, list) and isinstance(objHist[0], float))
- self.assertGreater(s.totalIterations, 0)
- self.assertTrue(isinstance(s.labels, list))
- self.assertTrue(isinstance(s.truePositiveRateByLabel, list))
- self.assertTrue(isinstance(s.falsePositiveRateByLabel, list))
- self.assertTrue(isinstance(s.precisionByLabel, list))
- self.assertTrue(isinstance(s.recallByLabel, list))
- self.assertTrue(isinstance(s.fMeasureByLabel(), list))
- self.assertTrue(isinstance(s.fMeasureByLabel(1.0), list))
- self.assertTrue(isinstance(s.roc, DataFrame))
- self.assertAlmostEqual(s.areaUnderROC, 1.0, 2)
- self.assertTrue(isinstance(s.pr, DataFrame))
- self.assertTrue(isinstance(s.fMeasureByThreshold, DataFrame))
- self.assertTrue(isinstance(s.precisionByThreshold, DataFrame))
- self.assertTrue(isinstance(s.recallByThreshold, DataFrame))
- self.assertAlmostEqual(s.accuracy, 1.0, 2)
- self.assertAlmostEqual(s.weightedTruePositiveRate, 1.0, 2)
- self.assertAlmostEqual(s.weightedFalsePositiveRate, 0.0, 2)
- self.assertAlmostEqual(s.weightedRecall, 1.0, 2)
- self.assertAlmostEqual(s.weightedPrecision, 1.0, 2)
- self.assertAlmostEqual(s.weightedFMeasure(), 1.0, 2)
- self.assertAlmostEqual(s.weightedFMeasure(1.0), 1.0, 2)
- # test evaluation (with training dataset) produces a summary with same values
- # one check is enough to verify a summary is returned, Scala version runs full test
- sameSummary = model.evaluate(df)
- self.assertAlmostEqual(sameSummary.areaUnderROC, s.areaUnderROC)
-
- def test_multiclass_logistic_regression_summary(self):
- df = self.spark.createDataFrame([(1.0, 2.0, Vectors.dense(1.0)),
- (0.0, 2.0, Vectors.sparse(1, [], [])),
- (2.0, 2.0, Vectors.dense(2.0)),
- (2.0, 2.0, Vectors.dense(1.9))],
- ["label", "weight", "features"])
- lr = LogisticRegression(maxIter=5, regParam=0.01, weightCol="weight", fitIntercept=False)
- model = lr.fit(df)
- self.assertTrue(model.hasSummary)
- s = model.summary
- # test that api is callable and returns expected types
- self.assertTrue(isinstance(s.predictions, DataFrame))
- self.assertEqual(s.probabilityCol, "probability")
- self.assertEqual(s.labelCol, "label")
- self.assertEqual(s.featuresCol, "features")
- self.assertEqual(s.predictionCol, "prediction")
- objHist = s.objectiveHistory
- self.assertTrue(isinstance(objHist, list) and isinstance(objHist[0], float))
- self.assertGreater(s.totalIterations, 0)
- self.assertTrue(isinstance(s.labels, list))
- self.assertTrue(isinstance(s.truePositiveRateByLabel, list))
- self.assertTrue(isinstance(s.falsePositiveRateByLabel, list))
- self.assertTrue(isinstance(s.precisionByLabel, list))
- self.assertTrue(isinstance(s.recallByLabel, list))
- self.assertTrue(isinstance(s.fMeasureByLabel(), list))
- self.assertTrue(isinstance(s.fMeasureByLabel(1.0), list))
- self.assertAlmostEqual(s.accuracy, 0.75, 2)
- self.assertAlmostEqual(s.weightedTruePositiveRate, 0.75, 2)
- self.assertAlmostEqual(s.weightedFalsePositiveRate, 0.25, 2)
- self.assertAlmostEqual(s.weightedRecall, 0.75, 2)
- self.assertAlmostEqual(s.weightedPrecision, 0.583, 2)
- self.assertAlmostEqual(s.weightedFMeasure(), 0.65, 2)
- self.assertAlmostEqual(s.weightedFMeasure(1.0), 0.65, 2)
- # test evaluation (with training dataset) produces a summary with same values
- # one check is enough to verify a summary is returned, Scala version runs full test
- sameSummary = model.evaluate(df)
- self.assertAlmostEqual(sameSummary.accuracy, s.accuracy)
-
- def test_gaussian_mixture_summary(self):
- data = [(Vectors.dense(1.0),), (Vectors.dense(5.0),), (Vectors.dense(10.0),),
- (Vectors.sparse(1, [], []),)]
- df = self.spark.createDataFrame(data, ["features"])
- gmm = GaussianMixture(k=2)
- model = gmm.fit(df)
- self.assertTrue(model.hasSummary)
- s = model.summary
- self.assertTrue(isinstance(s.predictions, DataFrame))
- self.assertEqual(s.probabilityCol, "probability")
- self.assertTrue(isinstance(s.probability, DataFrame))
- self.assertEqual(s.featuresCol, "features")
- self.assertEqual(s.predictionCol, "prediction")
- self.assertTrue(isinstance(s.cluster, DataFrame))
- self.assertEqual(len(s.clusterSizes), 2)
- self.assertEqual(s.k, 2)
- self.assertEqual(s.numIter, 3)
-
- def test_bisecting_kmeans_summary(self):
- data = [(Vectors.dense(1.0),), (Vectors.dense(5.0),), (Vectors.dense(10.0),),
- (Vectors.sparse(1, [], []),)]
- df = self.spark.createDataFrame(data, ["features"])
- bkm = BisectingKMeans(k=2)
- model = bkm.fit(df)
- self.assertTrue(model.hasSummary)
- s = model.summary
- self.assertTrue(isinstance(s.predictions, DataFrame))
- self.assertEqual(s.featuresCol, "features")
- self.assertEqual(s.predictionCol, "prediction")
- self.assertTrue(isinstance(s.cluster, DataFrame))
- self.assertEqual(len(s.clusterSizes), 2)
- self.assertEqual(s.k, 2)
- self.assertEqual(s.numIter, 20)
-
- def test_kmeans_summary(self):
- data = [(Vectors.dense([0.0, 0.0]),), (Vectors.dense([1.0, 1.0]),),
- (Vectors.dense([9.0, 8.0]),), (Vectors.dense([8.0, 9.0]),)]
- df = self.spark.createDataFrame(data, ["features"])
- kmeans = KMeans(k=2, seed=1)
- model = kmeans.fit(df)
- self.assertTrue(model.hasSummary)
- s = model.summary
- self.assertTrue(isinstance(s.predictions, DataFrame))
- self.assertEqual(s.featuresCol, "features")
- self.assertEqual(s.predictionCol, "prediction")
- self.assertTrue(isinstance(s.cluster, DataFrame))
- self.assertEqual(len(s.clusterSizes), 2)
- self.assertEqual(s.k, 2)
- self.assertEqual(s.numIter, 1)
-
-
-class KMeansTests(SparkSessionTestCase):
-
- def test_kmeans_cosine_distance(self):
- data = [(Vectors.dense([1.0, 1.0]),), (Vectors.dense([10.0, 10.0]),),
- (Vectors.dense([1.0, 0.5]),), (Vectors.dense([10.0, 4.4]),),
- (Vectors.dense([-1.0, 1.0]),), (Vectors.dense([-100.0, 90.0]),)]
- df = self.spark.createDataFrame(data, ["features"])
- kmeans = KMeans(k=3, seed=1, distanceMeasure="cosine")
- model = kmeans.fit(df)
- result = model.transform(df).collect()
- self.assertTrue(result[0].prediction == result[1].prediction)
- self.assertTrue(result[2].prediction == result[3].prediction)
- self.assertTrue(result[4].prediction == result[5].prediction)
-
-
-class OneVsRestTests(SparkSessionTestCase):
-
- def test_copy(self):
- df = self.spark.createDataFrame([(0.0, Vectors.dense(1.0, 0.8)),
- (1.0, Vectors.sparse(2, [], [])),
- (2.0, Vectors.dense(0.5, 0.5))],
- ["label", "features"])
- lr = LogisticRegression(maxIter=5, regParam=0.01)
- ovr = OneVsRest(classifier=lr)
- ovr1 = ovr.copy({lr.maxIter: 10})
- self.assertEqual(ovr.getClassifier().getMaxIter(), 5)
- self.assertEqual(ovr1.getClassifier().getMaxIter(), 10)
- model = ovr.fit(df)
- model1 = model.copy({model.predictionCol: "indexed"})
- self.assertEqual(model1.getPredictionCol(), "indexed")
-
- def test_output_columns(self):
- df = self.spark.createDataFrame([(0.0, Vectors.dense(1.0, 0.8)),
- (1.0, Vectors.sparse(2, [], [])),
- (2.0, Vectors.dense(0.5, 0.5))],
- ["label", "features"])
- lr = LogisticRegression(maxIter=5, regParam=0.01)
- ovr = OneVsRest(classifier=lr, parallelism=1)
- model = ovr.fit(df)
- output = model.transform(df)
- self.assertEqual(output.columns, ["label", "features", "prediction"])
-
- def test_parallelism_doesnt_change_output(self):
- df = self.spark.createDataFrame([(0.0, Vectors.dense(1.0, 0.8)),
- (1.0, Vectors.sparse(2, [], [])),
- (2.0, Vectors.dense(0.5, 0.5))],
- ["label", "features"])
- ovrPar1 = OneVsRest(classifier=LogisticRegression(maxIter=5, regParam=.01), parallelism=1)
- modelPar1 = ovrPar1.fit(df)
- ovrPar2 = OneVsRest(classifier=LogisticRegression(maxIter=5, regParam=.01), parallelism=2)
- modelPar2 = ovrPar2.fit(df)
- for i, model in enumerate(modelPar1.models):
- self.assertTrue(np.allclose(model.coefficients.toArray(),
- modelPar2.models[i].coefficients.toArray(), atol=1E-4))
- self.assertTrue(np.allclose(model.intercept, modelPar2.models[i].intercept, atol=1E-4))
-
- def
<TRUNCATED>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org