Spark SQL 分析 Nginx 访问日志

ゞ浴缸里的玫瑰 2022-04-04 11:15 283阅读 0赞

## 前言 ##

### 项目地址 ###

github: [Spark SQL 分析 Imooc 访问日志][Spark SQL _ Imooc]

### 环境说明 ###

*  Java版本：1.8
 *  Scala版本：2.11.12
 *  Hadoop版本：hadoop-2.6.0-cdh5.14.0
 *  spark版本：spark-2.3.1-bin-2.6.0-cdh5.14.0（自己编译）
 *  MySQL版本：5.7.22
 *  zeppelin版本：0.8

### 数据集 ###

**Imooc 访问日志文件**：access.20161111.log

**数据量**：一千多万条访问日志、5G多

**日志格式**：

60.165.39.1 - - [10/Nov/2016:00:01:53 +0800] "POST /course/ajaxmediauser HTTP/1.1" 200 54 "www.imooc.com" "http://www.imooc.com/code/1431" mid=1431&time=60 "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/38.0.2125.122 Safari/537.36 SE 2.X MetaSr 1.0" "-" 10.100.136.64:80 200 0.014 0.014
    14.145.74.175 - - [10/Nov/2016:00:01:53 +0800] "POST /course/ajaxmediauser/ HTTP/1.1" 200 54 "www.imooc.com" "http://www.imooc.com/video/678" mid=678&time=60&learn_time=551.5 "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/45.0.2454.101 Safari/537.36" "-" 10.100.136.64:80 200 0.014 0.014

**百度云盘下载地址**：链接：[https://pan.baidu.com/s/1VfOG14mGW4P4kj20nzKx8g][https_pan.baidu.com_s_1VfOG14mGW4P4kj20nzKx8g] 提取码：uwjg

**开发测试数据**：access.1w.log（10000条）

### 需求 ###

*  统计某天最受欢迎的TopN课程
 *  统计某天各个省市各自的 TopN 课程
 *  按照流量进行统计 TopN 课程
 *  某天最受欢迎的文章
 *  某天进行code最多的课程
 *  统计某天最勤奋的 IP
 *  欢迎补充…

### 统计结果可视化(zeppelin展示) ###

![167b6d3d71368a66_w_1861_h_606_f_png_s_71269][]

![167b6d42299204f6_w_1852_h_603_f_png_s_57034][]

![167b6d470fdc4e51_w_1856_h_602_f_png_s_53017][]

![167b6d4b68b1a130_w_1856_h_613_f_png_s_38529][]

## 开发步骤 ##

### 数据清洗 ###

根据需求，从日志中解析出我们需要的信息，譬如可能有：

1.  访问的系统属性： 操作系统、浏览器等等
2.  访问特征：url、referer (从哪个url跳转过来的)、页面上的停留时间等
3.  访问信息：session\_id、访问ip(访问城市)等

#### 主程序 ####

*  SparkStatFormatJob.scala 第一步，清洗出 ip, time, url, traffic
 *  SparkStatCleanJob.scala 第二步，最终清洗转换得到 url、cmsType、cmsId、traffic、ip、city、time、day
 *  AccessConvertUtil.scala 定义DataFrame schema，将日志信息转为对象，帮助RDD转为DataFrame
 *  DateUtils.scala 时间格式转换

#### 步骤 ####

1.  使用 Spark SQL 解析访问日志
2.  解析出课程编号类型
3.  根据IP解析出城市名称
4.  使用 Spark SQL 将访问时间按天进行分区输出

#### 关键代码 ####

清洗第一步

accessFile.map(line => {
          val splits = line.split(" ") // 按空格分割
          val ip = splits(0) // 第一个是IP
          // 原始日志的第三个和第四个字段拼接起来就是完整的访问时间： [10/Nov/2016:00:01:02 +0800] ==> yyyy-MM-dd HH:mm:ss
          val time = splits(3) + " " + splits(4)
          val url = splits(11).replaceAll("\"", "") // 第11个是 URL
          val traffic = splits(9) // 第9个是流量
          List(DateUtils.parse(time), url, traffic, ip)
        })
          // 过滤
          .filter(item => !"10.100.0.1".equals(item(3)))
          .filter(item => !"-".equals(item(1)))
          // 拼成一个对象 (DateUtils.parse(time), url, traffic, ip)
          .map(item => item(0) + "\t" + item(1) + "\t" + item(2) + "\t" + item(3))
          // 保存
          .saveAsTextFile(Constants.protocol + Constants.tempOut)

清洗第二步

val filterRDD = accessRDD.map(line => AccessConvertUtil.parseLog(line))
    val accessDF = spark.createDataFrame(filterRDD, AccessConvertUtil.struct)
    
    // 保存到 parquet
    accessDF.coalesce(1).write.format("parquet").mode(SaveMode.Overwrite).partitionBy("day").save(Constants.protocol + Constants.cleanedOut)

清洗结果样例

+--------------------------------------------+-------+-----+-------+---------------+----+-------------------+--------+
    |url                                         |cmsType|cmsId|traffic|ip             |city|time               |day     |
    +--------------------------------------------+-------+-----+-------+---------------+----+-------------------+--------+
    |http://www.imooc.com/code/1852              |code   |1852 |2345   |117.35.88.11   |陕西省 |2016-11-10 00:01:02|20161110|
    |http://www.imooc.com/learn/85/?src=360onebox|learn  |85   |14531  |115.34.187.133 |北京市 |2016-11-10 00:01:27|20161110|
    |http://www.imooc.com/course/list?c=fetool   |course |0    |66     |120.198.231.151|广东省 |2016-11-10 00:01:27|20161110|
    |http://www.imooc.com/code/10047             |code   |10047|54     |101.36.73.155  |北京市 |2016-11-10 00:01:27|20161110|
    +--------------------------------------------+-------+-----+-------+---------------+----+-------------------+--------+

### Spark SQL 统计 TopN ###

#### 主程序 ####

*  TopNStatJob.scala Spark SQL 统计主类
 *  StatDao.scala 将各个统计作业的统计结果写到数据库
 *  MySQLUtils.scala 管理 MySQL JDBC 连接

#### 关键代码 ####

/**
      * 统计某天各个省市各自的 TopN 课程
      */
    def cityAccessTopNStat(spark: SparkSession, accessDF: DataFrame, day: String): Unit = {
      import spark.implicits._
      val cityAccessTopNDF = accessDF.filter($"day" === day && $"cmsType" === "video" && $"cmsId" =!= "0")
        .groupBy("city", "day", "cmsId")
        .agg(count("cmsId").as("times"))
    
      // Window 函数在Spark SQL的使用: 窗口函数 row_number 的作用是根据表中字段进行分组，然后根据表中的字段排序，
      //  给组中的每条记录添加一个序号；且每组的序号都是从1开始，可利用它的这个特性进行分组取top-n
      val top3DF = cityAccessTopNDF.select(
        cityAccessTopNDF("day"), cityAccessTopNDF("city"),
        cityAccessTopNDF("cmsId"), cityAccessTopNDF("times"),
        row_number().over(Window.partitionBy(cityAccessTopNDF("city")) // 根据 city 分组，根据 times 降序排序
          .orderBy(cityAccessTopNDF("times").desc)).as("times_rank")
      ).filter("times_rank <= 3")
    
      // 保存到 MySQL，需创建结果表 day_video_city_access_topn_stat
      try {
        top3DF.foreachPartition(partition => {
          val list = new ListBuffer[DayCityVideoAccessStat]
    
          partition.foreach(item => {
            val day = item.getAs[String]("day")
            val cmsId = item.getAs[Long]("cmsId")
            val city = item.getAs[String]("city")
            val times = item.getAs[Long]("times")
            val timesRank = item.getAs[Int]("times_rank")
            list.append(DayCityVideoAccessStat(day, cmsId, city, times, timesRank))
          })
          StatDao.insertDayCityVideoAccessTopN(list)
        })
      } catch {
        case e: Exception => e.printStackTrace()
      }
    }

#### 统计结果样例 ####

+--------+-------+-----+-----+----------+
    |day     |city   |cmsId|times|times_rank|
    +--------+-------+-----+-----+----------+
    |20161110|北京市    |1309 |20   |1         |
    |20161110|北京市    |3369 |16   |2         |
    |20161110|北京市    |4018 |15   |3         |
    |20161110|辽宁省    |1336 |2    |1         |
    |20161110|辽宁省    |9028 |1    |2         |
    |20161110|辽宁省    |8141 |1    |3         |
    |20161110|浙江省    |3078 |19   |1         |
    |20161110|浙江省    |12552|16   |2         |
    |20161110|浙江省    |3237 |14   |3         |
    +--------+-------+-----+-----+----------+

## 项目开发说明 ##

1、 CDH相关的软件下载地址：[http://archive.cloudera.com/cdh5/cdh/5/，spark自己编译的，看官方文档即可][http_archive.cloudera.com_cdh5_cdh_5_spark]

2、IDEA需要安装Scala插件

3、 Windows上开发需解压Hadoop和spark源码，然后在环境变量中配置HADOOP\_HOME和SPARK\_HOME

4、 windows上需下载相应版本的 [winutils.exe ][winutils.exe] 文件放到 $HADOOP\_HOME/bin

5、 解析IP地址使用 [ipdatabase ][ipdatabase] ，三个步骤：

1）git clone https://github.com/wzhe06/ipdatabase.git
    2）编译下载的项目：mvn clean package -DskipTests
    3）安装jar包到自己的maven仓库
    mvn install:install-file -Dfile=/home/whirly/source/ipdatabase/target/ipdatabase-1.0-SNAPSHOT.jar -DgroupId=com.ggstar -DartifactId=ipdatabase -Dversion=1.0 -Dpackaging=jar

6、 需要创建相应的数据库和数据表，用于存储统计结果，具体的表结构见 imooc\_log.sql ，Navicat 导入MySQL即可，创建库表完毕后须修改 MySQLUtils.scala 中的配置信息

7、 zeppelin 可导入 最受欢迎的TopN课程.json 文件查看结果，也可以使用可视化方案，譬如echarts、highcharts、D3.js、HUE等等…

--------------------

> 更多内容可访问我的个人博客：[http://laijianfeng.org][http_laijianfeng.org]  
> 关注【小旋锋】微信公众号，及时接收博文推送  
> 原文地址：[Spark SQL 分析 Imooc 访问日志][Spark SQL _ Imooc 1]

![长按关注【小旋锋】微信公众号][167b6d1cd324168b_w_897_h_224_f_png_s_36467]

[Spark SQL _ Imooc]: https://github.com/whirlys/BigData-In-Practice/tree/master/ImoocLogAnalysis
[https_pan.baidu.com_s_1VfOG14mGW4P4kj20nzKx8g]: https://pan.baidu.com/s/1VfOG14mGW4P4kj20nzKx8g
[167b6d3d71368a66_w_1861_h_606_f_png_s_71269]: https://user-gold-cdn.xitu.io/2018/12/16/167b6d3d71368a66?w=1861&h=606&f=png&s=71269
[167b6d42299204f6_w_1852_h_603_f_png_s_57034]: https://user-gold-cdn.xitu.io/2018/12/16/167b6d42299204f6?w=1852&h=603&f=png&s=57034
[167b6d470fdc4e51_w_1856_h_602_f_png_s_53017]: https://user-gold-cdn.xitu.io/2018/12/16/167b6d470fdc4e51?w=1856&h=602&f=png&s=53017
[167b6d4b68b1a130_w_1856_h_613_f_png_s_38529]: https://user-gold-cdn.xitu.io/2018/12/16/167b6d4b68b1a130?w=1856&h=613&f=png&s=38529
[http_archive.cloudera.com_cdh5_cdh_5_spark]: http://archive.cloudera.com/cdh5/cdh/5/%EF%BC%8Cspark%E8%87%AA%E5%B7%B1%E7%BC%96%E8%AF%91%E7%9A%84%EF%BC%8C%E7%9C%8B%E5%AE%98%E6%96%B9%E6%96%87%E6%A1%A3%E5%8D%B3%E5%8F%AF
[winutils.exe]: https://github.com/steveloughran/winutils/tree/master/hadoop-2.6.0/bin
[ipdatabase]: https://github.com/wzhe06/ipdatabase
[http_laijianfeng.org]: http://laijianfeng.org
[Spark SQL _ Imooc 1]: http://laijianfeng.org/2018/12/Spark-SQL-%E5%88%86%E6%9E%90-Imooc-%E8%AE%BF%E9%97%AE%E6%97%A5%E5%BF%97/
[167b6d1cd324168b_w_897_h_224_f_png_s_36467]: https://user-gold-cdn.xitu.io/2018/12/16/167b6d1cd324168b?w=897&h=224&f=png&s=36467