Skip to content

Commit

Permalink
[SPARK-1550] [PySpark] Allow SparkContext creation after failed attempts
Browse files Browse the repository at this point in the history
  • Loading branch information
JoshRosen committed Jul 27, 2014
1 parent 1290164 commit ec7fadc
Show file tree
Hide file tree
Showing 2 changed files with 18 additions and 6 deletions.
18 changes: 12 additions & 6 deletions python/pyspark/context.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,16 @@ def __init__(self, master=None, appName=None, sparkHome=None, pyFiles=None,
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
Expand Down Expand Up @@ -249,17 +258,14 @@ def defaultMinPartitions(self):
"""
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:
Expand Down
6 changes: 6 additions & 0 deletions python/pyspark/tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,12 @@ def func():

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!"
Expand Down

0 comments on commit ec7fadc

Please sign in to comment.