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

news/2024/5/20 4:51:50 标签: spark, hdfs, scala, mysql

spark.apache.org/images/spark-logo-trademark.png" alt="" />

本文试图实现的需求场景为：以学习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)
  }
}



http://www.niftyadmin.cn/n/748520.html

相关文章

学习记录ES6反射内置对象Proxy与Reflect

学习记录ES6反射内置对象Proxy与Reflect

工作中遇见了一个这样的需求&#xff1a;网页预览视频文件&#xff08;格式有两种mp4与wmv&#xff09;时&#xff0c;mp4格式的采用浏览器自带的video标签进行播放&#xff0c;wmv格式的由于浏览器不支持直接播放&#xff0c;则需要调用定制的wmvH5播放器来播放。 经调研发现…
阅读更多...
Mybatis(二)——全局配置文件

Mybatis(二)——全局配置文件

引入dtd约束 想要在mybatis-config.xml中敲代码有提示,要引入.did文件&#xff0c;http://mybatis.org/dtd/mybatis-3-config.dtd&#xff0c;要连网。 <!DOCTYPE configurationPUBLIC "-//mybatis.org//DTD Config 3.0//EN""http://mybatis.org/dtd/mybat…
阅读更多...
通过过滤器Filter实现请求响应报文的日志输出

通过过滤器Filter实现请求响应报文的日志输出

背景知识 项目组需要打印所有Restful接口的请求及响应报文&#xff0c;而通过HttpServletRequest获取输入流InputStream最多只能获取一次&#xff0c;重复获取会抛异常——Required Request Body is Missing。 相关技术知识点&#xff1a; 接口/类描述主要方法javax.servlet…
阅读更多...
SpringBoot2.x——多环境配置

SpringBoot2.x——多环境配置

方式一&#xff1a;创建多个环境的配置文件&#xff0c;然后在application.yml中加上一个spring.profiles.activexxx ,分别指定配置文件的后缀就可以使用对应环境的配置了 https://blog.csdn.net/husong_/article/details/79784265方式二&#xff1a;使用spring为我们提供的Pr…
阅读更多...
springboot整合Druid(德鲁伊)数据库连接池

springboot整合Druid(德鲁伊)数据库连接池

目录方式一、直接整合引入数据源配置文件配置类方式二、通过SpringbootStarter整合界面信息简单说明方式一、直接整合 引入数据源 <!--druid --> <dependency><groupId>com.alibaba</groupId><artifactId>druid</artifactId><version…
阅读更多...
Mybatis(三)——映射文件1

Mybatis(三)——映射文件1

增删改查 在TblEmployeeMapper.java中加入 int deleteByPrimaryKey(Integer id); int insert(TblEmployeePO record); TblEmployeePO selectByPrimaryKey(Integer id); int updateByPrimaryKey(TblEmployeePO record);然后到SQL映射文件(TblEmployeeMapper.xml)中实现 <!…
阅读更多...
java命名规则以及java概述

java命名规则以及java概述

java发展史&#xff1a;1.1996年&#xff0c;sun发布java1.0&#xff0c;不适合真正的应用开发2.1998年发布的1.2版本的swing取代了GUI3.2000年发布的java1.3才出现java Webjava特点&#xff1a;1.简单2.面向对象3.分布式4.健壮性1.编译和运行时对问题进行检查2.提供自动垃圾收…
阅读更多...
Mybatis(三)——映射文件2

Mybatis(三)——映射文件2

select_返回List select元素Select元素来定义查询操作Id&#xff1a;唯一标识符 用来引用这条语句&#xff0c;需要和接口的方法名一致 parameterType&#xff1a;参数类型 可以不传&#xff0c;MyBatis会根据TypeHandler自动推断 resultType&#xff1a;返回值类型 别名或者全…
阅读更多...
最新文章

Copyright @ 2022~2023