Spark Streaming Listener 监控批次处理延迟进行告警

概述

StreamingListener 是针对spark streaming的各个阶段的事件监听机制。

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) { }

}

自定义StreamingListener

功能：监控批次处理时间，若超过阈值则告警，每次告警间隔2分钟

class SparkStreamingDelayListener(private val appName:String, private val duration: Int,private val times: Int) extends StreamingListener{

  private val logger = LoggerFactory.getLogger("SparkStreamingDelayListener")

//每个批次完成时执行

  override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = {

    val batchInfo = batchCompleted.batchInfo

    val processingStartTime = batchCompleted.batchInfo.processingStartTime

    val numRecords = batchCompleted.batchInfo.numRecords

    val processingEndTime = batchInfo.processingEndTime

    val processingDelay = batchInfo.processingDelay

    val totalDelay = batchInfo.totalDelay

    //将每次告警时间写入redis,用以判断告警间隔大于2分钟

    val jedis = RedisClusterClient.getJedisClusterClient()

    val current_time = (System.currentTimeMillis / 1000).toInt

    val redis_time = jedis.get(appName)

    var flag = false

    if(redis_time==null || current_time-redis_time.toInt>120){

      jedis.set(appName,current_time.toString)

      flag = true

    }

    //若批次处理延迟大于批次时长指定倍数，并且告警间隔大约2分钟，则告警

    if(totalDelay.get >= times * duration * 1000 && flag){

      val monitorContent = appName+": numRecords ->"+numRecords+",processingDelay ->"+processingDelay.get/1000+" s,totalDelay -> "+totalDelay.get/1000+"s"

      println(monitorContent)

      val msg = "Streaming_"+appName+"_DelayTime:"+totalDelay.get/1000+"S"

      val getURL = "http://node1:8002/message/weixin?msg="+msg

      HttpClient.doGet(getURL)

    }

  }

}

应用

//streamingListener不需要在配置中设置，可以直接添加到streamingContext中

object My{

    def main(args : Array[String]) : Unit = {

        val sparkConf = new SparkConf()

        val ssc = new StreamingContext(sparkConf,Seconds(20))

        ssc.addStreamingListener(new SparkStreamingDelayListener("Userid2Redis", duration,times))

        ....

    }

}

巴特西

Spark Streaming Listener 监控批次处理延迟进行告警

概述

StreamingListener接口

自定义StreamingListener

应用

最新文章

热门文章