需求
spark應(yīng)用程序中，只要task失敗就發(fā)送郵件，并攜帶錯(cuò)誤原因。
背景
在spark程序中，task有失敗重試機(jī)制（根據(jù) spark.task.maxFailures 配置，默認(rèn)是4次），當(dāng)task執(zhí)行失敗時(shí)，并不會(huì)直接導(dǎo)致整個(gè)應(yīng)用程序down掉，只有在重試了 spark.task.maxFailures 次后任然失敗的情況下才會(huì)使程序down掉。另外，spark on yarn模式還會(huì)受yarn的重試機(jī)制去重啟這個(gè)spark程序，根據(jù) yarn.resourcemanager.am.max-attempts 配置（默認(rèn)是2次）。
即使spark程序task失敗4次后，受yarn控制重啟后在第4次執(zhí)行成功了，一切都好像沒(méi)有發(fā)生，我們只有通過(guò)spark的監(jiān)控UI去看是否有失敗的task，若有還得去查找看是哪個(gè)task由于什么原因失敗了?；谝陨显?，我們需要做個(gè)task失敗的監(jiān)控，只要失敗就帶上錯(cuò)誤原因通知我們，及時(shí)發(fā)現(xiàn)問(wèn)題，促使我們的程序更加健壯。
捕獲Task失敗事件
順藤摸瓜，task在Executor中執(zhí)行，跟蹤源碼看task在失敗后都干了啥？
- 在executor中task執(zhí)行完不管成功與否都會(huì)向execBackend報(bào)告task的狀態(tài)；
execBackend.statusUpdate(taskId, TaskState.FINISHED, serializedResult)
- 在CoarseGrainedExecutorBackend中會(huì)向driver發(fā)送StatusUpdate狀態(tài)變更信息；
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")
}
}
- CoarseGrainedSchedulerBackend收到消息后有調(diào)用了scheduler的方法；
override def receive: PartialFunction[Any, Unit] = {
case StatusUpdate(executorId, taskId, state, data) =>
scheduler.statusUpdate(taskId, state, data.value)
......
- 由于代碼繁瑣，列出了關(guān)鍵的幾行代碼，嵌套調(diào)用關(guān)系，這里最后向eventProcessLoop發(fā)送了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))
- 在
DAGSchedulerEventProcessLoop處理方法中handleTaskCompletion(event: CompletionEvent)有著最為關(guān)鍵的一行代碼，這里listenerBus把task的狀態(tài)發(fā)了出去，凡是監(jiān)聽(tīng)了SparkListenerTaskEnd的listener都可以獲取到對(duì)應(yīng)的消息，而且這個(gè)是帶了失敗的原因（event.reason）。其實(shí)第一遍走源碼并沒(méi)有注意到前面提到的sched.dagScheduler.taskEnded(tasks(index), reason, null, accumUpdates, info)方法，后面根據(jù)SparkUI的page頁(yè)面往回追溯才發(fā)現(xiàn)。
listenerBus.post(SparkListenerTaskEnd(
stageId, task.stageAttemptId, taskType, event.reason, event.taskInfo, taskMetrics))
自定義監(jiān)聽(tīng)器
需要獲取到SparkListenerTaskEnd事件，得繼承SparkListener類并重寫onTaskEnd方法，
在方法中獲取task失敗的reason，發(fā)送郵件給對(duì)應(yīng)的負(fù)責(zé)人。這樣我們就可以第一時(shí)間知道哪個(gè)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任務(wù)監(jiān)控", errorMessage.get)
try {
MailUtil.sendMail(args)
} catch {
case e: Exception =>
}
}
}
}
}
}
注意這里還需要在我們的spark程序中注冊(cè)好這個(gè)listener：
.config("enableSendEmailOnTaskFail", "true")
.config("spark.extraListeners", "cn.i4.monitor.streaming.I4SparkAppListener")
總結(jié)
這里只是實(shí)現(xiàn)了一個(gè)小demo，可以做的更完善使之更通用，比如加上應(yīng)用程序的名字、host、stageid、taskid等，單獨(dú)達(dá)成jia包放到classPath，并把該listener的注冊(cè)放到默認(rèn)配置文件中永久有效，只需控制enableSendEmailOnTaskFail控制是否啟用。
我的GitHub，猛戳我
