利用Spark解析Tomcat日志,并将统计结果存入Mysql数据库-阿里云开发者社区

开发者社区> netbug_nb> 正文

利用Spark解析Tomcat日志,并将统计结果存入Mysql数据库

简介: 本文试图实现的需求场景为:以学习Spark知识点为目的,编写Scala利用Spark解析800M的tomcat日志文件,打印一段时间内ERROR级别记录的前10行,统计每分钟的日志记录数,并将统计结果存入mysql数据库中。之前曾用JAVA写过一次同样的处理逻辑,但在学习了Scala之后,真的感觉在计算方面Scala要比JAVA方便的多。没有学习Scala语言的同学速度速度了啊……
+关注继续查看

本文试图实现的需求场景为:以学习Spark知识点为目的,编写Scala利用Spark解析800M的tomcat日志文件,打印一段时间内ERROR级别记录的前10行,统计每分钟的日志记录数,并将统计结果存入mysql数据库中。之前曾用JAVA写过一次同样的处理逻辑,但在学习了Scala之后,真的感觉在计算方面Scala要比JAVA方便的多。没有学习Scala语言的同学速度速度了啊……

技术要点

  • 将日志文件写入HDFS中,相对路径PATH为“nova.log”
  • 注意JAVA堆栈异常日志的处理
  • 将解析后的异常日志全部存到SparkSQL中或Hive数据仓库中
  • 通过编写SQL查询一段时间内ERROR级别记录的前10行
  • 统计每分钟的日志记录数,并将统计结果存入mysql数据库中,便于上层应用直接使用计算结果

解析前后对比

解析前:

解析后:

解析代码

LoggerApp.scala:

import java.text.SimpleDateFormat
import java.util.Date

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.rdd.RDD.rddToPairRDDFunctions
import org.apache.spark.sql.Row
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.TimestampType

/**
 * 日志解析
 */
object LoggerApp {
  def main(args: Array[String]): Unit = {
    println("<!--开始解析-->")
    val reg = "^(\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}.\\d{3}) (\\[.*\\]) (.*) (.*) - ([\\s\\S]*)$"
    val path = "nova.log"
    val sc = new SparkContext(new SparkConf().setAppName("日志解析"))
    val textRDD = sc.textFile(path)

    /**
     * 处理一条日志包括多行的情况
     */
    var key = ""
    val formatRDD = textRDD.map { x =>
      if (x.matches(reg)) {
        key = x
        Pair.apply(key, "")
      } else {
        Pair.apply(key, x)
      }
    }.reduceByKey((a, b) => { a + "\n" + b }).map(x => x._1 + x._2)

    /**
     * 将字符串转换为Logger
     */
    val loggerRDD: RDD[Logger] = formatRDD.map { x =>
      {
        val reg.r(time, thread, level, logger, msg) = x //通过正则取值
        val log = new Logger(formatDate(time), thread, level, logger, msg)
        log
      }
    }.cache()

    /**
     * TODO 通过类的反射机制来定义数据库Scheme,但在scala语言中不知道为啥就是不成功,此处浪费了许久留着以后研究吧
     */
    /*val sqlc = new SQLContext(sc)
    sqlc.createDataFrame(loggerRDD, classOf[Logger]).registerTempTable("logger")*/

    /**
     * 定义数据库Scheme
     */
    val schemaString = "time thread level logger msg"
    val schema =
      StructType(
        schemaString.split(" ").map(fieldName =>
          if ("time".equals(fieldName))
            StructField(fieldName, TimestampType, true)
          else
            StructField(fieldName, StringType, true)))
    /**
     * 将Logger转换为Row
     */
    val rowRDD = loggerRDD.map { log =>
      Row(
        formatDate(log.time),
        log.thread,
        log.level,
        log.logger,
        log.msg)
    }
    /**
     * 利用SQL进行查询过滤
     */
    //    val sqlc = bySQLContext(sc, rowRDD, schema);
    val sqlc = byHiveContext(sc, rowRDD, schema);
    val df = sqlc.sql("select * from logger where level='ERROR' and time between '2016-03-21 11:00:00' and '2016-03-21 12:00:00' order by time")
    val errLogRDD = df.map { x =>
      new Logger(
        formatDate(x.getTimestamp(0)),
        x.getString(1),
        x.getString(2),
        x.getString(3),
        x.getString(4))
    }
    for (log <- errLogRDD.take(10)) {
      println("time:" + formatDateToStr(log.time))
      println("thread:" + log.thread)
      println("level:" + log.level)
      println("logger:" + log.logger)
      println("msg:" + log.msg)
    }
    println("<!--解析结束-->")
  }
  /**
   * 创建临时表
   */
  def bySQLContext(sc: SparkContext, rowRDD: RDD[Row], schema: StructType): SQLContext = {
    val sqlc = new SQLContext(sc)
    sqlc.createDataFrame(rowRDD, schema).registerTempTable("logger")
    sqlc
  }
  /**
   * 创建永久表,需要提前搭建好Spark与Hive的集成环境
   */
  def byHiveContext(sc: SparkContext, rowRDD: RDD[Row], schema: StructType): SQLContext = {
    val sqlc = new HiveContext(sc)
    sqlc.sql("drop table if exists logger")
    sqlc.sql("CREATE TABLE IF NOT EXISTS logger (time TIMESTAMP, thread STRING, level STRING, logger STRING, msg STRING)")
    sqlc.createDataFrame(rowRDD, schema).write.mode("overwrite").saveAsTable("logger")
    sqlc
  }
  def formatDate(str: String): Date = {
    new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").parse(str)
  }
  def formatDate(timestamp: java.sql.Timestamp): Date = {
    new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").parse(timestamp.toString())
  }
  def formatDate(date: Date): java.sql.Timestamp = {
    new java.sql.Timestamp(date.getTime)
  }
  def formatDateToStr(date: Date): String = {
    new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(date)
  }
}

Logger.scala:

import java.util.Date

class Logger extends Serializable {
  var time: Date = null
  var thread: String = ""
  var level: String = ""
  var logger: String = ""
  var msg: String = ""
  def this(time: Date, thread: String, level: String, logger: String, msg: String) {
    this()
    this.time = time;
    this.thread = thread;
    this.level = level;
    this.logger = logger;
    this.msg = msg;
  }
}

统计并写入Mysql

LoggerMysqlApp.scala:

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.SQLContext
import java.util.Date
import java.text.SimpleDateFormat
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.Row
import java.util.Properties

object LoggerMysqlApp {
  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf().setAppName("输出写入Mysql"))
    /**
     * 从hive中加载数据
     */
    val hivec = new HiveContext(sc)
    val df = hivec.sql("select * from logger")
    val loggerRDD = df.rdd.map { x =>
      new Logger(
        LoggerApp.formatDate(x.getTimestamp(0)),
        x.getString(1),
        x.getString(2),
        x.getString(3),
        x.getString(4))
    }
    val resultRDD = loggerRDD.map { logger =>
      Pair(formatDateToStr(logger.time), 1)
    }.reduceByKey((a, b) =>
      { a + b }).map(f =>
      Row(f._1, f._2)).sortBy(f => f.getInt(1), false, 2)
    for (r <- resultRDD.take(10)) {
      println(r.getString(0) + ":" + r.getInt(1))
    }
    /**
     * 定义数据库Scheme
     */
    val schemaString = "time count"
    val schema =
      StructType(
        schemaString.split(" ").map(fieldName =>
          if ("time".equals(fieldName))
            StructField(fieldName, StringType, true)
          else
            StructField(fieldName, IntegerType, true)))
    /**
     * TODO计算每分钟日志的个数
     */
    val connectionProperties = new Properties()
    connectionProperties.setProperty("user", "root")
    connectionProperties.setProperty("password", ".")
    new SQLContext(sc).createDataFrame(resultRDD, schema).write.jdbc(
      "jdbc:mysql://192.168.136.128:3306/logger",
      "logger",
      connectionProperties);
  }
  def formatDateToStr(date: Date): String = {
    new SimpleDateFormat("yyyy-MM-dd HH:mm").format(date)
  }
}


版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
OAF_文件系列5_实现OAF解析XML文件javax.xml.parsers(案例)
20150729 Created By BaoXinjian 一、摘要 通过javax.xml.parsers.DocumentBuilder解析从系统系统导入的XML文件 此过程需要用到一些包方法 1.
820 0
Apache Spark 3.0.0重磅发布 —— 重要特性全面解析
开发了近两年(自2018年10月份至今)的Apache SparkTM 3.0.0正式发布! Apache SparkTM 3.0.0版本包含3400多个补丁,是开源社区做出巨大贡献的结晶,在Python和SQL功能方面带来了重大进展并且将重点聚焦在了开发和生产的易用性上。同时,今年也是Spark开源10周年,这些举措反映了Spark自开源以来,是如何不断的满足更广泛的受众需求以及更多的应用场景
3161 0
c++ 开源日志库选择
liblogger 待选为 glog、log4cplus、log4cpp、log4cxx 目前准备使用glog,使用方便,性能也不错,待进一步试验,如果有不能满足的功能就转用 log4cplus,功能很全面,不过稍复杂些。
1501 0
JDBC判断数据库查询结果集是否为空
通常来说都是用rs.next()来判断结果集是否为空,但是由于执行rs.next()后指针指向的是结果集中的第一条记录,此时再用while(rs.next())取结果集中的数据就会导致第一条数据无法得到。
797 0
+关注
15
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
《2021云上架构与运维峰会演讲合集》
立即下载
《零基础CSS入门教程》
立即下载
《零基础HTML入门教程》
立即下载