From 1bc22d76f0d2c125e8a085301b606b2d90675233 Mon Sep 17 00:00:00 2001 From: witgo Date: Wed, 19 Mar 2014 22:37:53 +0800 Subject: [PATCH 1/3] Fix Stage.name return "apply at Option.scala:120" --- core/src/main/scala/org/apache/spark/util/Utils.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala index 13d9dbdd9af2d..2c8e3481cb03b 100644 --- a/core/src/main/scala/org/apache/spark/util/Utils.scala +++ b/core/src/main/scala/org/apache/spark/util/Utils.scala @@ -697,7 +697,7 @@ private[spark] object Utils extends Logging { var finished = false var firstUserClass = "" - for (el <- trace) { + for (el <- trace if el.getClassName != "scala.Option") { if (!finished) { if (SPARK_CLASS_REGEX.findFirstIn(el.getClassName).isDefined) { lastSparkMethod = if (el.getMethodName == "") { From d2b49803382979bf2880ee5a4e31d5e1fc236604 Mon Sep 17 00:00:00 2001 From: witgo Date: Wed, 19 Mar 2014 23:12:28 +0800 Subject: [PATCH 2/3] Modify the position of the filter --- core/src/main/scala/org/apache/spark/util/Utils.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala index 2c8e3481cb03b..70c746441888d 100644 --- a/core/src/main/scala/org/apache/spark/util/Utils.scala +++ b/core/src/main/scala/org/apache/spark/util/Utils.scala @@ -685,7 +685,7 @@ private[spark] object Utils extends Logging { */ def getCallSiteInfo: CallSiteInfo = { val trace = Thread.currentThread.getStackTrace().filter( el => - (!el.getMethodName.contains("getStackTrace"))) + ((!el.getMethodName.contains("getStackTrace")) && (el.getClassName != "scala.Option"))) // Keep crawling up the stack trace until we find the first function not inside of the spark // package. We track the last (shallowest) contiguous Spark method. This might be an RDD @@ -697,7 +697,7 @@ private[spark] object Utils extends Logging { var finished = false var firstUserClass = "" - for (el <- trace if el.getClassName != "scala.Option") { + for (el <- trace) { if (!finished) { if (SPARK_CLASS_REGEX.findFirstIn(el.getClassName).isDefined) { lastSparkMethod = if (el.getMethodName == "") { From f74aad108ef51dfd45cd36cee6ac0d671d314bbc Mon Sep 17 00:00:00 2001 From: Aaron Davidson Date: Mon, 24 Mar 2014 22:36:54 -0700 Subject: [PATCH 3/3] Avoid Option while generating call site & add unit tests --- .../scala/org/apache/spark/SparkContext.scala | 3 +- .../main/scala/org/apache/spark/rdd/RDD.scala | 2 +- .../scala/org/apache/spark/util/Utils.scala | 18 +++++----- .../apache/spark/SparkContextInfoSuite.scala | 36 ++++++++++++++++++- 4 files changed, 47 insertions(+), 12 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala index a1003b7925715..4dd298177f07d 100644 --- a/core/src/main/scala/org/apache/spark/SparkContext.scala +++ b/core/src/main/scala/org/apache/spark/SparkContext.scala @@ -877,7 +877,8 @@ class SparkContext( * has overridden the call site, this will return the user's version. */ private[spark] def getCallSite(): String = { - Option(getLocalProperty("externalCallSite")).getOrElse(Utils.formatCallSiteInfo()) + val defaultCallSite = Utils.getCallSiteInfo + Option(getLocalProperty("externalCallSite")).getOrElse(defaultCallSite.toString) } /** diff --git a/core/src/main/scala/org/apache/spark/rdd/RDD.scala b/core/src/main/scala/org/apache/spark/rdd/RDD.scala index 1b43040c6d918..4f9d39f86546c 100644 --- a/core/src/main/scala/org/apache/spark/rdd/RDD.scala +++ b/core/src/main/scala/org/apache/spark/rdd/RDD.scala @@ -1041,7 +1041,7 @@ abstract class RDD[T: ClassTag]( /** User code that created this RDD (e.g. `textFile`, `parallelize`). */ @transient private[spark] val creationSiteInfo = Utils.getCallSiteInfo - private[spark] def getCreationSite = Utils.formatCallSiteInfo(creationSiteInfo) + private[spark] def getCreationSite: String = creationSiteInfo.toString private[spark] def elementClassTag: ClassTag[T] = classTag[T] diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala index 70c746441888d..b843cd6b80c51 100644 --- a/core/src/main/scala/org/apache/spark/util/Utils.scala +++ b/core/src/main/scala/org/apache/spark/util/Utils.scala @@ -676,7 +676,13 @@ private[spark] object Utils extends Logging { private val SPARK_CLASS_REGEX = """^org\.apache\.spark(\.api\.java)?(\.util)?(\.rdd)?\.[A-Z]""".r private[spark] class CallSiteInfo(val lastSparkMethod: String, val firstUserFile: String, - val firstUserLine: Int, val firstUserClass: String) + val firstUserLine: Int, val firstUserClass: String) { + + /** Returns a printable version of the call site info suitable for logs. */ + override def toString = { + "%s at %s:%s".format(lastSparkMethod, firstUserFile, firstUserLine) + } + } /** * When called inside a class in the spark package, returns the name of the user code class @@ -684,8 +690,8 @@ private[spark] object Utils extends Logging { * This is used, for example, to tell users where in their code each RDD got created. */ def getCallSiteInfo: CallSiteInfo = { - val trace = Thread.currentThread.getStackTrace().filter( el => - ((!el.getMethodName.contains("getStackTrace")) && (el.getClassName != "scala.Option"))) + val trace = Thread.currentThread.getStackTrace() + .filterNot(_.getMethodName.contains("getStackTrace")) // Keep crawling up the stack trace until we find the first function not inside of the spark // package. We track the last (shallowest) contiguous Spark method. This might be an RDD @@ -718,12 +724,6 @@ private[spark] object Utils extends Logging { new CallSiteInfo(lastSparkMethod, firstUserFile, firstUserLine, firstUserClass) } - /** Returns a printable version of the call site info suitable for logs. */ - def formatCallSiteInfo(callSiteInfo: CallSiteInfo = Utils.getCallSiteInfo) = { - "%s at %s:%s".format(callSiteInfo.lastSparkMethod, callSiteInfo.firstUserFile, - callSiteInfo.firstUserLine) - } - /** Return a string containing part of a file from byte 'start' to 'end'. */ def offsetBytes(path: String, start: Long, end: Long): String = { val file = new File(path) diff --git a/core/src/test/scala/org/apache/spark/SparkContextInfoSuite.scala b/core/src/test/scala/org/apache/spark/SparkContextInfoSuite.scala index 5cb49d9a7fce1..cd3887dcc7371 100644 --- a/core/src/test/scala/org/apache/spark/SparkContextInfoSuite.scala +++ b/core/src/test/scala/org/apache/spark/SparkContextInfoSuite.scala @@ -17,7 +17,7 @@ package org.apache.spark -import org.scalatest.FunSuite +import org.scalatest.{Assertions, FunSuite} class SparkContextInfoSuite extends FunSuite with LocalSparkContext { test("getPersistentRDDs only returns RDDs that are marked as cached") { @@ -56,4 +56,38 @@ class SparkContextInfoSuite extends FunSuite with LocalSparkContext { rdd.collect() assert(sc.getRDDStorageInfo.size === 1) } + + test("call sites report correct locations") { + sc = new SparkContext("local", "test") + testPackage.runCallSiteTest(sc) + } +} + +/** Call site must be outside of usual org.apache.spark packages (see Utils#SPARK_CLASS_REGEX). */ +package object testPackage extends Assertions { + private val CALL_SITE_REGEX = "(.+) at (.+):([0-9]+)".r + + def runCallSiteTest(sc: SparkContext) { + val rdd = sc.makeRDD(Array(1, 2, 3, 4), 2) + val rddCreationSite = rdd.getCreationSite + val curCallSite = sc.getCallSite() // note: 2 lines after definition of "rdd" + + val rddCreationLine = rddCreationSite match { + case CALL_SITE_REGEX(func, file, line) => { + assert(func === "makeRDD") + assert(file === "SparkContextInfoSuite.scala") + line.toInt + } + case _ => fail("Did not match expected call site format") + } + + curCallSite match { + case CALL_SITE_REGEX(func, file, line) => { + assert(func === "getCallSite") // this is correct because we called it from outside of Spark + assert(file === "SparkContextInfoSuite.scala") + assert(line.toInt === rddCreationLine.toInt + 2) + } + case _ => fail("Did not match expected call site format") + } + } }