diff --git a/python/pyspark/context.py b/python/pyspark/context.py index e8ac9895cf54aa730fa5e51a3d4f696fa6c4963e..830a6ee03f2a6ec8a4866fb5c16af575c6886297 100644 --- a/python/pyspark/context.py +++ b/python/pyspark/context.py @@ -100,7 +100,16 @@ class SparkContext(object): tempNamedTuple = namedtuple("Callsite", "function file linenum") self._callsite = tempNamedTuple(function=None, file=None, linenum=None) SparkContext._ensure_initialized(self, gateway=gateway) - + try: + self._do_init(master, appName, sparkHome, pyFiles, environment, batchSize, serializer, + conf) + except: + # If an error occurs, clean up in order to allow future SparkContext creation: + self.stop() + raise + + def _do_init(self, master, appName, sparkHome, pyFiles, environment, batchSize, serializer, + conf): self.environment = environment or {} self._conf = conf or SparkConf(_jvm=self._jvm) self._batchSize = batchSize # -1 represents an unlimited batch size @@ -249,17 +258,14 @@ class SparkContext(object): """ return self._jsc.sc().defaultMinPartitions() - def __del__(self): - self.stop() - def stop(self): """ Shut down the SparkContext. """ - if self._jsc: + if getattr(self, "_jsc", None): self._jsc.stop() self._jsc = None - if self._accumulatorServer: + if getattr(self, "_accumulatorServer", None): self._accumulatorServer.shutdown() self._accumulatorServer = None with SparkContext._lock: diff --git a/python/pyspark/tests.py b/python/pyspark/tests.py index 8ba51461d106daa4f6fd0cb7bdf1bc62acfbfc83..63cc5e9ad96fa867a87c89b03479de410adf4328 100644 --- a/python/pyspark/tests.py +++ b/python/pyspark/tests.py @@ -209,6 +209,12 @@ class TestAddFile(PySparkTestCase): class TestRDDFunctions(PySparkTestCase): + def test_failed_sparkcontext_creation(self): + # Regression test for SPARK-1550 + self.sc.stop() + self.assertRaises(Exception, lambda: SparkContext("an-invalid-master-name")) + self.sc = SparkContext("local") + def test_save_as_textfile_with_unicode(self): # Regression test for SPARK-970 x = u"\u00A1Hola, mundo!"