有没有办法在Spark中捕获执行程序被杀死的异常？

Question

在执行我的Spark程序期间,有时(它的原因对我来说仍然是个谜)纱线杀死容器(执行器),给出了超出内存限制的消息.虽然Spark通过生成一个新容器重新执行任务,但我的程序确实恢复了.但是,在我的程序中,任务还会在磁盘上创建一些中间文件.当一个容器被杀死时,这些文件就会被遗忘.有没有办法我可以捕获执行程序 - 作为异常被杀死,以便我可以删除留下的中间文件.显然,异常处理代码也需要在运行执行程序的同一节点上运行,以便我可以从那里删除文件.

Answer 1

作为选项,您可以尝试使用SparkListener功能.因此,您可以创建自己的类并实现SparkListener接口,以挂钩到可以自我解释的可用事件.然后,您需要添加该自定义侦听器SparkContext.

有2种选择:

• SparkContext.addSparkListener(<your custom listener>)
• 通过spark.extraListeners属性,更多信息,请访问http://spark.apache.org/docs/latest/configuration.html#available-properties

Answer 2

添加在@Taras Matyashovskyy答案之上。

您可以使用 SparkListener 并拦截 SparkListener (Executor) 事件。

以下是可用的侦听器事件列表。

• SparkListener应用程序启动

• SparkListenerJobStart

• SparkListenerStage已提交

• SparkListenerTaskStart

• SparkListenerTaskGettingResult

• SparkListenerTaskEnd

• SparkListener阶段完成

• SparkListenerJobEnd

• SparkListener应用程序结束

• SparkListenerEnvironmentUpdate

• 已添加 SparkListenerBlockManager

• SparkListenerBlockManagerRemoved

• SparkListenerBlockUpdated

• SparkListenerUnpersistRDD

• 已添加 SparkListenerExecutor

• SparkListenerExecutorRemoved

例子 ： HeartBeatReceiver.scala

/**
 * Lives in the driver to receive heartbeats from executors..
 */
private[spark] class HeartbeatReceiver(sc: SparkContext, clock: Clock)
  extends SparkListener with ThreadSafeRpcEndpoint with Logging {

  def this(sc: SparkContext) {
    this(sc, new SystemClock)
  }

  sc.addSparkListener(this) ...

请查看可能适合您的删除原因（我还没有尝试过）