aboutsummaryrefslogtreecommitdiff
path: root/python/pyspark/rdd.py
diff options
context:
space:
mode:
Diffstat (limited to 'python/pyspark/rdd.py')
-rw-r--r--python/pyspark/rdd.py21
1 files changed, 14 insertions, 7 deletions
diff --git a/python/pyspark/rdd.py b/python/pyspark/rdd.py
index 5ab27ff402..e1043ad564 100644
--- a/python/pyspark/rdd.py
+++ b/python/pyspark/rdd.py
@@ -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
@@ -42,12 +43,14 @@ from py4j.java_collections import ListConverter, MapConverter
__all__ = ["RDD"]
def _extract_concise_traceback():
+ """
+ This function returns the traceback info for a callsite, returns a dict
+ with function name, file name and line number
+ """
tb = traceback.extract_stack()
+ callsite = namedtuple("Callsite", "function file linenum")
if len(tb) == 0:
- return "I'm lost!"
- # HACK: This function is in a file called 'rdd.py' in the top level of
- # everything PySpark. Just trim off the directory name and assume
- # everything in that tree is PySpark guts.
+ return None
file, line, module, what = tb[len(tb) - 1]
sparkpath = os.path.dirname(file)
first_spark_frame = len(tb) - 1
@@ -58,16 +61,20 @@ def _extract_concise_traceback():
break
if first_spark_frame == 0:
file, line, fun, what = tb[0]
- return "%s at %s:%d" % (fun, file, 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 "%s at %s:%d" % (sfun, ufile, uline)
+ return callsite(function=sfun, file=ufile, linenum=uline)
_spark_stack_depth = 0
class _JavaStackTrace(object):
def __init__(self, sc):
- self._traceback = _extract_concise_traceback()
+ tb = _extract_concise_traceback()
+ if tb is not None:
+ self._traceback = "%s at %s:%s" % (tb.function, tb.file, tb.linenum)
+ else:
+ self._traceback = "Error! Could not extract traceback info"
self._context = sc
def __enter__(self):