spark任务之Task失败监控

需求

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,猛戳我

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏码洞

Github上最受欢迎的Python框架Flask入门

flask最近终于发布了它的1.0版本更新,从项目开源到最近的1.0版本flask已经走过了8个年头。

7332
来自专栏互联网杂技

内存卡存储原理,你知道吗?

1、 简介: SD卡(Secure Digital Memory Card)是一种为满足安全性、容量、性能和使用环境等各方面的需求而设计的一种新型存储器...

4076
来自专栏双十二技术哥

Android性能优化(四)之内存优化实战

在上一篇《Android性能优化(三)之内存管理》中我们对Android的内存管理有了一定的认识,本篇文章从实际出发对内存进行优化,主要包含以下部分:

1243
来自专栏高性能服务器开发

(六)关于网络编程的一些实用技巧和细节

这些年,接触了形形色色的项目,写了不少网络编程的代码,从windows到linux,跌进了不少坑,由于网络编程涉及很多细节和技巧,一直想写篇文章来总结下这方面的...

3697
来自专栏Golang语言社区

论获取缓存值的正确姿势

论获取缓存值的正确姿势 cache 时至今日,大家对缓存想必不在陌生。我们身边各种系统中或多或少的都存在缓存,自从有个缓存,我们可以减少很多计算压力,提高应用程...

3728
来自专栏大史住在大前端

一统江湖的大前端(4)shell.js——穿上马甲我照样认识你

码农界存在着无数条鄙视链,linux使用者对windows的鄙视便是其中之一,cli使用者对GUI用户的嘲讽也是如此,在这样一个讲究逼格的时代,如果你的桌面上没...

1885
来自专栏逍遥剑客的游戏开发

基于Unity的编辑器开发(二): 进程间通信

先要做的, 是需要编辑器和Unity共享一部部分代码, 至少协议定义和解析我不想写两遍. 虽然有protobuf这样的工具库, 但是如果不是跨语言的话, 我觉得...

50516
来自专栏美团技术团队

Node.js Stream - 进阶篇

在构建较复杂的系统时,通常将其拆解为功能独立的若干部分。这些部分的接口遵循一定的规范,通过某种方式相连,以共同完成较复杂的任务。譬如,shell通过管道|连接各...

4054
来自专栏北京马哥教育

Python安全运维实战:针对几种特定隐藏方式的Webshell查杀

作者:北门喂猫 来源: http://www.freebuf.com/articles/web/131350.html Webshell一直都是...

2733
来自专栏高性能服务器开发

(六)关于网络编程的一些实用技巧和细节

这些年,接触了形形色色的项目,写了不少网络编程的代码,从windows到linux,跌进了不少坑,由于网络编程涉及很多细节和技巧,一直想写篇文章来总结下这方面的...

4235

扫码关注云+社区