Spark SQL实现日志离线批处理

时间  2019-11-08
标签 spark sql 实现 日志 离线 批处理 栏目 Spark 繁體版
原文   原文链接
1、 基本的离线数据处理架构：
 
  1. 数据采集   Flume：Web日志写入到HDFS
  2. 数据清洗   脏数据 Spark、Hive、MR等计算框架来完成。 清洗完以后再放回HDFS
  3. 数据处理   按照须要，进行业务的统计和分析。 也经过计算框架完成
  4. 处理结果入库   存放到RDBMS、NoSQL中
  5. 数据可视化    经过图形化展现出来。  ECharts、HUE、Zeppelin
 
处理框图：
 
1 2 3 4 5 6 7为离线处理，其中5不必定是Hive（还有Spark SQL等） 6不必定是RDBMS（NoSQL）
执行时，可用调度框架Oozie、Azkaban，指定任务执行的时间
 
另一条线是实时处理
 
 
拟定项目需求：
  1. 统计某时间段最受欢迎的某项的TopN和对应的访问次数
  2. 按地市统计最受欢迎  从IP提取城市信息
  3. 按访问流量统计
 
 
互联网日志通常包括有：
访问时间  访问URL  耗费流量   访问IP地址
从日志里提取以上咱们须要的数据
 
假设咱们如今仅有一台电脑供学习做为集群使用，为了防止内存溢出，有必要进行剪切日志：
用head -10000命令截取前10000条
数据量太大的话，在IDE中可能会报错
 
 
 
 2、日志处理过程
 
数据清洗：
 
第一步： 从原始日志提取有用信息，本例中就是拿到时间、URL、流量、IP
  1. 读取日志文件，获得RDD，经过map方法，split成一个数组，而后选择数组中有用的几项（用断点的方法分析哪几项有用，并匹配相应的变量）
  2. 获取到的信息有可能由于某些问题，如线程问题而致使生成了带有错误的信息，第一步中一开始用了SimpleDateFormat（线程不安全）来转变时间格式，会致使某些时间转换错误。通常要改为FastDateFormat来作
 
实现代码： 
//提取有用信息，转换格式
object SparkStatFormatJob {
  def main(args: Array[String]) = {
    val spark = SparkSession.builder().appName("SparkStatFormatJob").master("local[2]").getOrCreate()
    val access = spark.sparkContext.textFile("/Users/kingheyleung/Downloads/data/10000_access.log")
    //access.take(10).foreach(println)
    access.map(line => {
      val splits = line.split(" ")
      val ip = splits(0)
      //用断点的方法，观察splits数组，找出时间、url、流量对应哪个字段
      //建立时间类DateUtils，转换成经常使用的时间表达方式
      //把url多余的""引号清除掉
      val time = splits(3) + " " + splits(4)
      val url = splits(11).replaceAll("\"", "")
      val traffic = splits(9)
      //(ip, DateUtils.parse(time), url, traffic)  用来测试输出是否正常
      //把裁剪好的数据从新组合,用Tab分割
      DateUtils.parse(time) + "\t" + url + "\t" + traffic + "\t" + ip
    }).saveAsTextFile("file:///usr/local/mycode/immooclog/")
    spark.stop()
  }
}

 

//日期解析
object DateUtils {
  //输入格式
  val ORIGINAL_TIME_FORMAT = FastDateFormat.getInstance("dd/MMM/yyyy:HH:mm:sss Z", Locale.ENGLISH)
  //输出格式
  val TARGET_TIME_FORMAT = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")
 
  def parse(time:String) = {
    TARGET_TIME_FORMAT.format(new Date(getTime(time)))
  }
  def getTime(time:String) = {
    try {
      ORIGINAL_TIME_FORMAT.parse(time.substring(time.indexOf("[") + 1, time.lastIndexOf("]"))).getTime
    } catch { 
      case e : Exception => {
        0l
      }
    }
  }

 

通常日志处理须要进行分区java

本例中按照日志中的访问时间进行分区
 
 
第二步：解析上一步获得的有用信息，我把它称为解析日志
 
其实就是把较为整洁的数据日志，解析出每一个字段的含义，并把RDD转成DF
在此案例中，完成的是：
输入：访问时间  访问URL  耗费流量   访问IP地址  =>转变为输出：url、类型（本例中url的后缀有article仍是video）、对应ID号、流量、ip、城市、时间、天（用于分组）
而且建立DataFrame（也就是定义Row和StructType，其中Row要和原日志的每一个字段对应，而StructType是根据所须要的输出来定义就行）
 
实现代码： 
//解析日志
object SparkStatCleanJob {
  def main(args: Array[String]) = {
    val spark = SparkSession.builder().appName("SparkStatCleanJob").master("local[2]").getOrCreate()
    val accessRDD = spark.sparkContext.textFile("file:///Users/kingheyleung/Downloads/data/access_10000.log")
    //RDD convert to DF, define Row and StructType
    val accessDF = spark.createDataFrame(accessRDD.map(line => LogConvertUtils.convertToRow(line)), LogConvertUtils.struct)
    //accessDF.printSchema()
    //accessDF.show(false)
    spark.stop()
  }
}

 

//RDD转换成DF的工具类
object LogConvertUtils {
  //构建Struct
  val struct = StructType(
    Array(
      StructField("url", StringType),
      StructField("cmsType", StringType),
      StructField("cmsId", LongType),
      StructField("traffic", LongType),
      StructField("ip", StringType),
      StructField("city", StringType),
      StructField("time", StringType),
      StructField("day", StringType)
    )
  )
  //提取信息，构建Row
  def convertToRow(line:String) = {
 
    try {
      val splits = line.split("\t")
      val url = splits(1)
      val traffic = splits(2).toLong
      val ip = splits(3)
      val domain = "http://www.imooc.com/"
      val cms = url.substring(url.indexOf(domain) + domain.length())
      val cmsSplits = cms.split("/")
 
      var cmsType = ""
      var cmsId = 0l
      //判断是否存在
      if (cmsSplits.length > 1) {
        cmsType = cmsSplits(0)
        cmsId = cmsSplits(1).toLong
      }
      val city = IpUtils.getCity(ip)     //经过Ip解析工具传进，具体看下面
      val time = splits(0)
      val day = time.substring(0, 10).replaceAll("-", "")
 
      //定义Row，与Struct同样
      Row(url, cmsType, cmsId, traffic, ip, city, time, day)
    } catch {
      case e: Exception => Row(0)
    }
  }
}

注意：转换时必定要记得类型转换！！！！mysql

 
进一步解析：对IP地址解析来得到城市信息
 
在这里，为了让IP地址转换成直观的城市信息，我使用了GitHub上的开源项目来实现：
用Maven编译下载的项目
mvn clean package -DskipTests
 
安装jar包到本身的Maven仓库中： 
mvn install:install-file -Dfile=路径.jar -DgroupId=com.ggstar -DartifactId=ipdatabase -Dversion=1.0 -Dpackaging=jar
 
在IDE里面的pom.xml添加dependency，参照GitHub主页上的pom.xml中的dependency
可是出现报错了：
java.io.FileNotFoundException:
file:/Users/rocky/maven_repos/com/ggstar/ipdatabase/1.0/ipdatabase-1.0.jar!/ipRegion.xlsx (No such file or directory)
根据提示，咱们须要在项目源码中找到相应的文件拷进去IDE中的main/resources中！
 
 存储清洗后的数据：
按day分区来进行存储  partitionBy
存储模式：mode(SaveMode.Overwrite)  覆盖存储
coalesce：听说生产中常常用，是项目的调优势，控制文件的输出大小，个数
 
 
3、统计功能实现
 
功能实现一：统计TopN视频  
 
第一步：读取数据，read.format().load
第二步：
  1. 使用DataFrame API统计分析
  2. SQL API
最后把统计结果保存在MySQL数据库中
 
调优势：
读取parquet文件时，系统会默认解析各字段相应的数据类型，但有时候咱们就只须要它是String类型，须要在SparkSession定义时添加：
config("spark.sql.sources.partitionColumnTypeInference.enabled, "false"")
变成只会按照原类型读入
 
两种方法：
若使用DataFrame API来作：
用$号时候须要导入 隐式转换（这里是列名转换成列）！spark.implicits._
用到dataframe的count()函数要导入包：org.apache.spark.sql.functions._
 
若使用SQL API来作：
建立临时表createTempView
当心写SQL语句换行时不注意而忽略空格
 
实现代码：
  
//完成统计操做
object TopNStatJob {
  def main(args: Array[String]) {
    val spark = SparkSession.builder().appName("TopNStatJob")
      .config("spark.sql.sources.partitionColumnTypeInference.enabled", "false")
      .master("local[2]").getOrCreate()
    val accessDF = spark.read.format("parquet").load("/Users/kingheyleung/Downloads/data/clean/")
    dfCountTopNVideo(spark, accessDF)
    sqlCountTopNVideo(spark, accessDF)
    //accessDF.printSchema()
 
 
    spark.stop()
  }
  def dfCountTopNVideo(spark: SparkSession, accessDF: DataFrame): Unit = {
    /*
    * DF API
    * */
 
    //导入隐式转换, 留意$号的使用, 而且导入functions包，使agg聚合函数count可以使用，此处若不用$的话，就没法让times进行desc排序了
    import spark.implicits._
    val topNDF = accessDF.filter($"day" === "20170511" && $"cmsType" === "video")
      .groupBy("day", "cmsId").agg(count("cmsId").as("times")).orderBy($"times".desc)
    topNDF.show(false)
  }
  def sqlCountTopNVideo(spark: SparkSession, accessDF: DataFrame): Unit = {
    /*
    * SQL API
    * */
 
    //建立临时表access_view，注意换行时，很容易忽略掉空格
    accessDF.createTempView("access_view")
    val topNDF = spark.sql("select day, cmsId, count(1) as times from access_view " +
      "where day == '20170511' and cmsType == 'video' " +
      "group by day, cmsId " +
      "order by times desc")
    topNDF.show(false)
  }
}
 
 
在保存数据以前，须要写链接MySQL数据库的工具类，用到java.sql包
  1. 使用DriverManager，链接到mysql 3306
  2. 释放资源，connection和preparedstatement都要，注意处理异常
 
注意：若测试时拿不到链接，出现如下报错，那就是没有在dependency中添加或者选对mysql-connetor包
java.sql.SQLException: No suitable driver found for jdbc:mysql://localhost:3306/imooc_project?user=root&password=666
Error:scalac: error while loading <root>, Error accessing /Users/kingheyleung/.m2/repository/mysql/mysql-connector-java/5.0.8/mysql-connector-java-5.0.8.jar
 
我最终选的是5.1.40版本才对了
 
 
实现代码： 
/*
* 链接MySQL数据库
* 操做工具类
* */
object MySQLUtils {
  //得到链接
  def getConnection(): Unit = {
    DriverManager.getConnection("jdbc:mysql://localhost:3306/imooc_project?user=root&password=666")
  }
  //释放资源
  def release(connection: Connection, pstmt: PreparedStatement): Unit = {
    try {
      if (pstmt != null) {
        pstmt.close()
      }
    } catch {
      case e: Exception => e.printStackTrace()
    } finally {
      connection.close()
    }
  }
}

 

 
把统计数据保存到MySQL
  1. 在mysql中建立一张表，包含day，cms_Id，times三个字段（注意各自的数据类型，以及定义不容许为NULL，并把day和cms_Id做为PRI KEY）
  2. 建立模型类case class，三个输入参数，day、cms_Id，times
  3. 建立操做数据库DAO类，输入的参数是一个list，list装的是上面的模型类，目的是插入insert记录到数据库中，DAO中分如下几步：
  4. 首先，作jdbc链接的准备，建立connection和prepareStatement，把关闭链接也写好，用try catch finally抛出异常；
  5. 而后写sql语句，preparestatement须要赋值的地方用占位符放着；
  6. 进行对list遍历，把每一个对象都放进pstmt中
  7. 调优势！！！遍历前把自动提交关掉，遍历中把pstmt加入批处理中，遍历完后执行批处理操做！最后手工提交链接
 
实现代码： 
//课程访问次数实体类
case class VideoAccessStat(day: String, cmsId:Long, times: Long)
 
 
/*
* 各个维度统计的DAO操做
* */
object StatDAO {
  /*
  * 批量保存VideoAccessStat到数据库
  * */
  def insertDayAccessTopN(list: ListBuffer[VideoAccessStat]): Unit = {
 
    var connection: Connection = null  //jdbc的准备工做, 定义链接
    var pstmt: PreparedStatement = null
 
    try {
      connection = MySQLUtils.getConnection() //真正获取链接
 
      connection.setAutoCommit(false)   //为了实现批处理，要关掉默认的自动提交
 
      val sql = "insert into day_topn_video(day, cms_id, times) values (?, ?, ?)"  //占位符
      pstmt = connection.prepareStatement(sql)  //把SQL语句生成pstmt对象，后面才能够填充占位符中的数据
 
      for (ele <- list) {
        pstmt.setString(1, ele.day)
        pstmt.setLong(2, ele.cmsId)
        pstmt.setLong(3, ele.times)
 
        pstmt.addBatch()   //加入批处理
      }
 
      pstmt.execute()    //执行批量处理
      connection.commit()    //手工提交
 
    } catch {
      case e: Exception => e.printStackTrace()
    } finally {
      MySQLUtils.release(connection, pstmt)
    }
  }
}

 

 
为了对应以上的第3步，要把统计记录的DF生成一个个对象，放进list中：
  1. 建立模型类对应的list
  2. 对记录进行遍历，把记录的每一个字段当作参数，建立模型类对象
  3. 把每一个对象添加到list中
  4. 把list传进DAO类中 
 
如下代码添加到上面的TopNJob类里面中就能够把以前生成到的topDF的结果记录保存到MySQL当中了： 
try {
  topNDF.foreachPartition(partitionOfRecords => { //
    val list = new ListBuffer[VideoAccessStat]  //建立list来装统计记录
 
    //遍历每一条记录，取出来上面对应的三个字段day，cmsId，times
    partitionOfRecords.foreach(info => {
      val day = info.getAs[String]("day")   //后面的就是取出来的记录的每一个字段
      val cmsId = info.getAs[Long]("cmsId")
      val times = info.getAs[Long]("times")
 
      //每一次循环建立一个VideoAccessStat对象，添加一次进入list中
      list.append(VideoAccessStat(day, cmsId, times))
    })
    //把list传进DAO类
    StatDAO.insertDayAccessTopN(list)
  })
} catch {
  case e: Exception => e.printStackTrace()
}

 

 到此为止已经把项目需求一完成。
 
 
 功能实现二：按照城市来找出topN视频
 
在功能一的基础上，运用row_number函数来实现
 
具体的实现代码：
  
  //先计算访问次数，并按照day，cmsId，city分组
  val cityAccessTopNDF = accessDF.filter(accessDF.col("day") === "20170511" && accessDF.col("cmsType") === "video")
    .groupBy("day", "cmsId", "city").agg(count("cmsId").as("times"))
 
  //进行分地市排序，使用到row_number函数，生成一个排名，定义为time_rank, 而且取排名前3
  cityAccessTopNDF.select(
    cityAccessTopNDF.col("day"),
    cityAccessTopNDF.col("cmsId"),
    cityAccessTopNDF.col("times"),
    cityAccessTopNDF.col("city"),
    row_number().over(Window.partitionBy(cityAccessTopNDF.col("city"))
      .orderBy(cityAccessTopNDF.col("times").desc)
    ).as("times_rank")
  ).filter("times_rank <= 3").show(false)
}

 

 其余步骤和功能一同样，可是 插入Mysql的时候报错，缘由是MySQL不支持插入中文！！！！
首先能够在mysql命令行中用SET character来改：
SET character_set_client = utf8
 
可经过
show variables like 'character_set_%’;
查看当前的character编码设置
 
而后在jdbc链接时，加上：
useUnicode=true&characterEncoding=utf8
 
改了以后，虽然可以导入MySQL了，并且不出现乱码，但只有一部分数据，而且在控制台报错：
 
com.mysql.jdbc.PreparedStatement.fillSendPacket
com.mysql.jdbc.PreparedStatement.execute
 
后来把批处理删掉居然就能够把全部数据导入了：
 
pstmt.executeUpdate  //不使用批处理的pstmt插入
 
 
功能三：按流量来排序topN视频
和功能一几乎彻底同样，只不过计算流量总和时用的不是count函数而是要用sum函数
 
为了代码的复用性，防止生成重复的数据，在StatDAO定义删除的函数：
  
def deleteDayData(day: String) = {
 
  var connection: Connection = null
  var pstmt: PreparedStatement = null
  var tables = Array("day_topn_video",
    "day_city_topn_video",
    "traffic_topn_video"
  )
 
  try {
    connection = MySQLUtils.getConnection()
 
    for (table <- tables) {
      val deleteSql = s"delete from $table where day = ?”  //Scala特殊处理
      pstmt = connection.prepareStatement(deleteSql)
      pstmt.setString(1, table)
      pstmt.setString(2, day)
      pstmt.executeUpdate()
    }
  } catch {
    case e: Exception => e.printStackTrace()
  } finally {
    MySQLUtils.release(connection, pstmt)
  }
}

 

 
须要注意的是，table在pstmt中的特殊用法！！
 
 
后续会对以上内容进行可视化处理、跑在YARN上的修改、性能调优
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程