Хорошо, поэтому я спросил несколько похожий question, связанный с тем, как Spark обрабатывает исключения внутри, но пример, который у меня был тогда, на самом деле не был ясно или полно. Ответ там указал мне в каком-то направлении, но я не могу объяснить некоторые вещи.
Я установил фиктивное искрообразование, и на этапе трансформации у меня есть выражение русско-рулетка, которое может или не должно вызывать исключение. Если возникает исключение, я останавливаю контекст потока искрообразования. То, что это, никакая другая логика, не преобразование RDD.
object ImmortalStreamingJob extends App {
val conf = new SparkConf().setAppName("fun-spark").setMaster("local[*]")
val ssc = new StreamingContext(conf, Seconds(1))
val elems = (1 to 1000).grouped(10)
.map(seq => ssc.sparkContext.parallelize(seq))
.toSeq
val stream = ssc.queueStream(mutable.Queue[RDD[Int]](elems: _*))
val transformed = stream.transform { rdd =>
try {
if (Random.nextInt(6) == 5) throw new RuntimeException("boom")
else println("lucky bastard")
rdd
} catch {
case e: Throwable =>
println("stopping streaming context", e)
ssc.stop(stopSparkContext = true, stopGracefully = false)
throw e
}
}
transformed.foreachRDD { rdd =>
println(rdd.collect().mkString(","))
}
ssc.start()
ssc.awaitTermination()
}
Запуск этого в IntelliJ в какой-то момент вызовет исключение. Веселая часть:
- Если исключение выбрано в первом преобразовании (когда обрабатывается первый RDD), контекст искры останавливается и приложение умирает, что я хочу
- если исключение генерируется после обработки хотя бы одного
RDD, приложение зависает после печати сообщения об ошибке и никогда не останавливается, что не то, что я хочу
Почему приложение во время замены висит, а не умирает?
Я запускаю Spark 2.1.0 на Scala 2.11.8. Выход из try-catch решает проблему (Spark останавливается сам по себе). Кроме того, устранение проблемы с помощью try-catch внутри foreachRDD решает проблему.
Однако я ищу ответ, который поможет мне понять, что происходит в этом конкретном примере.
