spark任务之Task失败监控

作者: BIGUFO | 来源:发表于2018-04-13 22:16 被阅读140次

    需求

    spark应用程序中,只要task失败就发送邮件,并携带错误原因。

    背景

    在spark程序中,task有失败重试机制(根据 spark.task.maxFailures 配置,默认是4次),当task执行失败时,并不会直接导致整个应用程序down掉,只有在重试了 spark.task.maxFailures 次后任然失败的情况下才会使程序down掉。另外,spark on yarn模式还会受yarn的重试机制去重启这个spark程序,根据 yarn.resourcemanager.am.max-attempts 配置(默认是2次)。

    即使spark程序task失败4次后,受yarn控制重启后在第4次执行成功了,一切都好像没有发生,我们只有通过spark的监控UI去看是否有失败的task,若有还得去查找看是哪个task由于什么原因失败了。基于以上原因,我们需要做个task失败的监控,只要失败就带上错误原因通知我们,及时发现问题,促使我们的程序更加健壮。

    捕获Task失败事件

    顺藤摸瓜,task在Executor中执行,跟踪源码看task在失败后都干了啥?

    1. 在executor中task执行完不管成功与否都会向execBackend报告task的状态;
     execBackend.statusUpdate(taskId, TaskState.FINISHED, serializedResult)
    
    1. 在CoarseGrainedExecutorBackend中会向driver发送StatusUpdate状态变更信息;
    override def statusUpdate(taskId: Long, state: TaskState, data: ByteBuffer) {
        val msg = StatusUpdate(executorId, taskId, state, data)
        driver match {
          case Some(driverRef) => driverRef.send(msg)
          case None => logWarning(s"Drop $msg because has not yet connected to driver")
        }
      }
    
    1. CoarseGrainedSchedulerBackend收到消息后有调用了scheduler的方法;
    override def receive: PartialFunction[Any, Unit] = {
          case StatusUpdate(executorId, taskId, state, data) =>
            scheduler.statusUpdate(taskId, state, data.value)
            ......
    
    1. 由于代码繁琐,列出了关键的几行代码,嵌套调用关系,这里最后向eventProcessLoop发送了CompletionEvent事件;
    taskResultGetter.enqueueFailedTask(taskSet, tid, state, serializedData)
    scheduler.handleFailedTask(taskSetManager, tid, taskState, reason)
    taskSetManager.handleFailedTask(tid, taskState, reason)
    sched.dagScheduler.taskEnded(tasks(index), reason, null, accumUpdates, info)
    eventProcessLoop.post(CompletionEvent(task, reason, result, accumUpdates, taskInfo)) 
    
    1. DAGSchedulerEventProcessLoop处理方法中 handleTaskCompletion(event: CompletionEvent)有着最为关键的一行代码,这里listenerBus把task的状态发了出去,凡是监听了SparkListenerTaskEnd的listener都可以获取到对应的消息,而且这个是带了失败的原因(event.reason)。其实第一遍走源码并没有注意到前面提到的sched.dagScheduler.taskEnded(tasks(index), reason, null, accumUpdates, info)方法,后面根据SparkUI的page页面往回追溯才发现。
     listenerBus.post(SparkListenerTaskEnd(
           stageId, task.stageAttemptId, taskType, event.reason, event.taskInfo, taskMetrics))
    

    自定义监听器

    需要获取到SparkListenerTaskEnd事件,得继承SparkListener类并重写onTaskEnd方法,
    在方法中获取task失败的reason,发送邮件给对应的负责人。这样我们就可以第一时间知道哪个task是以什么原因失败了。

    import cn.i4.utils.MailUtil
    import org.apache.spark._
    import org.apache.spark.internal.Logging
    import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd}
    
    class I4SparkAppListener(conf: SparkConf) extends SparkListener with Logging {
    
      override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = synchronized {
        val info = taskEnd.taskInfo
        // If stage attempt id is -1, it means the DAGScheduler had no idea which attempt this task
        // completion event is for. Let's just drop it here. This means we might have some speculation
        // tasks on the web ui that's never marked as complete.
        if (info != null && taskEnd.stageAttemptId != -1) {
          val errorMessage: Option[String] =
            taskEnd.reason match {
              case kill: TaskKilled =>
                Some(kill.toErrorString)
              case e: ExceptionFailure =>
                Some(e.toErrorString)
              case e: TaskFailedReason =>
                Some(e.toErrorString)
              case _ => None
            }
          if (errorMessage.nonEmpty) {
            if (conf.getBoolean("enableSendEmailOnTaskFail", false)) {
              val args = Array("********@qq.com", "spark任务监控", errorMessage.get)
              try {
                MailUtil.sendMail(args)
              } catch {
                case e: Exception =>
              }
            }
          }
        }
      }
    }
    

    注意这里还需要在我们的spark程序中注册好这个listener:

    .config("enableSendEmailOnTaskFail", "true")
    .config("spark.extraListeners", "cn.i4.monitor.streaming.I4SparkAppListener")
    

    总结

    这里只是实现了一个小demo,可以做的更完善使之更通用,比如加上应用程序的名字、host、stageid、taskid等,单独达成jia包放到classPath,并把该listener的注册放到默认配置文件中永久有效,只需控制enableSendEmailOnTaskFail控制是否启用。

    我的GitHub,猛戳我

    相关文章

      网友评论

        本文标题:spark任务之Task失败监控

        本文链接:https://www.haomeiwen.com/subject/kemhkftx.html