1. 实现

当task执行failed时,发送邮件

  1. import org.apache.spark._
  2. import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd}
  4. class SparkAppListener(conf: SparkConf) extends SparkListener{
  6.   override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = synchronized {
  7.     val info = taskEnd.taskInfo
  8.     val AppName = conf.get("spark.app.name")
  9.     val taskId = info.taskId
  10.     val stageId = taskEnd.stageId
  12.     if (info != null && taskEnd.stageAttemptId != -1) {
  13.       val errorMessage: Option[String] =
  14.         taskEnd.reason match {
  15.           case e: ExceptionFailure =>
  16.             Some(e.toErrorString)
  17.           case e: TaskFailedReason =>
  18.             Some(e.toErrorString)
  19.           case _ => None
  20.         }
  21.       if (errorMessage.nonEmpty) {
  22.         if (conf.getBoolean("enableSendEmailOnTaskFail", false)) {
  23.           val messageInfo = "please note: ******AppName:"+AppName+"******\n"+
  24.             "******stageId:"+stageId+"******\n"+
  25.             "******taskId:"+taskId+"******\n"+ errorMessage.get
  26.           val args = Array("wangsheying@qq.com", "spark任务监控", messageInfo)
  27.           try {
  28.             Message.sendMail(args)
  29.           } catch {
  30.             case e: Exception => e.printStackTrace()
  31.           }
  32.         }
  33.       }
  34.     }
  35.   }
  36. }