Skip to content

Commit

Permalink
replaced dict with namedtuple
Browse files Browse the repository at this point in the history
  • Loading branch information
jyotiska committed Mar 5, 2014
1 parent a6bf4cd commit c9439be
Show file tree
Hide file tree
Showing 2 changed files with 9 additions and 5 deletions.
6 changes: 4 additions & 2 deletions python/pyspark/context.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import sys
from threading import Lock
from tempfile import NamedTemporaryFile
from collections import namedtuple

from pyspark import accumulators
from pyspark.accumulators import Accumulator
Expand Down Expand Up @@ -87,7 +88,8 @@ def __init__(self, master=None, appName=None, sparkHome=None, pyFiles=None,
if rdd._extract_concise_traceback() is not None:
self._callsite = rdd._extract_concise_traceback()
else:
self._callsite = {"function": None, "file": None, "line": None}
tempNamedTuple = namedtuple("Callsite", "function file linenum")
self._callsite = tempNamedTuple(function=None, file=None, linenum=None)
SparkContext._ensure_initialized(self, gateway=gateway)

self.environment = environment or {}
Expand Down Expand Up @@ -181,7 +183,7 @@ def _ensure_initialized(cls, instance=None, gateway=None):
# Raise error if there is already a running Spark context
raise ValueError("Cannot run multiple SparkContexts at once; existing SparkContext(app=%s, master=%s)" \
" created by %s at %s:%s " \
% (currentAppName, currentMaster, callsite['function'], callsite['file'], callsite['line']))
% (currentAppName, currentMaster, callsite.function, callsite.file, callsite.linenum))
else:
SparkContext._active_spark_context = instance

Expand Down
8 changes: 5 additions & 3 deletions python/pyspark/rdd.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
from base64 import standard_b64encode as b64enc
import copy
from collections import defaultdict
from collections import namedtuple
from itertools import chain, ifilter, imap
import operator
import os
Expand Down Expand Up @@ -47,6 +48,7 @@ def _extract_concise_traceback():
with function name, file name and line number
"""
tb = traceback.extract_stack()
callsite = namedtuple("Callsite", "function file linenum")
if len(tb) == 0:
return None
file, line, module, what = tb[len(tb) - 1]
Expand All @@ -59,18 +61,18 @@ def _extract_concise_traceback():
break
if first_spark_frame == 0:
file, line, fun, what = tb[0]
return {"function": fun, "file": file, "line": line}
return callsite(function=fun, file=file, linenum=line)
sfile, sline, sfun, swhat = tb[first_spark_frame]
ufile, uline, ufun, uwhat = tb[first_spark_frame-1]
return {"function": sfun, "file": ufile, "line": uline}
return callsite(function=sfun, file=ufile, linenum=uline)

_spark_stack_depth = 0

class _JavaStackTrace(object):
def __init__(self, sc):
tb = _extract_concise_traceback()
if tb is not None:
self._traceback = "%s at %s:%s" % (tb["function"], tb["file"], tb["line"])
self._traceback = "%s at %s:%s" % (tb.function, tb.file, tb.linenum)
else:
self._traceback = "Error! Could not extract traceback info"
self._context = sc
Expand Down

0 comments on commit c9439be

Please sign in to comment.