Scala实现超长任务监控程序

本程序为Scala通过http获取yarn任务列表，判断任务执行时长，并将超时任务写入MySQL，且发送邮件提醒。

http获取yarn任务列表工具类

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.http.client.methods.{CloseableHttpResponse, HttpGet}
import org.apache.http.impl.client.{CloseableHttpClient, HttpClients}
import org.apache.http.util.EntityUtils

/**
 * http获取yarn任务列表
 */
object HttpGetResponse {

    def getResponse(url: String, header: String = null): String = {

        val httpClient: CloseableHttpClient = HttpClients.createDefault() // 创建 client 实例
        val get = new HttpGet(url) // 创建 get 实例

        if (header != null) { // 设置 header
            val json: JSONObject = JSON.parseObject(header)
            json.keySet().toArray.map((_: AnyRef).toString).foreach((key: String) => get.setHeader(key, json.getString(key)))
        }

        val response: CloseableHttpResponse = httpClient.execute(get) // 发送请求
        EntityUtils.toString(response.getEntity) // 获取返回结果
    }


}

解析返回json工具类

import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import scala.collection.mutable.ListBuffer

/**
 * 解析MR任务的json
 */
object analysisTaskUtil {

    def analysisJson(jsonStr: String): List[List[String]] = {
        //创建结果集
        var resultList: ListBuffer[List[String]] = scala.collection.mutable.ListBuffer[List[String]]()
        //resultList += "aa"

        //解析返回json
        val jsonOBJ: JSONObject = JSON.parseObject(jsonStr)
        val apps: String = jsonOBJ.getString("apps")
        val appsJson: JSONObject = JSON.parseObject(apps)
        val appJsonArray: JSONArray = appsJson.getJSONArray("app")
        //获取任务个数
        if (appJsonArray != null) {
            //遍历每个任务返回获取值
            for (index <- appJsonArray.toArray().indices) {
                val result: JSONObject = appJsonArray.getJSONObject(index)
                val id: String = result.getString("id")
                val user: String = result.getString("user")
                val queue: String = result.getString("queue")
                val state: String = result.getString("state")
                val startedTime: String = result.getString("startedTime")
                val name: String = result.getString("name")
                val ApplicationType: String = result.getString("applicationType")
                resultList += List(id, user, queue, state, startedTime,name,ApplicationType)
            }
        }

        resultList.toList

    }

读取配置文件工具类

import java.io.{BufferedInputStream, FileInputStream}
import java.util.Properties

/**
 * 读取配置信息
 */
object propertiesUtil {

    //读取配置文件
    def getProperties(fileName: String): Properties = {
        val prop = new Properties()
        val inputStream = new BufferedInputStream(new FileInputStream(fileName))
        prop.load(inputStream)
        prop
    }

}

时间处理工具类

import java.text.{ParseException, SimpleDateFormat}
import java.util.Date


/**
 * 时间处理类
 */
object timeUtil {
    val df: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")



    def tranTimeToString(tm: String): String = {
        val tim: String = df.format(new Date(tm.toLong))
        tim
    }

    def NowDate(): String = {
        val now: Date = new Date()
        val date: String = df.format(now)
        date
    }


    def timediff(startTime: String, endTime: String): Long = {
        var min: Long = 0
        try {
            val startDate: Date = df.parse(startTime)
            val endDate: Date = df.parse(endTime)
            val diff: Long = endDate.getTime - startDate.getTime
            if (diff < 0) sys.error("输入时间错误")
            min = diff / 1000 / 60
        } catch {
            case e: ParseException => e.printStackTrace()
        }
        min
    }

}

发送邮件工具类

import com.typesafe.config.ConfigFactory
import org.apache.log4j.Logger
import play.api.libs.mailer._
import java.io.{FileReader, InputStreamReader}

/**
 * 发送email
 */
object EmailUtil {

  def sendEmail(ehost:String,
                eport:Int,
                euser:String ,
                epassword:String,
                subject: String,
                content: List[List[String]],
                to: Seq[String]): Unit = {

    val logger: Logger = Logger.getLogger(EmailUtil.getClass)
    val from: String = euser //填写邮件发送地址
    val subject_pre = "   " //邮件主题告警前缀

    var emailText = ""
    content.foreach{
      task=>
//        emailText += s"""|<p align="left">${con}</p> \n"""
        emailText += s"""|<tr>
                         |<td>${task.head}</td>
                         |<td>${task(1)}</td>
                         |<td>${task(2)}</td>
                         |<td>${task(3)}</td>
                         |<td>${task(4)}</td>
                         |<td>${task(5)}</td>
                         |<td>${task(6)}</td>
                         |</tr> \n"""
    }


    val bodyHtml: Option[String] = Option(
      s"""
         |<html>
         |<body>
         |<h1 align="center">${subject_pre}  ${subject}</h1>
         |<br />
         |<h3 align="left">超时任务：</h3>
         |<table border="1px red solid" cellspacing="0">
         |<tr>
         |<th>application_id</th>
         |<th>queue</th>
         |<th>user</th>
         |<th>applicationtype</th>
         |<th>started_time</th>
         |<th>duration</th>
         |<th>name</th>
         |</tr>
         ${emailText}
         |</table>
         |<br />
         |<br />
         |<br />
         |<h3 align="center">请及时处理</h3>
         |</body>
         |</html>
      """.stripMargin)
    val charset: Option[String] = Option("utf-8") // 字符编码 默认utf-8
    // 生成邮件
    val email: Email = Email(subject, from, to, None, bodyHtml, charset)

    // STMP服务参数
    val host: String = ehost
    // STMP服务端口号
    val port: Int = eport
    // STMP服务发送者用户邮箱
    val user: Option[String] = Option(euser)
    // 在126网站中开通smtp服务，会返回一个密码
    // 这个密码很重要，填错了发不了邮件
    val password: Option[String] = Option(epassword)

    val timeout: Option[Int] = Option(10000)
    //setSocketTimeout 默认60s
    val connectionTimeout: Option[Int] = Option(10000)
    //setSocketConnectionTimeout 默认60s
    // STMP服务SMTPConfiguration
    val configuration: SMTPConfiguration = new SMTPConfiguration(
      host, port, false, false, false,
      user, password, false, timeout,
      connectionTimeout, ConfigFactory.empty(), false)
    val mailer: SMTPMailer = new SMTPMailer(configuration)

    // 发送邮件
    mailer.send(email)
    logger.info("==========邮件已发送成功！！！==========")
  }

超长任务监控程序主类

import com.sinosig.tianshu.utils.{EmailUtil, HttpGetResponse, analysisTaskUtil, propertiesUtil, timeUtil}
import org.apache.log4j.Logger
import java.util.Properties
import scala.collection.mutable.ListBuffer
import java.sql.{Connection, DriverManager}


/**
 * 定时监控长任务，任务超时邮件告警
 */

object LongTaskMonitoring {
  val logger: Logger = Logger.getLogger(LongTaskMonitoring.getClass)

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

    if (args.length != 1) {
      sys.error("缺少文件路径")
    }

    val filePath: String = args(0)

    val prop: Properties = propertiesUtil.getProperties(filePath)

    val yarnHost: String = prop.getProperty("yarnHost")

    val emailHost: String = prop.getProperty("emailHost")
    val emailPort: Int = prop.getProperty("emailPort").toInt
    val emailUser: String = prop.getProperty("emailUser")
    val emailPassword: String = prop.getProperty("emailPassword")

    val taskTime: Int = prop.getProperty("taskTime").toInt
    val emailTaskTime: Int = prop.getProperty("emailTaskTime").toInt

    val dbdriver: String = prop.getProperty("db.driver")
    val dburl: String = prop.getProperty("db.url")
    val dbuser: String = prop.getProperty("db.user")
    val dbpassword: String = prop.getProperty("db.password")

    val ApplicationType: String = prop.getProperty("ApplicationType")

    //        val dburl: String = args(1)
    //        val dbuser: String = args(2)
    //        val dbpassword: String = args(3)



    logger.info("==========配置获取完成==========")
    Class.forName(dbdriver)
    val connection = DriverManager.getConnection(dburl, dbuser, dbpassword)
    val statement = connection.createStatement

    val warnLongTask_email: ListBuffer[List[String]] = scala.collection.mutable.ListBuffer[List[String]]()
    val warnLongTask_db: ListBuffer[List[String]] = scala.collection.mutable.ListBuffer[List[String]]()
    //遍历队列，分别获取任务列表
    val yarnQueues = statement.executeQuery("select distinct queue from long_task_minitor_properties")
    while (yarnQueues.next()) {
      val queue = yarnQueues.getString("queue")
      val yarnHttpApi = s"http://$yarnHost:8088/ws/v1/cluster/apps?queue=$queue"
      val getResult = HttpGetResponse.getResponse(yarnHttpApi)

      //解析返回json
      val taskList: List[List[String]] = analysisTaskUtil.analysisJson(getResult)

      //遍历结果集，判断任务是否超时


      taskList.foreach {
        task =>
          val taskTimeDiff = timeUtil.timediff(timeUtil.tranTimeToString(task(4)), timeUtil.NowDate())

          if (task(6).toLowerCase.contains(ApplicationType)) {
          }
          else {
            if (taskTimeDiff > taskTime) {
              warnLongTask_db += task

              if (taskTimeDiff > emailTaskTime) {
                warnLongTask_email += task
              }
            }
          }
      }


      logger.info(s"""==========${yarnQueues}队列任务列表获取完成==========""")
    }

    //超时任务写入db

    val applications = statement.executeQuery("select application_id from long_task_minitor_log where update_time > DATE_SUB(now(), INTERVAL 2 DAY) ")
    var applicationLis = ListBuffer[String]()
    while (applications.next()) {
      applicationLis += applications.getString("application_id")
    }
    if (warnLongTask_db.nonEmpty) {
      val groups = statement.executeQuery("select queue,user,group_name from long_task_minitor_properties")
      val users_email = ListBuffer[(String, String, String)]()
      while (groups.next()) {
        val db_queue = groups.getString("queue")
        val db_user = groups.getString("user")
        val group = groups.getString("group_name")
        users_email += ((db_queue, db_user, group))
      }

      // 对每个用户发送邮件
      users_email.foreach { case (db_queue, db_user, group) =>

        warnLongTask_db.foreach { task =>
          if (db_queue == task(2) && db_user == task(1)) {
            val group_name = group
            val duration = Math.round(timeUtil.timediff(timeUtil.tranTimeToString(task(4)), timeUtil.NowDate()))
            val insert_sql = s"('${task.head}','${task(2)}','${task(1)}','$group_name','${task(6)}','${timeUtil.tranTimeToString(task(4))}','$duration','${task(5)}','N')"
            val oql = s"replace into long_task_minitor_log (application_id,queue,user,group_name,applicationtype,started_time,duration,name,end_flag) values $insert_sql"
            if (applicationLis.contains(task.head)) {
              statement.executeUpdate(oql)
              logger.info(s"==========${task.head}近期未结束任务再次写入数据库成功============")
              var currentApllicList = List(task.head)
              applicationLis --= currentApllicList
            } else {
              statement.executeUpdate(oql)
              logger.info(s"==========${task.head}新增任务插入数据库成功============")
            }
          }
        }
      }
    }
    applicationLis.foreach {
      task =>
        val sql = s"UPDATE long_task_minitor_log SET end_flag='Y' WHERE application_id='$task' "
        statement.executeUpdate(sql)

    }




    //超时任务告警

    if (warnLongTask_email.nonEmpty) {

      val subject = "任务超时告警"

      // 查询用户邮箱信息
      val rs = statement.executeQuery("select queue,user,email from long_task_minitor_properties")
      val users_email = ListBuffer[(String, String, String)]()
      while (rs.next()) {
        val queue = rs.getString("queue")
        val user = rs.getString("user")
        val email = rs.getString("email")
        users_email += ((queue, user, email))
      }

      // 对每个用户发送邮件
      users_email.foreach { case (queue, user, email) =>

        val content: ListBuffer[List[String]] = scala.collection.mutable.ListBuffer[List[String]]()

        // 获取对应用户的报错信息
        val userTasks = warnLongTask_email.filter(task => queue == task(2) && user == task(1))
        userTasks.foreach {
          task =>
            val duration = timeUtil.timediff(timeUtil.tranTimeToString(task(4)), timeUtil.NowDate())
            val info = s"超时任务用户：${task(1)}，超时任务ID：${task.head}"
            logger.info(info)
            val lst = s"${task.head},${task(2)},${task(1)},${task(6)},${timeUtil.tranTimeToString(task(4))},$duration,${task(5)}".split(",").toList
            content += lst
        }

        // 如果有报错信息，则发送邮件
        if (content.nonEmpty) {
          if (email.contains(";")) {
            val to = email.split(";").toSeq
            EmailUtil.sendEmail(emailHost, emailPort, emailUser, emailPassword, subject, content.toList, to)
          } else {
            val to = Seq(email)
            EmailUtil.sendEmail(emailHost, emailPort, emailUser, emailPassword, subject, content.toList, to)
          }
        }
      }
    }


    logger.info("==========本次检测完成！！！==========")
    statement.close()

  }
}