0
0

Spark流处理日志+SSM前端展示（详细）

1.前言

最近在学习Spark Streaming，对流数据日志处理有了一些了解，加上学校近期开展了SSM框架的课程。一上头，就想把这东西给整和一下，源码我整合在Gitee上了，可自行取。

此项属于对前后端+Spark数据处理的一个整合，难度并不大，属于对Spark学习的横向扩展，适用于Spark Streaming的初学者。

考虑篇幅过长,先上源码:Spark流处理日志

2.实战内容

  1. python编写脚本定时生成伪随机日志
  2. SparkStream获取数据，将数据清洗过滤非法数据，然后分析日志中用户的访问量，统计访问课程数；访问搜索引擎，统计不同搜索引擎的点击量 - 将Spark Streaming处理的结果初步存入Mysql（可以缓存到Hbase中，取数可以更快）- Spark获取Mysql数据，统计课程和搜索引擎的点击量，存入Mysql
  3. 前端使用Spring, Spring Mvc, MyBatis展示数据
  4. 考虑到图像运用Echarts,所以使用Ajax异步传输数据给jsp

Spark工作流程图如下：

3.实战框架

实战工具：Intellij IDEA2021专业版( 因为社区版没有Tomcat，也没有具体试验过插件Smart Tomcat，其他功能社区版均能实现 )， PyCharm社区版，JDK版本1.8，Scala版本2.13

架构和涉及的技术栈：

  • Python-3.9
  • Spark-3.2.1
  • Mybatis-3.5.9
  • Spring Mvc-5.3.16
  • Spring-5.3.16
  • Echarts

4.项目实战

注意：本项目将Scala和Java板块合并在同一个工程中，所以只需要创建一个工程即可

1.环境搭建：

1）创建Maven工程，引入依赖pom.xml(源文件是完整的)：

  2.     <dependencies>
  3.         <dependency>
  4.             <groupId>mysql</groupId>
  5.             <artifactId>mysql-connector-java</artifactId>
  6.             <version>${mysql.version}</version>
  7.         </dependency>
  9.         <dependency>
  10.             <groupId>org.mybatis</groupId>
  11.             <artifactId>mybatis</artifactId>
  12.             <version>${mybatis.version}</version>
  13.         </dependency>
  15.         <!-- https://mvnrepository.com/artifact/org.mybatis/mybatis-spring -->
  16.         <dependency>
  17.             <groupId>org.mybatis</groupId>
  18.             <artifactId>mybatis-spring</artifactId>
  19.             <version>2.0.7</version>
  20.         </dependency>
  22.         <!-- https://mvnrepository.com/artifact/commons-logging/commons-logging -->
  23.         <dependency>
  24.             <groupId>commons-logging</groupId>
  25.             <artifactId>commons-logging</artifactId>
  26.             <version>1.2</version>
  27.         </dependency>
  29.         <!-- https://mvnrepository.com/artifact/org.springframework/spring-expression -->
  30.         <dependency>
  31.             <groupId>org.springframework</groupId>
  32.             <artifactId>spring-expression</artifactId>
  33.             <version>${spring.version}</version>
  34.         </dependency>
  36.         <dependency>
  37.             <groupId>org.springframework</groupId>
  38.             <artifactId>spring-context-support</artifactId>
  39.             <version>${spring.version}</version>
  40.         </dependency>
  42.         <dependency>
  43.             <groupId>org.springframework</groupId>
  44.             <artifactId>spring-jdbc</artifactId>
  45.             <version>${spring.version}</version>
  46.         </dependency>
  48.         <dependency>
  49.             <groupId>org.springframework</groupId>
  50.             <artifactId>spring-webmvc</artifactId>
  51.             <version>${spring_mvc.vserion}</version>
  52.         </dependency>
  54.         <!-- https://mvnrepository.com/artifact/org.springframework/spring-core -->
  55.         <dependency>
  56.             <groupId>org.springframework</groupId>
  57.             <artifactId>spring-core</artifactId>
  58.             <version>${spring.version}</version>
  59.         </dependency>
  61.         <!-- https://mvnrepository.com/artifact/org.springframework/spring-context -->
  62.         <dependency>
  63.             <groupId>org.springframework</groupId>
  64.             <artifactId>spring-context</artifactId>
  65.             <version>${spring.version}</version>
  66.         </dependency>
  68.         <dependency>
  69.             <groupId>net.sf.json-lib</groupId>
  70.             <artifactId>json-lib</artifactId>
  71.             <version>2.4</version>
  72.             <classifier>jdk15</classifier>
  73.         </dependency>
  75.         <dependency>
  76.             <groupId>com.fasterxml.jackson.core</groupId>
  77.             <artifactId>jackson-databind</artifactId>
  78.             <version>2.12.2</version>
  79.         </dependency>
  81.         <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
  82.         <dependency>
  83.             <groupId>org.apache.spark</groupId>
  84.             <artifactId>spark-core_2.13</artifactId>
  85.             <version>${spark.version}</version>
  86.         </dependency>
  88.         <!-- https://mvnrepository.com/artifact/junit/junit -->
  89.         <dependency>
  90.             <groupId>junit</groupId>
  91.             <artifactId>junit</artifactId>
  92.             <version>4.13.2</version>
  93.             <scope>test</scope>
  94.         </dependency>
  96.         <!-- https://mvnrepository.com/artifact/org.scala-lang/scala-library -->
  97.         <dependency>
  98.             <groupId>org.scala-lang</groupId>
  99.             <artifactId>scala-library</artifactId>
  100.             <version>2.13.8</version>
  101.         </dependency>
  103.         <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
  104.         <dependency>
  105.             <groupId>org.apache.spark</groupId>
  106.             <artifactId>spark-sql_2.13</artifactId>
  107.             <version>${spark.version}</version>
  108.         </dependency>
  110.         <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming -->
  111.         <dependency>
  112.             <groupId>org.apache.spark</groupId>
  113.             <artifactId>spark-streaming_2.13</artifactId>
  114.             <version>${spark.version}</version>
  115.         </dependency>
  117.     </dependencies>

2）下载Scala插件,再右键项目名添加框架支持，导入Scala模块

2.后端开发

1.在PyCharm创建python脚本，生成随机日志

一次脚本的完整运行，将会一次性产生2000条日志，以追加的方式平摊在文档中。根据脚本的编码，产生一条正确完整的Web日志的概率仅为20%，即一次运行，只有400条左右的日志会存入数据库中，详情可查看步骤2）

  1. import random
  2. import time
  4. #  创建url访问数组class
  5. url_paths = [
  6.     "class/112.html",
  7.     "class/128.html",
  8.     "class/145.html",
  9.     "class/146.html",
  10.     "class/500.html",
  11.     "class/250.html",
  12.     "class/131.html",
  13.     "learn/821",
  14.     "learn/823",
  15.     "learn/987",
  16.     "learn/500",
  17.     "course/list"
  18. ]
  20. #  创建ip数组，随机选择4个数字作为ip如132.168.30.87
  21. ip_slices = [132, 156, 124, 10, 29, 167, 143, 187, 30, 46, 55, 63, 72, 87, 98, 168, 192, 134, 111, 54, 64, 110, 43, 127, 12, 53, 99]
  23. #  搜索引擎访问数组
  24. # {query}
  25. # 代表搜索关键字
  26. http_referers = [
  27.     "http://www.baidu.com/s?wd={query}",
  28.     "https://www.sogou.com/web?query={query}",
  29.     "http://cn.bing.com/search?q={query}",
  30.     "https://search.yahoo.com/search?p={query}",
  31. ]
  33. #  搜索关键字数组
  34. search_keyword = [
  35.     "Spark SQL",
  36.     "Hadoop",
  37.     "Storm",
  38.     "Spark Streaming",
  39.     "10 HoursCrash",
  40.     "Blink",
  41.     "SpringBoot",
  42.     "Linux",
  43.     "Vue.js",
  44.     "Hive",
  45.     "Flink",
  46. ]
  48. #  状态码数组
  49. status_codes = ["200", "504", "404", "403"]
  51. #  随机选择一个url
  52. def sample_url():
  53.     return random.sample(url_paths, 1)[0]
  55. #  随机组合一个ip
  56. def sample_ip():
  57.     slice = random.sample(ip_slices, 4)
  58.     return ".".join([str(item) for item in slice])
  60. #  随机产生一个搜索url
  61. def sample_referer():
  62.     # // 0.2的概率会产生非法url，用于模拟非法的用户日志
  63.     if random.uniform(0, 1) > 0.8:
  64.         return "-"
  66.     refer_str = random.sample(http_referers, 1)
  67.     query_str = random.sample(search_keyword, 1)
  68.     return refer_str[0].format(query=query_str[0])
  70. #  随机产生一个数组
  71. def sample_status_code():
  72.     return random.sample(status_codes, 1)[0]
  74. def generate_log(count=10, index=10):
  75.     # 获取本机时间并将其作为访问时间写进访问日志中
  76.     time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
  77.     indexes = str(index)
  78.     file = "./data/OfflineWareHouse/随即日志生成器" + indexes + ".log"
  79.     # 存储日志的目标文件(换成你自己的) w覆盖 a追加
  80.     f = open(file, "a")
  82.     # 组合用户日志
  83.     while count >= 1:
  84.         query_log = "{ip}\t{local_time}\t\"GET /{url} HTTP/1.1\"\t{status_code}\t{referer}".format(url=sample_url(),
  85.                                                                                                    ip=sample_ip(),
  86.                                                                                                    referer=sample_referer(),
  87.                                                                                                    status_code=sample_status_code(),
  88.                                                                                                    local_time=time_str)
  89.         f.write(query_log + "\n")
  91.         count = count - 1
  92.     f.close()
  94. # 执行main，每次产生100*20条用户日志
  95. counts = 20
  96. while counts:
  97.     sleepTime = random.randint(6, 12)
  98.     time.sleep(sleepTime)
  99.     print("第%d个Web日志新增%d条数据" % (counts, 100))
  100.     generate_log(100, counts)
  101.     counts = counts - 1

2.SparkStreaming数据的初步清洗

Spark Streaming可整合多种输入数据源，如Kafka，Flume，HDFS，甚至是普通的TCP套接字流，本次是对本地模式中的固定目录进行监听，一旦发现有新的文件内容生成，Spark Streaming就会自动把文件内容读取出来，使用自定义的处理逻辑处理.

①.在主函数创建对应的工程包scala并标记为"源目录"：

②.在数据清洗前，需要了解Web日志的规格设置，本日志数据与数据之间是通过"\t"也就是Tab键位分隔开的，下面是一条常规的Web日志，其规格如下,我们需要将method和url拆开，获取单独的GET，yahoo，Spark Streaming字段

  1. userIp = 168.46.55.10
  2. local_time = 2022-04-03 19:03:52
  3. method = "GET /class/146.html HTTP/1.1"    
  4. status = 404    
  5. url = https://search.yahoo.com/search?p=Spark Streaming

③.Spark Streaming的代码如下：

SparkStreaming的数据抽象为DStream，不同的数据抽象有不同的函数方法，为了完成需求，、整个日志的RDD格式走向为DStream[String]->RDD[String]->DataFrame

  1. package ssm
  3. import org.apache.spark.sql.{SQLContext}
  4. import java.util.Properties
  5. import org.apache.spark.SparkConf
  6. import org.apache.spark.streaming.{Seconds, StreamingContext}
  8. object WareHouseStreaming {
  9.   def main(args:Array[String]): Unit ={
  10.     val sparkConf = new SparkConf().setAppName("WareHouseStreaming").setMaster("local")
  11.     val prop = new Properties()
  12.     prop.put("user", "root")
  13.     prop.put("password", "******")
  14.     prop.put("driver","com.mysql.jdbc.Driver")
  15.     val scc = new StreamingContext(sparkConf, Seconds(10))
  17.     val file = "file:///C:/Users/Lenovo/Desktop/Working/Python/data/OfflineWareHouse/"
  18.     val lineDStream = scc.textFileStream(file)
  20.     val lines = lineDStream.flatMap(_.split("\n"))
  21.     val line = lines.map(_.replaceAll("\"","")).map(_.split("\t"))
  22.     line.foreachRDD(rdd=>{
  23.       val sqlContext = SQLContext.getOrCreate(rdd.sparkContext)
  24.       import sqlContext.implicits._
  25.       rdd
  26.         .map(x=>(x(0).trim,x(1).trim,x(2).trim,x(3).trim,x(4).trim))
  27.         .toDF("userIp","local_time","method","status","url")
  28.         .registerTempTable("speedtable")
  29.       val dataFrame = sqlContext.sql("select * from  speedtable ")
  30.       //dataFrame.show(5)
  31.       //println(dataFrame.count())
  33.       //过滤状态码非200字段，过滤非法url字段
  34.       //状态码为200的概率为1/4,产生合法url字段的概率为4/5,即一条正确的日志概率为1/5
  35.       val d1 = dataFrame
  36.         .filter(x => x(3) == "200")
  37.         .filter(x => x(4) != "-")
  39.       val dfDetail = d1.map( row =>{
  40.         val methods = row.getAs[String]("method").split(" ")
  41.         val urls = row.getAs[String]("url").split("\\?")
  42.         val SE = urls(0)
  43.         val names = urls(1)
  44.           .split("=")
  45.         val name = names(1)
  46.         var map = Map("params" -> "null")
  47.         val method = methods(0)
  48.         val url = methods(1)
  49.         val agreement = methods(2)
  50.         (
  51.           row.getAs[String]("userIp"),
  52.           row.getAs[String]("local_time"),
  53.           map.getOrElse("method",method),
  54.           map.getOrElse("url",url),
  55.           map.getOrElse("agreement",agreement),
  56.           row.getAs[String]("status"),
  57.           map.getOrElse("SE",SE),
  58.           map.getOrElse("name",name)
  59.         )
  60.       }).toDF("userIp","local_time","method","url","agreement","status","SE","name")
  61.       dfDetail.show(5)
  63.       val url = "jdbc:mysql://localhost:3306/python_db"
  64.       println("开始写入数据库")
  65.       dfDetail.write.mode("Append").jdbc(url,"warehouse",prop)
  66.       println("完成写入数据库,新增"+dfDetail.count()+"条数据")
  67.     })
  69.     scc.start()
  70.     scc.awaitTermination()
  71.   }
  72. }

** 3.统计课程点击量和搜索引擎的访问量**

相比较采用流数据处理问题，采用离线数据处理问题的方式简单快捷，程序可以独立运行，方便维护，缺点是计算量比流式计算模式大得多，因为会重复计算数据。但前端页面的刷新时间为10s，数据的时效性并不明显，10s对离线计算时间足以

课程点击量代码：

  1. package ssm
  3. import org.apache.spark.sql.{Row, SparkSession}
  4. import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
  6. import java.util.Properties
  8. object NameCount {
  9.   def main(args:Array[String]): Unit ={
  10.     val prop = new Properties()
  11.     prop.put("user", "root")
  12.     prop.put("password", "******")
  13.     prop.put("driver", "com.mysql.jdbc.Driver")
  14.     val url = "jdbc:mysql://localhost:3306/python_db"
  15.     while(true){
  16.       Thread.sleep(10000)
  17.       println("开始更新课程点击量")
  18.       lessCount(prop,url)
  19.     }
  20.   }
  21.   def lessCount(prop:Properties,url:String): Unit = {
  22.     val spark = SparkSession.builder().appName("NameCount").master("local").getOrCreate()
  23.     val sc = spark.sparkContext
  24. /**
  25. 整体流程的数据类型为Mysql->
  26. DataFrame->{ RDD[Row]->RDD[String] }
  27. ->replace归并课程类型
  28. ->RDD[String,Int]统计点击量
  29. ->RDD[Row(String,Integer)]
  30. ->DataFrame->Mysql
  31.  */
  33.     import spark.implicits._
  34.     val dataFrame = spark.read.jdbc(url, "warehouse", prop).select("name")
  36. //    DataFrame->{ RDD[Row]->RDD[String] }
  37.     val dataRDD = dataFrame.rdd.map(x=>x.mkString(""))
  38. //    dataRDD.foreach(println)
  40. //    ->replace归并课程类型
  41.     val word = dataRDD
  42.       .map(_.replaceAll("10 HoursCrash","Hadoop"))
  43.       .map(_.replaceAll("Spark Streaming","Spark"))
  44.       .map(_.replaceAll("Hive","Hadoop"))
  45.       .map(_.replaceAll("Spark SQL","Spark"))
  46.       .map(_.replaceAll("Blink","Flink"))
  47.     //word.foreach(println)
  49. //    ->RDD[String,Int]统计点击量
  50.     val wordCount = word.map((_,1))
  51.       .reduceByKey(_+_)
  52.       .sortBy(_._2,false)
  53.     //wordCount.foreach(println)
  55. //    ->RDD[Row(String,Integer)]
  56.     val dF = wordCount.map(x=>{Row(x._1,x._2)})
  57.     //dF.foreach(println)
  59. //    ->DataFrame->Mysql
  60.     val schema = StructType(Array(
  61.       StructField("lesson",StringType),
  62.       StructField("counts",IntegerType)
  63.     ))
  64.     val orgDF = spark.createDataFrame(dF, schema)
  65.     orgDF.show()
  67.     println("开始写入数据库")
  68.     orgDF.write.mode("overwrite").jdbc(url,"lesson_counts",prop)
  69.     println("完成写入数据库,因为报错会中断进程")
  70.     sc.stop()
  71.   }
  72. }

** 搜索引擎访问量统计代码如下：**

  1. package ssm
  3. import org.apache.spark.sql.{Row, SparkSession}
  4. import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
  6. import java.util.Properties
  8. object UrlCount {
  9.   def main(args:Array[String]): Unit ={
  10.     val prop = new Properties()
  11.     prop.put("user", "root")
  12.     prop.put("password", "******")
  13.     prop.put("driver", "com.mysql.jdbc.Driver")
  14.     val url = "jdbc:mysql://localhost:3306/python_db"
  15.     while(true) {
  16.       Thread.sleep(10000)
  17.       println("开始更新搜索点击量")
  18.       SECounts(prop, url)
  19.     }
  20.   }
  21.   def SECounts(prop:Properties,url:String): Unit ={
  22.     val spark = SparkSession.builder().appName("urlCount").master("local").getOrCreate()
  23.     val sc = spark.sparkContext
  25.     val SEFrame = spark.read.jdbc(url, "warehouse", prop).select("SE")
  26. //    dataFrame.show(5)
  28.     val SERdd = SEFrame.rdd.map(_.mkString(""))
  30.     val SE = SERdd
  31.       .map(_.split("[.]"))
  32.       .map(x=>x(1))
  33.       .map(_.replaceAll("baidu","百度"))
  34.       .map(_.replaceAll("sogou","搜狗"))
  35.       .map(_.replaceAll("bing","必应"))
  36.       .map(_.replaceAll("yahoo","雅虎"))
  38.     val seCount = SE.map((_,1))
  39.       .reduceByKey(_+_)
  40.       .sortBy(_._1)
  41.       .map(x=>{Row(x._1,x._2)})
  42.     seCount.foreach(println)
  44.     val schema = StructType(Array(
  45.       StructField("Se",StringType),
  46.       StructField("counts",IntegerType)
  47.     ))
  48.     val detailDF = spark.createDataFrame(seCount,schema)
  49.     println("正在存入数据库")
  50.     detailDF.write.mode("Overwrite").jdbc(url,"se_counts",prop)
  51.     println("数据库存入完成")
  53.     sc.stop()
  54.   }
  55. }

4.检验步骤

运行SparkStreaming代码挂后台

运行UrlCount和NameCount挂后台

运行Python脚本挂后台

前往Spark的窗口查看数据是否成功收集

Spark Streaming成功样例：

** NameCount成功样例：**

** urlCount成功样例：**

** 5.注意事项：**

①.Spark Streaming处理日志需要花费的时间最好不要超过一次检测（目前是10s）的时间，当Python一次性生成的数据量过大（目前是100）导致上述问题发生时，Spark的计算会出现严重问题，甚至可能会宕机。合理分配Spark检测时间和数据量很重要，这属于经验，具体要看后续其他业务的数据量的大小来调优。

---有一次作死，让python每5s产生1500条日志（总计一次3万），Spark检测时间也是5s，结果一次存入数据库的数据量也在3万上下（理论是6000）。后续发现当第一批数据读入还未计算完，第二批就进来了，结果就是第一批数据夹杂着第二批的数据一并存入数据库中，而当我意识到这个问题的时候，数据库中的数据量已经超过20多万了，我一气之下删库了.....

至此Spark Streaming的开发已经完成了

3.前端开发（IDEA专业版）

创建工程文件搭建SSM框架

1.配置mapper和pojo目录文件

工程文件显示如下，其中SE为搜索引擎的类，Lesson为课程点击量的类

Lesson.java (SE.java同理，可以直接去发的源码查看)

  1. package cn.pojo;
  3. public class Lesson {
  4.     private String lesson = null;
  5.     private Integer lesson_count = null;
  7.     public String getLesson() {return lesson;}
  8.     public void setLesson(String lesson) {this.lesson = lesson;}
  9.     public Integer getLesson_count() {return lesson_count;}
  10.     public void setLesson_count(Integer lesson_count) {this.lesson_count = lesson_count;}
  12.     @Override
  13.     public String toString() {
  14.         return "Lesson{" +
  15.                 " lesson= " + lesson +
  16.                 " ,lesson_count= " + lesson_count +
  17.                 '}';
  18.     }
  19. }

接口LessonMapper.java (SEMapper.java同理)

  1. package cn.mapper;
  3. import org.apache.ibatis.annotations.Param;
  4. import cn.pojo.Lesson;
  6. import java.util.List;
  8. public interface LessonMapper {
  9.     Integer selectLessonCount(Lesson user);
  10.     List<Lesson> selectAll();
  11. }

接口的实现LessonMapper.xml (SE同理)

  1. <?xml version="1.0" encoding="UTF-8" ?>
  2. <!DOCTYPE mapper
  3.         PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
  4.         "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
  5. <mapper namespace="cn.mapper.LessonMapper">
  6.     <select id="selectLessonCount" resultType="int">
  7.         select counts from lesson_counts where lesson = CONCAT(#{lesson})
  8.     </select>
  9.     <!-- 左边为Lesson设置的参数名字，右边为MySQL表格的参数名字,如果是联表查询都可以resultMap为这个 -->
  10.     <resultMap type="cn.pojo.Lesson" id="userLesson">
  11.         <result property="lesson" column="lesson"/>
  12.         <result property="lesson_count" column="counts"/>
  13.     </resultMap>
  14.     <select id="selectAll" resultType="cn.pojo.Lesson" resultMap="userLesson">
  15.         select * from lesson_counts
  16.     </select>
  17. </mapper>

2.util配置

在util下创建MybatisUtils.java用于Mybatis连接Mysql

  1. package cn.util;
  3. import org.apache.ibatis.io.Resources;
  4. import org.apache.ibatis.session.SqlSession;
  5. import org.apache.ibatis.session.SqlSessionFactory;
  6. import org.apache.ibatis.session.SqlSessionFactoryBuilder;
  8. import java.io.IOException;
  9. import java.io.InputStream;
  11. public class MybatisUtils {
  12.     private static SqlSessionFactory sqlSessionFactory;
  13.     static {
  14.         try {
  15.             String resource = "mybatis-config.xml";
  16.             InputStream inputStream = Resources.getResourceAsStream(resource);
  17.             sqlSessionFactory = new SqlSessionFactoryBuilder().build(inputStream);
  18.         } catch (IOException e) {
  19.             e.printStackTrace();
  20.         }
  21.     }
  23.     //获取SqlSession连接
  24.     public static SqlSession getSession(){
  25.         return sqlSessionFactory.openSession();
  26.     }
  27. }

3.配置dao层和service层代码

DAO层代码

接口CourseClickDao.java （SE同理）

  1. package cn.dao;
  3. import cn.pojo.Lesson;
  5. import java.util.List;
  7. public interface CourseClickDao {
  8.     List<Lesson> selectLessonAll(Lesson user);
  9. }

接口CourseClickDao.java的实现CourseDaoImpl.java

  1. package cn.dao.Impl;
  3. import cn.dao.CourseClickDao;
  4. import cn.pojo.Lesson;
  5. import org.mybatis.spring.SqlSessionTemplate;
  7. import java.util.List;
  9. public class CourseDaoImpl implements CourseClickDao{
  10.     public SqlSessionTemplate session;
  11.     public SqlSessionTemplate getSession(){return session;}
  12.     public void setSession(SqlSessionTemplate session){this.session = session;}
  14.     @Override
  15.     public List selectLessonAll(Lesson user){
  16.         return session.selectList("cn.mapper.LessonMapper.selectAll");
  17.     }
  18. }

4.resources配置

Spring配置文件beans.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans"
  3.        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.        xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"
  5.        xmlns:aop="http://www.springframework.org/schema/aop"
  6.        xmlns:p="http://www.springframework.org/schema/p"
  7.        xsi:schemaLocation="http://www.springframework.org/schema/beans
  8.                                 http://www.springframework.org/schema/beans/spring-beans.xsd
  9.                                   http://www.springframework.org/schema/context
  10.                                   http://www.springframework.org/schema/context/spring-context-4.3.xsd
  11.                                   http://www.springframework.org/schema/tx
  12.                                   http://www.springframework.org/schema/tx/spring-tx-4.3.xsd
  13.                                   http://www.springframework.org/schema/aop
  14.                                   http://www.springframework.org/schema/aop/spring-aop-4.3.xsd"
  15.        default-autowire="byType">
  16.     <!-- 配置数据源 换成你的数据库url-->
  17.     <bean id="dataSource" class="org.apache.commons.dbcp2.BasicDataSource"
  18.           destroy-method="close">
  19.         <property name="driverClassName" value="com.mysql.cj.jdbc.Driver"/>
  20.         <property name="url" value="jdbc:mysql://localhost:3306/python_db"/>
  21.         <property name="username" value="root"/>
  22.         <property name="password" value="******"/>
  23.     </bean>
  25.     <!-- 配置SessionFactory -->
  26.     <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
  27.         <property name="dataSource" ref="dataSource"/>
  28.         <property name="configLocation" value="classpath:mybatis-config.xml"/>
  29.     </bean>
  31.     <!--内容可以不改动,配置sqlSessionTemplate    -->
  32.     <bean id="sqlSessionTemplate" class="org.mybatis.spring.SqlSessionTemplate">
  33.         <constructor-arg name="sqlSessionFactory" ref="sqlSessionFactory"/>
  34.     </bean>
  36.     <bean id="clickDao" class="cn.dao.Impl.CourseDaoImpl">
  37.         <property name="session" ref="sqlSessionTemplate"/>
  38.     </bean>
  39.     <bean id="searchDao" class="cn.dao.Impl.SearchDaoImpl">
  40.         <property name="session" ref="sqlSessionTemplate"/>
  41.     </bean>
  43.     <bean id="clickService" class="cn.service.Impl.CourseServiceImpl">
  44.         <property name="dao" ref = "clickDao"/>
  45.     </bean>
  46.     <bean id="searchService" class="cn.service.Impl.SearchServiceImpl">
  47.         <property name="dao" ref = "searchDao"/>
  48.     </bean>
  50.     <bean id="Lesson" class="cn.pojo.Lesson"
  51.     p:lesson="Hadoop" p:lesson_count="1"/>
  53.     <bean id="SE" class="cn.pojo.SE"
  54.     p:lesson="百度" p:lesson_count="1"/>
  56.     <!-- 包扫描 -->
  57.     <context:component-scan base-package="cn.dao,cn.service" />
  59.     <!-- 事务管理器 -->
  60.     <bean id="txManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
  61.         <property name="dataSource" ref="dataSource"/>
  62.     </bean>
  64.     <!-- 配置事务 -->
  65.     <tx:advice id="txAdvice" transaction-manager="txManager">
  66.         <tx:attributes>
  67.             <tx:method name="*" propagation="REQUIRED" isolation="DEFAULT"/>
  68.         </tx:attributes>
  69.     </tx:advice>
  70. </beans>

Spring MVC配置文件dispatcher.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans"
  3.        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.        xmlns:mvc="http://www.springframework.org/schema/mvc"
  5.        xmlns:context="http://www.springframework.org/schema/context"
  6.        xsi:schemaLocation="http://www.springframework.org/schema/beans
  7.                                 http://www.springframework.org/schema/beans/spring-beans.xsd
  8.                                 http://www.springframework.org/schema/mvc
  9.                                 http://www.springframework.org/schema/mvc/spring-mvc-4.3.xsd
  10.                                 http://www.springframework.org/schema/context
  11.                                 http://www.springframework.org/schema/context/spring-context-4.3.xsd">
  12.     <!-- 配置包扫描路径 -->
  13.     <context:component-scan base-package="cn.controller"/>
  14.     <!-- 配置使用注解驱动 -->
  15.     <mvc:annotation-driven />
  16.     <mvc:default-servlet-handler/>
  17.     <!-- 配置视图解析器 -->
  18.     <bean class="org.springframework.web.servlet.view.InternalResourceViewResolver">
  19.         <property name="prefix" value="/WEB-INF/jsp"/>
  20.         <property name="suffix" value=".jsp"/>
  21.     </bean>
  22. </beans>

Mybatis配置文件mybatis-config.xml

  1. <?xml version="1.0" encoding="UTF-8" ?>
  2. <!DOCTYPE configuration
  3.         PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
  4.         "http://mybatis.org/dtd/mybatis-3-config.dtd">
  5. <configuration>
  6.     <!--这一部分的datasource可以不需要,只需要mappers-->
  7.     <environments default="development">
  8.         <environment id="development">
  9.             <transactionManager type="JDBC"/>
  10.             <dataSource type="POOLED">
  11.                 <!--<property name="driver" value="com.mysql.jdbc.Driver"/>-->
  12.                 <property name="driver" value="com.mysql.cj.jdbc.Driver"/>
  13.                 <property name="url" value="jdbc:mysql://localhost:3306/python_db"/>
  14.                 <property name="username" value="root"/>
  15.                 <property name="password" value="******"/>
  16.             </dataSource>
  17.         </environment>
  18.     </environments>
  20.     <mappers>
  21.         <mapper class="cn.mapper.SEMapper"/>
  22.         <mapper class="cn.mapper.LessonMapper"/>
  23.     </mappers>
  25. </configuration>

5.配置controller

配置MVC控制器CourseClickController.java如下(SearchClickController.java同理)

  1. package cn.controller;
  3. import cn.pojo.Lesson;
  4. import cn.service.CourseClickService;
  5. import net.sf.json.JSONArray;
  6. import net.sf.json.JSONObject;
  7. import org.springframework.context.ApplicationContext;
  8. import org.springframework.context.support.ClassPathXmlApplicationContext;
  9. import org.springframework.stereotype.Controller;
  10. import org.springframework.web.bind.annotation.RequestMapping;
  11. import org.springframework.web.bind.annotation.RequestMethod;
  12. import org.springframework.web.bind.annotation.ResponseBody;
  14. import java.util.List;
  16. /**
  17.  * 课程点击Controller
  18.  */
  19. @Controller("COURSECLICK")
  20. @RequestMapping("/")
  21. public class CourseClickController {
  23.     //页面跳转
  24.     @RequestMapping("/courseClick")
  25.     public String toGetCourseClick(){
  26.         return "courseClick";
  27.     }
  29.     /**
  30.      *  sponseBody注解的作用是将controller的方法返回的对象通过适当的转换器转
  31.      *  换为指定的格式之后，写入到response对象的body区，通常用来返回JSON数据或者是XML
  32.      */
  33.     @ResponseBody
  34.     @RequestMapping(value = "/getCourseClickCount",method = RequestMethod.GET)
  35.     public JSONArray courseClickCount(){
  36.         JSONArray json = new JSONArray();
  38. //       Lesson类有课程名name(String)和对应的人点击量count(int)
  39.         ApplicationContext context = new ClassPathXmlApplicationContext(
  40.                 "beans.xml");
  41.         CourseClickService service = (CourseClickService) context.getBean("clickService");
  42.         Lesson use = (Lesson) context.getBean("Lesson");
  43.         List<Lesson> lessons = service.selectLessonALlSer(use);
  44.         for(Lesson les : lessons){
  45.             JSONObject jo = new JSONObject();
  46.             jo.put("name",les.getLesson());
  47.             jo.put("count",les.getLesson_count());
  49.             json.add(jo);
  50.         }
  51. //      list有多个课程名和对应的点击量
  52.         //封装JSON数据
  53.         return json;
  54.     }
  55. }

6.配置webapp

创建webapp，添加Web框架支持

以上文件的代码如下：

1.WEB-INF文件代码如下

  1.     web.xml
  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/web-app_4_0.xsd"
  5.          version="4.0">
  7.     <!-- 指定spring的配置文件beans.xml -->
  8.     <context-param>
  9.         <param-name>contextConfigLocation</param-name>
  10.         <param-value>classpath:dispatcher-servlet.xml</param-value>
  11.     </context-param>
  13.     <!-- 确保web服务器启动时，完成spring的容器初始化 -->
  14.     <listener>
  15.         <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
  16.     </listener>
  18.     <!-- 配置分发器 -->
  19.     <servlet>
  20.         <servlet-name>dispatcher</servlet-name>
  21.         <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
  22.     </servlet>
  24.     <servlet-mapping>
  25.         <servlet-name>dispatcher</servlet-name>
  26.         <url-pattern>/</url-pattern>
  27.     </servlet-mapping>
  28. </web-app>
  1.     courseClick.jsp
  1. <%@ taglib prefix="c" uri="http://java.sun.com/jsp/jstl/core" %>
  2. <%@ page contentType="text/html;charset=UTF-8" language="java" %>
  3. <html>
  4. <head>
  5.     <!-- 设置每隔10秒刷新一次页面-->
  6.     <meta http-equiv="refresh" content="10">
  7.     <script type="text/javascript" src="<c:url value="/js/echarts.js"/>"></script>
  8.     <script src="<c:url value="/js/jquery-3.6.0.js"/>"></script>
  9. </head>
  10. <body>
  11.     <div id="main1" style="width: 600px;height:500px;float: left;margin-top:100px"></div>
  12.     <div id="main2" style="width: 600px;height:500px;float: right;margin-top:100px"></div>
  13. </body>>
  15. <script src="<c:url value="/js/getCourseClickData.js"/>"></script>
  16. </html>

2.js文件代码如下

** echarts和jquery源文件有，或者自行在官网下载**

  1.     getCourseClickData.js
  1. function addScript(url){
  2.     document.write("<script language=javascript src="+echarts+"></script>");
  3. }
  4. var scources = [];
  5. var scources2 = [];
  6. var scources3 = [];
  7. var scources4 = [];
  8. //获得url上参数date的值
  9. function GetQueryString()
  10. {
  11.     var reg = new RegExp("(^|&)=([^&]*)(&|$)");
  12.     var r = window.location.search.substr(1).match(reg);//search,查询？后面的参数，并匹配正则
  13.     if(r!=null)return  unescape(r[2]); return null;
  14. }
  15. var date = GetQueryString();
  16. $.ajax({
  17.     type:"GET",
  18.     url:"/getCourseClickCount",
  19.     dataType:"json",
  20.     async:false,
  21.     success:function (result) {
  22.         if(scources.length != 0){
  23.             scources.clean();
  24.             scources2.clean();
  25.             scources3.clean();
  26.             scources4.clean();
  27.         }
  28.         for(var i = 0; i < result.length; i++){//饼图外侧所有数据
  29.             scources3.push({"value":result[i].count,"name":result[i].name});
  31.         }
  32.         for(var i = 0; i < 5; i++){//柱状图前五数据
  33.             scources.push(result[i].name);
  34.             scources2.push(result[i].count);
  35.         }
  36.         for(var i = 0; i < 3; i++){//内测饼图前三数据
  37.             scources4.push({"value":result[i].count,"name":result[i].name});
  38.         }
  39.     }
  40. })
  42. // 基于准备好的dom，初始化echarts实例
  43. var myChart = echarts.init(document.getElementById('main1'));
  45. // 指定图表的配置项和数据
  46. var option = {
  47.     title: {
  48.         text: '学习网实时实战课程访问量',
  49.         subtext: '课程点击数',
  50.         x:'center'
  51.     },
  52.     tooltip: {
  53.         legend: {
  54.             data: ['点击数']
  55.         }
  56.     },
  57.     xAxis: {
  58.         data: scources
  59.     },
  60.     yAxis: {},
  61.     series: [{
  62.         name: '点击数',
  63.         type: 'bar',
  64.         data: scources2
  65.     }]
  66. };
  67. // 使用刚指定的配置项和数据显示图表。
  68. myChart.setOption(option);
  70. var myChart = echarts.init(document.getElementById('main2'));
  72. // 指定图表的配置项和数据
  73. var option = {
  74.     title: {
  75.         text: '学习网实时实战课程访问量',
  76.         subtext: '课程点击数',
  77.         x:'center'
  78.     },
  79.     tooltip: {
  80.         trigger: 'item',
  81.         formatter: "{a} <br/>{b}: {c} ({d}%)"
  82.     },
  83.     legend: {
  84.         type: 'scroll',
  85.         orient: 'vertical',
  86.         right:  50,
  87.         top: 20,
  88.         data: scources
  89.     },
  90.     series: [
  91.         {
  92.             name: 'Access From',
  93.             type: 'pie',
  94.             selectedMode: 'single',
  95.             radius: [0, '30%'],
  96.             label: {
  97.                 position: 'inner',
  98.                 fontSize: 20
  99.             },
  100.             labelLine: {
  101.                 show: false
  102.             },
  103.             data: scources4
  104.         },
  105.         {
  106.             name: 'Access From',
  107.             type: 'pie',
  108.             radius: ['40%', '55%'],
  109.             labelLine: {
  110.                 length: 30
  111.             },
  112.             label: {
  113.                 formatter: '{a|{a}}{abg|}\n{hr|}\n  {b|{b} :}{c}  {per|{d}%}  ',
  114.                 backgroundColor: '#F6F8FC',
  115.                 borderColor: '#8C8D8E',
  116.                 borderWidth: 1,
  117.                 borderRadius: 5,
  118.                 rich: {
  119.                     a: {
  120.                         color: '#6E7079',
  121.                         lineHeight: 20,
  122.                         align: 'center'
  123.                     },
  124.                     hr: {
  125.                         borderColor: '#8C8D8E',
  126.                         width: '100%',
  127.                         borderWidth: 1,
  128.                         height: 0
  129.                     },
  130.                     b: {
  131.                         color: '#4C5058',
  132.                         fontSize: 15,
  133.                         fontWeight: 'bold',
  134.                         lineHeight: 30
  135.                     },
  136.                     per: {
  137.                         color: '#fff',
  138.                         backgroundColor: '#4C5058',
  139.                         padding: [3, 4],
  140.                         borderRadius: 5
  141.                     }
  142.                 }
  143.             },
  144.             data: scources3
  145.         }
  146.     ]
  147. };
  149. // 使用刚指定的配置项和数据显示图表。
  150. myChart.setOption(option);

7.配置tomcat服务器

先去官网下载tomcat服务器，这里使用的时8.5版本

一般此时IDEA会提醒你发现Web，是否配置Web框架，点击配置，前往“项目结构”，为Web配置工件，将右边SSMWork的与SSM相关的元素带入SSMWork_Web war中

进入编辑配置，添加工件，大致配置如下，删除“部署”中“应用程序上下文”的地址

5.效果展示

运行tomcat服务器后，默认先打开index.jsp网页，显示内容则表示tomcat配置没问题

在url地址栏输入/courseClcik或者/searchClick显示如下

网站默认10s刷新一次，此时开启scala目录的三个进程，再开启python脚本，页面便可以10秒更新一次数据，从而达到实时数据的展示

标签： spark 大数据
本文转载自: https://blog.csdn.net/Legosnow/article/details/124227289
版权归原作者 Legosnow 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Spark流处理日志+SSM前端展示（详细）”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？