利用spark分析慕课网千万条日志实战项目二

谁借莪１个温暖的怀抱￠ 2021-11-09 12:04 232阅读 0赞

Imooc网主站日志介绍：

1.  访问时间
2.  访问url
3.  访问过程耗费的流量
4.  访问ip地址

**第一步是数据清洗：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70][]

一般的日志处理方式，我们是需要进行分区的，

按照日志中的访问时间进行相应的分区，比如：d,h,m5(每5分钟一个分区)

输入：访问时间、访问URL、耗费的流量、访问IP地址信息

输出：URL、cmsType(video/article)、cmsId(编号)、流量、ip、城市信息、访问时间、天

**利用spark分析慕课网5G日志实战项目**一：[https://blog.csdn.net/qq\_41479464/article/details/98211937][https_blog.csdn.net_qq_41479464_article_details_98211937]

*  **项目需求：**
 *  **课程视频及所有资料和代码以及镜像链接：**
    
    [https://download.csdn.net/download/qq\_41479464/11502023][https_download.csdn.net_download_qq_41479464_11502023]

实战一的链接包括基本操作：[https://blog.csdn.net/qq\_41479464/article/details/98212216][https_blog.csdn.net_qq_41479464_article_details_98212216]

整体三个代码：如箭头所指

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 1][]

需求二：按照地理位置（跟据IP地址提取出城市信息）（地市）（窗口函数在SparkSql中的使用）统计imooc的主站最受欢迎的课程/或手记的TopN访问次数

数据类型：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 2][]

定义一个scala类：DayCityVideoAccessStat

package log
    
    case class DayCityVideoAccessStat(day:String, cmsId:Long, city:String,times:Long,timesRank:Int)

定义一个StatDao层留把数据存放进数据库中：

package log
    
    import java.sql.{Connection, PreparedStatement}
    
    import scala.collection.mutable.ListBuffer
    
    /**
      * 各个维度统计的Dao操作
      */
    object StatDao {
      /**
        * 批量保存到DayVideoAccessStat数据库中
        *
        */
    
      /**
        * 批量保存DayVideoAccessStat到数据库
        */
      def insertDayVideoAccessTopN(list: ListBuffer[DayVideoAccessStat]): Unit = {
    
        var connection: Connection = null
        var pstmt: PreparedStatement = null
    
        try {
          connection = MysqlUtils.getConnection()
    
          connection.setAutoCommit(false) //设置手动提交
    
          val sql = "insert into day_video_access_topn_stat(day,cms_id,times) values (?,?,?) "
          pstmt = connection.prepareStatement(sql)
    
          for (ele <- list) {
            pstmt.setString(1, ele.day)
            pstmt.setLong(2, ele.cmsId)
            pstmt.setLong(3, ele.times)
    
            pstmt.addBatch()
          }
    
          pstmt.executeBatch() // 执行批量处理
          connection.commit() //手工提交
        } catch {
          case e: Exception => e.printStackTrace()
        } finally {
          MysqlUtils.release(connection,pstmt)
        }
      }
    }

定义作业：

TopNStatJob2：类

package log
    import org.apache.spark.sql.expressions.Window
    import org.apache.spark.sql.{DataFrame, SparkSession}
    import org.apache.spark.sql.functions._
    
    import scala.collection.mutable.ListBuffer
    object TopNStatJob2 {
      /**
        * TopN统计spark作业
        */
        def main(args: Array[String]) {
          val spark = SparkSession.builder().appName("TopNStatJob2")
            .config("spark.sql.sources.partitionColumnTypeInference.enabled", false)
            .master("local[2]").getOrCreate()
          val accessDF = spark.read.format("parquet").load("D:\\BigDataTest\\data\\clean")
          accessDF.printSchema()
          accessDF.show()
    
          /**
            * 按照地市统计topN的课程
            *
            */
          cityAccessTopNStat(spark,accessDF)
    
    
          spark.stop()
        }
        /**
          * 按照地市进行统计
          * accessDF
          */
      /**
        * 按照地市进行统计TopN课程
        */
      def cityAccessTopNStat(spark: SparkSession, accessDF:DataFrame): Unit = {
        import spark.implicits._
    //    //过滤出当天的视频的个数
        val cityAccessTopNDF = accessDF.filter($"day"==="20170511" && $"cmsType"==="video")
          .groupBy("day","city","cmsId")
          .agg(count("cmsId").as("times"))
    
        //cityAccessTopNDF.show(false)
    
        //Window函数在Spark SQL的使用
    
        val top3DF = cityAccessTopNDF.select(
          cityAccessTopNDF("day"),
          cityAccessTopNDF("city"),
          cityAccessTopNDF("cmsId"),
          cityAccessTopNDF("times"),
          row_number().over(Window.partitionBy(cityAccessTopNDF("city"))
            .orderBy(cityAccessTopNDF("times").desc)
          ).as("times_rank")
        ).filter("times_rank <=3")//.show(false)  //Top3
        /**
          * 将统计结果写入到MySQL中
          */
        try {
          top3DF.foreachPartition(partitionOfRecords => {
            val list = new ListBuffer[DayCityVideoAccessStat]
    
            partitionOfRecords.foreach(info => {
              val day = info.getAs[String]("day")
              val cmsId = info.getAs[Long]("cmsId")
              val city = info.getAs[String]("city")
              val times = info.getAs[Long]("times")
              val timesRank = info.getAs[Int]("times_rank")
              list.append(DayCityVideoAccessStat(day,cmsId,city,times,timesRank))
            })
    
            StatDao2.insertDayCityVideoAccessTopN(list)
          })
        } catch {
          case e:Exception => e.printStackTrace()
        }
      }
    }

最后得到的某天的数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 3][]

数据可视化的过程：在项目实战一中有基础可视化的教程的介绍：这里面我用的Zeppelin

实战一的链接包括基本操作：[https://blog.csdn.net/qq\_41479464/article/details/98212216][https_blog.csdn.net_qq_41479464_article_details_98212216]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 4][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 5][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 6][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 7][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70]: /images/20211109/bea75b8b8ac24bc4b99c159219dd7ffc.png
[https_blog.csdn.net_qq_41479464_article_details_98211937]: https://blog.csdn.net/qq_41479464/article/details/98211937
[https_download.csdn.net_download_qq_41479464_11502023]: https://download.csdn.net/download/qq_41479464/11502023
[https_blog.csdn.net_qq_41479464_article_details_98212216]: https://blog.csdn.net/qq_41479464/article/details/98212216
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 1]: /images/20211109/fa401be35c6f429898584781065283ae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 2]: /images/20211109/7279173a4db94739ba9c8e674f3fc6d9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 3]: /images/20211109/e00556f0a37a49ae82e2059eb0825ce8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 4]: /images/20211109/533c0fa531514bd398b7ea3cc4ca3114.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 5]: /images/20211109/3acfa8740db046d39cf5cb7ec69db13c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 6]: /images/20211109/e6c2ce9e6441423eabc3a40a17380d8f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 7]: /images/20211109/b5f55246c1644865a483c0a6ae3b6207.png