1. 设置监控

  1. val spark = SparkSession.builder()
  2.   .appName(this.getClass.getSimpleName)
  3.   .master("yarn")
  4.   .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  5.   .config("spark.default.parallelism", "300")
  6.   .config("spark.sql.shuffle.partitions", "200")
  7.   .config("enableSendEmailOnTaskFail", "true")
  8.   .config("spark.extraListeners", "com.wsy.util.SparkAppListener")
  9.   .getOrCreate()
  10. spark.streams.addListener(new SparkStreamingQueryListener(this.getClass.getSimpleName))

2. 代码实现

  1. import org.apache.spark.sql.streaming.StreamingQueryListener
  3. class SparkStreamingQueryListener(appName: String) extends StreamingQueryListener {
  5.   override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {}
  7.   override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = {}
  9.   override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit = {
  10.     val mailTo = "cloud@huawei.com"
  11.     val mailTittle = s"Spark Structured Streaming Application ${appName} Exception"
  12.     val mailText = event.exception.getOrElse("OK")
  13.     val txt = Array(mailTo, mailTittle, mailText)
  14.     Message.sendMail(txt)
  15.   }
  16. }