Sorry, I guess my code and the exception didn't make it to the mailing
list. Here's my code:
def main(args: Array[String]) {
val conf = new SparkConf().setAppName("Test app")
val sc = new SparkContext(conf)
val rdd = sc.parallelize(Array(1, 2, 3))
val rdd1 = rdd.map({x =>
throw new Exception("Test exception 12345")
x
})
rdd1.foreachPartition(part => {
val t = new Thread(new Runnable {
override def run(): Unit = {
for (row <- part) {
Console.println(s"Got $row")
}
}
})
t.start()
t.join()
})
}
And here's the exception:
15/08/31 19:42:20 ERROR SparkUncaughtExceptionHandler: Uncaught exception
in thread Thread[Thread-7,5,main]
java.lang.Exception: Test exception 12345
at TestApp$anonfun$1.apply$mcII$sp(TestApp.scala:15)
at TestApp$anonfun$1.apply(TestApp.scala:14)
at TestApp$anonfun$1.apply(TestApp.scala:14)
at scala.collection.Iterator$anon$11.next(Iterator.scala:328)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at TestApp$anonfun$main$1$anon$1.run(TestApp.scala:21)
at java.lang.Thread.run(Thread.java:745)
Anyways, I realized that this is occurs because of the
SparkUncaughtExceptionHandler, which executors set as their threads'
default uncaught exception handler. Any exceptions that happen in the main
thread get caught here:
https://github.com/apache/spark/blob/69c9c177160e32a2fbc9b36ecc52156077fca6fc/core/src/main/scala/org/apache/spark/executor/Executor.scala#L294
And thus get logged to task metrics. The SparkUncaughtExceptionHandler,
however, catches stuff in subthreads, and therefore those exceptions don't
get logged in task metrics (and thus not in the web UI).
My fix was to catch exceptions within my threads, like this:
rdd1.foreachPartition(part => {
var exception = null
val t = new Thread(new Runnable {
override def run(): Unit = {
try {
for (row <- part) {
Console.println(s"Got $row")
}
} catch {
case e: Exception => exception = e
}
}
})
t.start()
t.join()
if (exception != null) {
throw e
}
})
It'd be nice if the uncaught exception handler in executors logged
exceptions to task metrics, but I'm not sure how feasible that'd be.
On Thu, Sep 3, 2015 at 12:26 AM, Akhil Das <ak...@sigmoidanalytics.com>
wrote:
> [image: Inline image 1]
>
> I'm not able to find the piece of code that you wrote, but you can use a
> try...catch to catch your user specific exceptions and log it in the logs.
>
> Something like this:
>
> myRdd.map(x => try{ //something }catch{ case e:Exception =>
> log.error("Whoops!! :" + e) })
>
>
>
>
> Thanks
> Best Regards
>
> On Tue, Sep 1, 2015 at 1:22 AM, Wayne Song <wayne.e.s...@gmail.com> wrote:
>
>> We've been running into a situation where exceptions in rdd.map() calls
>> will
>> not get recorded and shown on the web UI properly. We've discovered that
>> this seems to occur because we're creating our own threads in
>> foreachPartition() calls. If I have code like this:
>>
>>
>>
>> The tasks on the executors will fail because rdd1 will raise an exception
>> for each record as we iterate across the "part" iterator inside the thread
>> in the foreachPartition call. Usually, exceptions in Spark apps show up
>> in
>> the web UI on the application detail page, making problems easy to debug.
>> However, if any exceptions get raised inside of these user threads, they
>> don't show up in the web UI (instead, it just says that the executor was
>> lost), and in the executor logs, we see errors like:
>>
>>
>>
>> What's going on here? Why are these exceptions not caught? And is there
>> a
>> way to have user threads register their exceptions properly?
>>
>>
>>
>> --
>> View this message in context:
>> http://apache-spark-user-list.1001560.n3.nabble.com/Exceptions-in-threads-in-executor-code-don-t-get-caught-properly-tp24525.html
>> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>>
>> ---------------------------------------------------------------------
>> To unsubscribe, e-mail: user-unsubscr...@spark.apache.org
>> For additional commands, e-mail: user-h...@spark.apache.org
>>
>>
>
