diff --git a/python/pyspark/ml/tests.py b/python/pyspark/ml/tests.py
index 4eb17bfdcca90bd9bb92880d54b8eba8c7afe00d..9ea639dc4f96039c16ff564800d8c6672f2eaa69 100644
--- a/python/pyspark/ml/tests.py
+++ b/python/pyspark/ml/tests.py
@@ -394,6 +394,7 @@ class CrossValidatorTests(PySparkTestCase):
if __name__ == "__main__":
+ from pyspark.ml.tests import *
if xmlrunner:
unittest.main(testRunner=xmlrunner.XMLTestRunner(output='target/test-reports'))
else:
diff --git a/python/pyspark/mllib/tests.py b/python/pyspark/mllib/tests.py
index 32ed48e10388e3d09d7ad0b0fd126f6f0bca72a6..ea7d297cba2ae0d48b203f26d7a13465e75fba59 100644
--- a/python/pyspark/mllib/tests.py
+++ b/python/pyspark/mllib/tests.py
@@ -77,21 +77,24 @@ except:
pass
ser = PickleSerializer()
-sc = SparkContext('local[4]', "MLlib tests")
class MLlibTestCase(unittest.TestCase):
def setUp(self):
- self.sc = sc
+ self.sc = SparkContext('local[4]', "MLlib tests")
+
+ def tearDown(self):
+ self.sc.stop()
class MLLibStreamingTestCase(unittest.TestCase):
def setUp(self):
- self.sc = sc
+ self.sc = SparkContext('local[4]', "MLlib tests")
self.ssc = StreamingContext(self.sc, 1.0)
def tearDown(self):
self.ssc.stop(False)
+ self.sc.stop()
@staticmethod
def _eventually(condition, timeout=30.0, catch_assertions=False):
@@ -1166,7 +1169,7 @@ class StreamingKMeansTest(MLLibStreamingTestCase):
clusterWeights=[1.0, 1.0, 1.0, 1.0])
predict_data = [[[1.5, 1.5]], [[-1.5, 1.5]], [[-1.5, -1.5]], [[1.5, -1.5]]]
- predict_data = [sc.parallelize(batch, 1) for batch in predict_data]
+ predict_data = [self.sc.parallelize(batch, 1) for batch in predict_data]
predict_stream = self.ssc.queueStream(predict_data)
predict_val = stkm.predictOn(predict_stream)
@@ -1197,7 +1200,7 @@ class StreamingKMeansTest(MLLibStreamingTestCase):
# classification based in the initial model would have been 0
# proving that the model is updated.
batches = [[[-0.5], [0.6], [0.8]], [[0.2], [-0.1], [0.3]]]
- batches = [sc.parallelize(batch) for batch in batches]
+ batches = [self.sc.parallelize(batch) for batch in batches]
input_stream = self.ssc.queueStream(batches)
predict_results = []
@@ -1230,7 +1233,7 @@ class LinearDataGeneratorTests(MLlibTestCase):
self.assertEqual(len(point.features), 3)
linear_data = LinearDataGenerator.generateLinearRDD(
- sc=sc, nexamples=6, nfeatures=2, eps=0.1,
+ sc=self.sc, nexamples=6, nfeatures=2, eps=0.1,
nParts=2, intercept=0.0).collect()
self.assertEqual(len(linear_data), 6)
for point in linear_data:
@@ -1406,7 +1409,7 @@ class StreamingLinearRegressionWithTests(MLLibStreamingTestCase):
for i in range(10):
batch = LinearDataGenerator.generateLinearInput(
0.0, [10.0, 10.0], xMean, xVariance, 100, 42 + i, 0.1)
- batches.append(sc.parallelize(batch))
+ batches.append(self.sc.parallelize(batch))
input_stream = self.ssc.queueStream(batches)
slr.trainOn(input_stream)
@@ -1430,7 +1433,7 @@ class StreamingLinearRegressionWithTests(MLLibStreamingTestCase):
for i in range(10):
batch = LinearDataGenerator.generateLinearInput(
0.0, [10.0], [0.0], [1.0 / 3.0], 100, 42 + i, 0.1)
- batches.append(sc.parallelize(batch))
+ batches.append(self.sc.parallelize(batch))
model_weights = []
input_stream = self.ssc.queueStream(batches)
@@ -1463,7 +1466,7 @@ class StreamingLinearRegressionWithTests(MLLibStreamingTestCase):
0.0, [10.0, 10.0], [0.0, 0.0], [1.0 / 3.0, 1.0 / 3.0],
100, 42 + i, 0.1)
batches.append(
- sc.parallelize(batch).map(lambda lp: (lp.label, lp.features)))
+ self.sc.parallelize(batch).map(lambda lp: (lp.label, lp.features)))
input_stream = self.ssc.queueStream(batches)
output_stream = slr.predictOnValues(input_stream)
@@ -1494,7 +1497,7 @@ class StreamingLinearRegressionWithTests(MLLibStreamingTestCase):
for i in range(10):
batch = LinearDataGenerator.generateLinearInput(
0.0, [10.0], [0.0], [1.0 / 3.0], 100, 42 + i, 0.1)
- batches.append(sc.parallelize(batch))
+ batches.append(self.sc.parallelize(batch))
predict_batches = [
b.map(lambda lp: (lp.label, lp.features)) for b in batches]
@@ -1580,6 +1583,7 @@ class ALSTests(MLlibTestCase):
if __name__ == "__main__":
+ from pyspark.mllib.tests import *
if not _have_scipy:
print("NOTE: Skipping SciPy tests as it does not seem to be installed")
if xmlrunner:
diff --git a/python/pyspark/sql/tests.py b/python/pyspark/sql/tests.py
index c03cb9338ae685d72883ccce508054d3acf61c7f..ae8620274dd207523ecc4dc0e6bd8cbc74bdf64c 100644
--- a/python/pyspark/sql/tests.py
+++ b/python/pyspark/sql/tests.py
@@ -1259,6 +1259,7 @@ class HiveContextSQLTests(ReusedPySparkTestCase):
if __name__ == "__main__":
+ from pyspark.sql.tests import *
if xmlrunner:
unittest.main(testRunner=xmlrunner.XMLTestRunner(output='target/test-reports'))
else:
diff --git a/python/pyspark/streaming/tests.py b/python/pyspark/streaming/tests.py
index 86b05d9fd2424d2d09b86ae098922903c0e25b36..24b812615cbb455cd63848302c797d0ebfacf33f 100644
--- a/python/pyspark/streaming/tests.py
+++ b/python/pyspark/streaming/tests.py
@@ -1635,6 +1635,7 @@ kinesis_test_environ_var = "ENABLE_KINESIS_TESTS"
are_kinesis_tests_enabled = os.environ.get(kinesis_test_environ_var) == '1'
if __name__ == "__main__":
+ from pyspark.streaming.tests import *
kafka_assembly_jar = search_kafka_assembly_jar()
flume_assembly_jar = search_flume_assembly_jar()
mqtt_assembly_jar = search_mqtt_assembly_jar()
diff --git a/python/pyspark/tests.py b/python/pyspark/tests.py
index 5bd94476597abcd5cb0f615c806470511f06c76a..23720502a82c86065e6a945583084aa187df924c 100644
--- a/python/pyspark/tests.py
+++ b/python/pyspark/tests.py
@@ -2008,6 +2008,7 @@ class NumPyTests(PySparkTestCase):
if __name__ == "__main__":
+ from pyspark.tests import *
if not _have_scipy:
print("NOTE: Skipping SciPy tests as it does not seem to be installed")
if not _have_numpy: