spark streaming作为实时任务,出了问题并不像离线任务重跑就可以了.对监控要求个方面要求较高.在任务失败 堵塞 卡死等情况下都需要发邮件或者短信报警.
比较普遍的方式是利用spark streaming自带的StreamingListener接口来监控.
如果前者不满足要求,我们也可以简单写个静态爬虫轮询爬取spark ui上的各种指标来diy监控.

方案一 StreamingListener接口

StreamingListener接口只需要新建一个监控类继承StreamingListener,然后重写需要的方法即可.然后记得在主类里加上 ssc.addStreamingListener执行.
以下是个简单的示例,在batch开始时监控schedulingDelay

class StreamingMonitor(ssc:StreamingContext) extends StreamingListener{
  override def onBatchStarted(batchStarted: StreamingListenerBatchStarted): Unit = {
    val Delay_ts = batchStarted.batchInfo.schedulingDelay.get
    if(Delay_ts > DELAY_MAX ){
        sendEmail(...)
    }
  }
}

...
//在main里加
    ssc.addStreamingListener(new StreamingMonitor(ssc))

值得注意的是,StreamingListener接口有多个方法可以重写

//需要监听spark streaming中各个阶段的事件只需实现这个特质中对应的事件函数即可
//本身既有注释说明
trait StreamingListener {

 /** Called when the streaming has been started */
 /** streaming 启动的事件 */
 def onStreamingStarted(streamingStarted: StreamingListenerStreamingStarted) { }

 /** Called when a receiver has been started */
 /** 接收启动事件 */
 def onReceiverStarted(receiverStarted: StreamingListenerReceiverStarted) { }

 /** Called when a receiver has reported an error */
 def onReceiverError(receiverError: StreamingListenerReceiverError) { }

 /** Called when a receiver has been stopped */
 def onReceiverStopped(receiverStopped: StreamingListenerReceiverStopped) { }

 /** Called when a batch of jobs has been submitted for processing. */
 /** 每个批次提交的事件 */
 def onBatchSubmitted(batchSubmitted: StreamingListenerBatchSubmitted) { }

 /** Called when processing of a batch of jobs has started.  */
 /** 每个批次启动的事件 */
 def onBatchStarted(batchStarted: StreamingListenerBatchStarted) { }

 /** Called when processing of a batch of jobs has completed. */
 /** 每个批次完成的事件  */
 def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted) { }

 /** Called when processing of a job of a batch has started. */
 def onOutputOperationStarted(
     outputOperationStarted: StreamingListenerOutputOperationStarted) { }

 /** Called when processing of a job of a batch has completed. */
 def onOutputOperationCompleted(
     outputOperationCompleted: StreamingListenerOutputOperationCompleted) { }
}

独立监控程序监控spark ui

有些指标可能我们用接口实现不了. 使用scala实现一个简单的静态爬虫来监控spark ui,相当于一个脚本程序替我们不停查看spark ui

//需要jsoup包来读接口并解析html
        <dependency>
            <groupId>org.jsoup</groupId>
            <artifactId>jsoup</artifactId>
        </dependency>

以下是个简单的示例来监控job页面的job执行时间是否过长.
注意有些公司 spark ui可能需要cookie信息.

object StreamingUIMonitorJob {

    val app_id = args(0)
    var dc = getJobDoc(app_id)
    if(dc == null){
      println("get Job document failed")    
      ...
    }

    //如果没有active job则等待1min,等待10min报警
    var active_job_table = dc.getElementById("activeJob-table")
    var alarm_num = 0
    while (active_job_table == null){
      if(alarm_num > 10){
        ...
      }
      Thread.sleep(60000)
      dc = getJobDoc(app_id)
      active_job_table = dc.getElementById("activeJob-table")
      alarm_num = alarm_num + 1
    }


    var durs = Array[String]()
    var batchs = Array[String]()

    //只有一个tbody tbody里可能有多个tr,一个job 一个tr
    val active_jobs = active_job_table.getElementsByTag("tbody")(0).getElementsByTag("tr")
    for(active_job <- active_jobs){
      val job_infos = active_job.getElementsByTag("td")
      println(job_infos(2).text()) //batch 时间
      batchs :+= job_infos(2).text()
      println(job_infos(3).text()) //dur 时间
      durs :+= job_infos(3).text()
    }


    if(durs.length > 0){
      try{
        var delay:Double = 0.0
        for(dur <- durs){
          val ls = dur.split(" ")
          val num = ls(0).toDouble
          if(ls(1) == "min" && num > delay) delay = num
        }
        if(delay > MAX_DELAY_TIME) {
          println(s"active job delay ${delay} min !")
          //报警
          ...
        }
      }catch {
        case e:Exception=>
          println(s"解析durs出错 Exception:${e}")
      }
    }
}

  def getJobDoc(app_id :String): Document ={
    var con = Jsoup.connect(SPARK_UI_URL + app_id )
    var cookie = ""
    // get cookie
    try {
      cookie = con.execute().cookies().toString
      cookie = cookie.substring(cookie.indexOf("{")+1,cookie.lastIndexOf("}"))
    }catch {
      case e:Exception=>
        println("get cookie exception!")
    }
    // get document
    var doc :Document = null
    try {
      con = Jsoup.connect(SPARK_UI_URL + app_id + "/jobs/?proxyapproved=true").header("Cookie",cookie)
      doc = con.get()
    }catch {
      case e:Exception=>
        println("get Job document exception!")
    }
    doc
  }

这只是一种场景需要,实际上spark ui上的所有指标都可以通过jsoup解析html监控的.