【从0开始の全记录】Flume+Kafka+Spark+Spring Boot 统计网页访问量项目

点击上方蓝色字体，选择“设为星标”

回复”资源“获取更多资源

大数据技术与架构点击右侧关注，大数据开发领域最强公众号！

暴走大数据点击右侧关注，暴走大数据！

1.需求说明

1.1 需求

到现在为止的网页访问量

到现在为止从搜索引擎引流过来的网页访问量

项目总体框架如图所示：

1.2 用户行为日志内容

2.模拟日志数据制作

用Python制作模拟数据，数据包含：

• 不同的URL地址->url_paths

• 不同的跳转链接地址->http_refers

• 不同的搜索关键词->search_keyword

• 不同的状态码->status_codes

• 不同的IP地址->ip_slices

#coding=UTF-8
import random
import time

url_paths = [
    "class/112.html",
    "class/128.html",
    "class/145.html",
    "class/146.html",
    "class/131.html",
    "class/130.html",
    "class/145.html",
    "learn/821.html",
    "learn/825.html",
    "course/list"
]

http_refers=[
    "http://www.baidu.com/s?wd={query}",
    "https://www.sogou.com/web?query={query}",
    "http://cn.bing.com/search?q={query}",
    "http://search.yahoo.com/search?p={query}",
]

search_keyword = [
    "Spark+Sql",
    "Hadoop",
    "Storm",
    "Spark+Streaming",
    "大数据",
    "面试"
]

status_codes = ["200","404","500"]

ip_slices = [132,156,132,10,29,145,44,30,21,43,1,7,9,23,55,56,241,134,155,163,172,144,158]

def sample_url():
    return random.sample(url_paths,1)[0]

def sample_ip():
    slice = random.sample(ip_slices,4)
    return ".".join([str(item) for item in slice])

def sample_refer():
    if random.uniform(0,1) > 0.2:
        return "-"
    refer_str = random.sample(http_refers,1)
    query_str = random.sample(search_keyword,1)
    return refer_str[0].format(query=query_str[0])

def sample_status():
    return random.sample(status_codes,1)[0]

def generate_log(count = 10):
    time_str = time.strftime("%Y-%m-%d %H:%M:%S",time.localtime())

    f = open("/home/hadoop/tpdata/project/logs/access.log","w+")
    while count >= 1:
        query_log = "{ip}\t{local_time}\t\"GET /{url} HTTP/1.1\"\t{status}\t{refer}".format(
            local_time=time_str,
            url=sample_url(),
            ip=sample_ip(),
            refer=sample_refer(),
            status=sample_status())
        print(query_log)
        f.write(query_log + "\n")
        count = count - 1

if __name__ == '__main__':
    generate_log(100)

使用Linux Crontab定时调度工具，使其每一分钟产生一批数据。

表达式：

*/1 * * * *

编写python运行脚本：

vi log_generator.sh
python /home/hadoop/tpdata/log.py
chmod u+x log_generator.sh

配置Crontab： 

crontab -e
*/1 * * * * /home/hadoop/tpdata/project/log_generator.sh

2.Flume实时收集日志信息

开发时选型：

编写streaming_project.conf：

vi streaming_project.conf

exec-memory-logger.sources = exec-source
exec-memory-logger.sinks = logger-sink
exec-memory-logger.channels = memory-channel

exec-memory-logger.sources.exec-source.type = exec
exec-memory-logger.sources.exec-source.command = tail -F /home/hadoop/tpdata/project/logs/access.log
exec-memory-logger.sources.exec-source.shell = /bin/sh -c

exec-memory-logger.channels.memory-channel.type = memory

exec-memory-logger.sinks.logger-sink.type = logger

exec-memory-logger.sources.exec-source.channels = memory-channel
exec-memory-logger.sinks.logger-sink.channel = memory-channel

启动Flume测试：

flume-ng agent \
--name exec-memory-logger \
--conf $FLUME_HOME/conf \
--conf-file /home/hadoop/tpdata/project/streaming_project.conf \
-Dflume.root.logger=INFO,console

启动Zookeeper:

./zkServer.sh start

启动Kafka Server：

./kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

其中server.properties：

broker.id=0
############################# Socket Server Settings #############################
listeners=PLAINTEXT://:9092
host.name=hadoop000
advertised.host.name=192.168.1.9
advertised.port=9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
############################# Log Basics #############################
log.dirs=/home/hadoop/app/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
############################# Log Retention Policy #############################
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable=false
############################# Zookeeper #############################
zookeeper.connect=hadoop000:2181
zookeeper.connection.timeout.ms=6000

启动一个Kafka的消费者(topic用的之前的，没有的话可以新建一个)：

kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic streamingtopic

修改Flume配置文件，使得Flume的sink链接到Kafka：

vi streaming_project2.conf

exec-memory-kafka.sources = exec-source
exec-memory-kafka.sinks = kafka-sink
exec-memory-kafka.channels = memory-channel

exec-memory-kafka.sources.exec-source.type = exec
exec-memory-kafka.sources.exec-source.command = tail -F /home/hadoop/tpdata/project/logs/access.log
exec-memory-kafka.sources.exec-source.shell = /bin/sh -c

exec-memory-kafka.channels.memory-channel.type = memory

exec-memory-kafka.sinks.kafka-sink.type = org.apache.flume.sink.kafka.KafkaSink
exec-memory-kafka.sinks.kafka-sink.brokerList = hadoop000:9092
exec-memory-kafka.sinks.kafka-sink.topic = streamingtopic
exec-memory-kafka.sinks.kafka-sink.batchSize = 5
exec-memory-kafka.sinks.kafka-sink.requiredAcks = 1

exec-memory-kafka.sources.exec-source.channels = memory-channel
exec-memory-kafka.sinks.kafka-sink.channel = memory-channel

启动Flume：

flume-ng agent \
--name exec-memory-kafka \
--conf $FLUME_HOME/conf \
--conf-file /home/hadoop/tpdata/project/streaming_project2.conf \
-Dflume.root.logger=INFO,console

kafka消费者拿到数据：

4.Spark Streaming对接Kafka对数据消费

4.1 pom.xml：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.taipark.spark</groupId>
  <artifactId>sparktrain</artifactId>
  <version>1.0</version>
  <inceptionYear>2008</inceptionYear>
  <properties>
    <scala.version>2.11.8</scala.version>
    <kafka.version>0.9.0.0</kafka.version>
    <spark.version>2.2.0</spark.version>
    <hadoop.version>2.6.0-cdh5.7.0</hadoop.version>
    <hbase.version>1.2.0-cdh5.7.0</hbase.version>
  </properties>

  <repositories>
    <repository>
      <id>cloudera</id>
      <url>https://repository.cloudera.com/artifactory/cloudera-repos</url>
    </repository>
  </repositories>

  <dependencies>
    <dependency>
      <groupId>org.scala-lang</groupId>
      <artifactId>scala-library</artifactId>
      <version>${scala.version}</version>
    </dependency>

    <!--
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka_2.11</artifactId>
      <version>${kafka.version}</version>
    </dependency>
    -->

    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-client</artifactId>
      <version>${hadoop.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.hbase</groupId>
      <artifactId>hbase-client</artifactId>
      <version>${hbase.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.hbase</groupId>
      <artifactId>hbase-server</artifactId>
      <version>${hbase.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming_2.11</artifactId>
      <version>${spark.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
      <version>2.2.0</version>
    </dependency>

    <!--SS整合Flume依赖-->
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming-flume_2.11</artifactId>
      <version>${spark.version}</version>
    </dependency>

    <!--SS整合sink依赖-->
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming-flume-sink_2.11</artifactId>
      <version>${spark.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.commons</groupId>
      <artifactId>commons-lang3</artifactId>
      <version>3.5</version>
    </dependency>

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_2.11</artifactId>
      <version>${spark.version}</version>
    </dependency>

    <dependency>
      <groupId>mysql</groupId>
      <artifactId>mysql-connector-java</artifactId>
      <version>8.0.13</version>
    </dependency>

    <dependency>
      <groupId>com.fasterxml.jackson.module</groupId>
      <artifactId>jackson-module-scala_2.11</artifactId>
      <version>2.6.5</version>
    </dependency>

    <dependency>
      <groupId>net.jpountz.lz4</groupId>
      <artifactId>lz4</artifactId>
      <version>1.3.0</version>
    </dependency>

    <dependency>
      <groupId>org.apache.flume.flume-ng-clients</groupId>
      <artifactId>flume-ng-log4jappender</artifactId>
      <version>1.6.0</version>
    </dependency>
  </dependencies>

  <build>
    <sourceDirectory>src/main/scala</sourceDirectory>
    <testSourceDirectory>src/test/scala</testSourceDirectory>
    <plugins>
      <plugin>
        <groupId>org.scala-tools</groupId>
        <artifactId>maven-scala-plugin</artifactId>
        <executions>
          <execution>
            <goals>
              <goal>compile</goal>
              <goal>testCompile</goal>
            </goals>
          </execution>
        </executions>
        <configuration>
          <scalaVersion>${scala.version}</scalaVersion>
          <args>
            <arg>-target:jvm-1.5</arg>
          </args>
        </configuration>
      </plugin>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-eclipse-plugin</artifactId>
        <configuration>
          <downloadSources>true</downloadSources>
          <buildcommands>
            <buildcommand>ch.epfl.lamp.sdt.core.scalabuilder</buildcommand>
          </buildcommands>
          <additionalProjectnatures>
            <projectnature>ch.epfl.lamp.sdt.core.scalanature</projectnature>
          </additionalProjectnatures>
          <classpathContainers>
            <classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer>
            <classpathContainer>ch.epfl.lamp.sdt.launching.SCALA_CONTAINER</classpathContainer>
          </classpathContainers>
        </configuration>
      </plugin>
    </plugins>
  </build>
  <reporting>
    <plugins>
      <plugin>
        <groupId>org.scala-tools</groupId>
        <artifactId>maven-scala-plugin</artifactId>
        <configuration>
          <scalaVersion>${scala.version}</scalaVersion>
        </configuration>
      </plugin>
    </plugins>
  </reporting>
</project>

4.2 连通Kafka

新建Scala文件——WebStatStreamingApp.scala，首先使用Direct模式连通Kafka：

package com.taipark.spark.project

import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 使用Spark Streaming消费Kafka的数据
  */
object WebStatStreamingApp {
  def main(args: Array[String]): Unit = {

    if(args.length != 2){
      System.err.println("Userage:WebStatStreamingApp <brokers> <topics>");
      System.exit(1);
    }
    val Array(brokers,topics) = args

    val sparkConf = new SparkConf()
      .setAppName("WebStatStreamingApp")
      .setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf,Seconds(60))

    val kafkaParams = Map[String,String]("metadata.broker.list"-> brokers)
    val topicSet = topics.split(",").toSet
    val messages = KafkaUtils
      .createDirectStream[String,String,StringDecoder,StringDecoder](
      ssc,kafkaParams,topicSet
    )

    messages.map(_._2).count().print()

    ssc.start()
    ssc.awaitTermination()

  }
}

设定参数：

hadoop000:9092 streamingtopic

在本地测试是否连通：

连通成功，可以开始编写业务代码完成数据清洗(ETL)。

4.3 ETL

新建工具类DateUtils.scala：

package com.taipark.spark.project.utils

import java.util.Date

import org.apache.commons.lang3.time.FastDateFormat

/**
  * 日期时间工具类
  */
object DateUtils {
  val YYYYMMDDHHMMSS_FORMAT = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")
  val TARGET_FORMAT = FastDateFormat.getInstance("yyyyMMddHHmmss")

  def getTime(time:String)={
    YYYYMMDDHHMMSS_FORMAT.parse(time).getTime
  }

  def parseToMinute(time:String)={
    TARGET_FORMAT.format(new Date(getTime(time)))
  }

  def main(args: Array[String]): Unit = {
//    println(parseToMinute("2020-03-10 15:00:05"))
  }
}

新建ClickLog.scala：

package com.taipark.spark.project.domian

/**
  * 清洗后的日志信息
  */
case class ClickLog(ip:String,time:String,courseId:Int,statusCode:Int,referer:String)

修改WebStatStreamingApp.scala：

package com.taipark.spark.project.spark

import com.taipark.spark.project.domian.ClickLog
import com.taipark.spark.project.utils.DateUtils
import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 使用Spark Streaming消费Kafka的数据
  */
object WebStatStreamingApp {
  def main(args: Array[String]): Unit = {

    if(args.length != 2){
      System.err.println("Userage:WebStatStreamingApp <brokers> <topics>");
      System.exit(1);
    }
    val Array(brokers,topics) = args

    val sparkConf = new SparkConf()
      .setAppName("WebStatStreamingApp")
      .setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf,Seconds(60))

    val kafkaParams = Map[String,String]("metadata.broker.list"-> brokers)
    val topicSet = topics.split(",").toSet
    val messages = KafkaUtils
      .createDirectStream[String,String,StringDecoder,StringDecoder](
      ssc,kafkaParams,topicSet
    )

    //messages.map(_._2).count().print()

    //ETL
//    30.163.55.7  2020-03-10 14:32:01  "GET /class/112.html HTTP/1.1"  404  http://www.baidu.com/s?wd=Hadoop
    val logs = messages.map(_._2)
    val cleanData = logs.map(line => {
      val infos = line.split("\t")
      //infos(2) = "GET /class/112.html HTTP/1.1"
      val url = infos(2).split(" ")(1)
      var courseId = 0

      //拿到课程编号
      if(url.startsWith("/class")){
        val courseIdHTML = url.split("/")(2)
        courseId = courseIdHTML.substring(0,courseIdHTML.lastIndexOf(".")).toInt
      }

      ClickLog(infos(0),DateUtils.parseToMinute(infos(1)),courseId,infos(3).toInt,infos(4))
    }).filter(clicklog => clicklog.courseId != 0)

    cleanData.print()

    ssc.start()
    ssc.awaitTermination()

  }
}

run起来测试一下：

ETL完成。

4.4 功能一：到现在为止某网站的访问量

使用数据库来存储统计结果，可视化前端根据yyyyMMdd courseid把数据库里的结果展示出来。

选择HBASE作为数据库。要启动HDFS与Zookeeper。

启动HDFS：

./start-dfs.sh

启动HBASE：

./start-hbase.sh

./hbase shell
list

HBASE表设计：

create 'web_course_clickcount','info'

hbase(main):008:0> desc 'web_course_clickcount'
Table web_course_clickcount is ENABLED                                                                 
web_course_clickcount                                                                                  
COLUMN FAMILIES DESCRIPTION                                                                            
{NAME => 'info', BLOOMFILTER => 'ROW', VERSIONS => '1', IN_MEMORY => 'false', KEEP_DELETED_CELLS => 'FA
LSE', DATA_BLOCK_ENCODING => 'NONE', TTL => 'FOREVER', COMPRESSION => 'NONE', MIN_VERSIONS => '0', BLOC
KCACHE => 'true', BLOCKSIZE => '65536', REPLICATION_SCOPE => '0'}                                      
1 row(s) in 0.1650 seconds

Rowkey设计：

day_courseid

使用Scala来操作HBASE：

新建网页点击数实体类 CourseClickCount.scala：

package com.taipark.spark.project.domian

/**
  * 课程网页点击数
  * @param day_course HBASE中的rowkey
  * @param click_count 对应的点击总数
  */
case class CourseClickCount(day_course:String,click_count:Long)

新建数据访问层 CourseClickCountDAO.scala：

package com.taipark.spark.project.dao

import com.taipark.spark.project.domian.CourseClickCount

import scala.collection.mutable.ListBuffer

object CourseClickCountDAO {
  val tableName = "web_course_clickcount"
  val cf = "info"
  val qualifer = "click_count"

  /**
    * 保存数据到HBASE
    * @param list
    */
  def save(list:ListBuffer[CourseClickCount]): Unit ={

  }

  /**
    * 根据rowkey查询值
    * @param day_course
    * @return
    */
  def count(day_course:String):Long={
    0l
  }
}

利用Java实现HBaseUtils打通其与HBASE：

package com.taipark.spark.project.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * HBase操作工具类:Java工具类采用单例模式封装
 */
public class HBaseUtils {
    HBaseAdmin admin = null;
    Configuration configuration = null;

    //私有构造方法(单例模式)
    private HBaseUtils(){
        configuration = new Configuration();
        configuration.set("hbase.zookeeper.quorum",
                "hadoop000:2181");
        configuration.set("hbase.rootdir",
                "hdfs://hadoop000:8020/hbase");

        try {
            admin = new HBaseAdmin(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    private static HBaseUtils instance = null;

    public static synchronized HBaseUtils getInstance(){
        if(instance == null){
            instance = new HBaseUtils();
        }
        return instance;
    }

    //根据表名获取HTable实例
    public HTable getTable(String tableName){
        HTable table = null;
        try {
            table = new HTable(configuration,tableName);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return table;
    }

    /**
     * 添加一条记录到HBASE表
     * @param tableName 表名
     * @param rowkey    表rowkey
     * @param cf    表的columnfamily
     * @param column    表的列
     * @param value     写入HBASE的值
     */
    public void put(String tableName,String rowkey,String cf,String column,String value){
        HTable table = getTable(tableName);
        Put put = new Put(Bytes.toBytes(rowkey));
        put.add(Bytes.toBytes(cf),Bytes.toBytes(column),Bytes.toBytes(value));
        try {
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
//        HTable hTable = HBaseUtils.getInstance().getTable("web_course_clickcount");
//        System.out.println(hTable.getName().getNameAsString());
        String tableName = "web_course_clickcount";
        String rowkey = "20200310_88";
        String cf = "info";
        String column = "click_count";
        String value = "2";
        HBaseUtils.getInstance().put(tableName,rowkey,cf,column,value);

    }
}

测试运行：

测试工具类成功后继续编写DAO的代码：

package com.taipark.spark.project.dao

import com.taipark.spark.project.domian.CourseClickCount
import com.taipark.spark.project.utils.HBaseUtils
import org.apache.hadoop.hbase.client.Get
import org.apache.hadoop.hbase.util.Bytes

import scala.collection.mutable.ListBuffer

object CourseClickCountDAO {
  val tableName = "web_course_clickcount"
  val cf = "info"
  val qualifer = "click_count"

  /**
    * 保存数据到HBASE
    * @param list
    */
  def save(list:ListBuffer[CourseClickCount]): Unit ={
    val table = HBaseUtils.getInstance().getTable(tableName)

    for(ele <- list){
      table.incrementColumnValue(
        Bytes.toBytes(ele.day_course),
        Bytes.toBytes(cf),
        Bytes.toBytes(qualifer),
        ele.click_count)
    }
  }

  /**
    * 根据rowkey查询值
    * @param day_course
    * @return
    */
  def count(day_course:String):Long={
    val table = HBaseUtils.getInstance().getTable(tableName)

    val get = new Get(Bytes.toBytes(day_course))
    val value = table.get(get).getValue(cf.getBytes,qualifer.getBytes)

    if (value == null){
      0L
    }else{
      Bytes.toLong(value)
    }
  }

  def main(args: Array[String]): Unit = {
    val list = new ListBuffer[CourseClickCount]
    list.append(CourseClickCount("2020311_8",8))
    list.append(CourseClickCount("2020311_9",9))
    list.append(CourseClickCount("2020311_10",1))
    list.append(CourseClickCount("2020311_2",15))

    save(list)
  }
}

测试运行一下，用hbase shell查看：

scan 'web_course_clickcount'

将Spark Streaming处理结果写到HBASE中：

package com.taipark.spark.project.spark

import com.taipark.spark.project.dao.CourseClickCountDAO
import com.taipark.spark.project.domian.{ClickLog, CourseClickCount}
import com.taipark.spark.project.utils.DateUtils
import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable.ListBuffer

/**
  * 使用Spark Streaming消费Kafka的数据
  */
object WebStatStreamingApp {
  def main(args: Array[String]): Unit = {

    if(args.length != 2){
      System.err.println("Userage:WebStatStreamingApp <brokers> <topics>");
      System.exit(1);
    }
    val Array(brokers,topics) = args

    val sparkConf = new SparkConf()
      .setAppName("WebStatStreamingApp")
      .setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf,Seconds(60))

    val kafkaParams = Map[String,String]("metadata.broker.list"-> brokers)
    val topicSet = topics.split(",").toSet
    val messages = KafkaUtils
      .createDirectStream[String,String,StringDecoder,StringDecoder](
      ssc,kafkaParams,topicSet
    )

    //messages.map(_._2).count().print()

    //ETL
//    30.163.55.7  2020-03-10 14:32:01  "GET /class/112.html HTTP/1.1"  404  http://www.baidu.com/s?wd=Hadoop
    val logs = messages.map(_._2)
    val cleanData = logs.map(line => {
      val infos = line.split("\t")
      //infos(2) = "GET /class/112.html HTTP/1.1"
      val url = infos(2).split(" ")(1)
      var courseId = 0

      //拿到课程编号
      if(url.startsWith("/class")){
        val courseIdHTML = url.split("/")(2)
        courseId = courseIdHTML.substring(0,courseIdHTML.lastIndexOf(".")).toInt
      }

      ClickLog(infos(0),DateUtils.parseToMinute(infos(1)),courseId,infos(3).toInt,infos(4))
    }).filter(clicklog => clicklog.courseId != 0)

//    cleanData.print()

    cleanData.map(x => {
      //HBase rowkey设计：20200311_9
      ((x.time.substring(0,8)) + "_" + x.courseId,1)
    }).reduceByKey(_+_).foreachRDD(rdd =>{
      rdd.foreachPartition(partitionRecords =>{
        val list = new ListBuffer[CourseClickCount]

        partitionRecords.foreach(pair =>{
          list.append(CourseClickCount(pair._1,pair._2))
        })

        CourseClickCountDAO.save(list)
      })
    })

    ssc.start()
    ssc.awaitTermination()

  }
}

测试：

4.5 功能二：到现在为止某网站的搜索引擎引流访问量

HBASE表设计：

create 'web_course_search_clickcount','info'

设计rowkey：

day_search_1

确定实体类：

package com.taipark.spark.project.domian

/**
  * 网站从搜索引擎过来的点击数实体类
  * @param day_search_course
  * @param click_count
  */
case class CourseSearchClickCount (day_search_course:String,click_count:Long)

开发DAO CourseSearchClickCountDAO.scala：

package com.taipark.spark.project.dao

import com.taipark.spark.project.domian.{CourseClickCount, CourseSearchClickCount}
import com.taipark.spark.project.utils.HBaseUtils
import org.apache.hadoop.hbase.client.Get
import org.apache.hadoop.hbase.util.Bytes

import scala.collection.mutable.ListBuffer

object CourseSearchClickCountDAO {
  val tableName = "web_course_search_clickcount"
  val cf = "info"
  val qualifer = "click_count"

  /**
    * 保存数据到HBASE
    * @param list
    */
  def save(list:ListBuffer[CourseSearchClickCount]): Unit ={
    val table = HBaseUtils.getInstance().getTable(tableName)

    for(ele <- list){
      table.incrementColumnValue(
        Bytes.toBytes(ele.day_search_course),
        Bytes.toBytes(cf),
        Bytes.toBytes(qualifer),
        ele.click_count)
    }
  }

  /**
    * 根据rowkey查询值
    * @param day_search_course
    * @return
    */
  def count(day_search_course:String):Long={
    val table = HBaseUtils.getInstance().getTable(tableName)

    val get = new Get(Bytes.toBytes(day_search_course))
    val value = table.get(get).getValue(cf.getBytes,qualifer.getBytes)

    if (value == null){
      0L
    }else{
      Bytes.toLong(value)
    }
  }

  def main(args: Array[String]): Unit = {
    val list = new ListBuffer[CourseSearchClickCount]
    list.append(CourseSearchClickCount("2020311_www.baidu.com_8",8))
    list.append(CourseSearchClickCount("2020311_cn.bing.com_9",9))

    save(list)
    println(count("020311_www.baidu.com_8"))
  }
}

测试：

在Spark Streaming中写到HBASE：

package com.taipark.spark.project.spark

import com.taipark.spark.project.dao.{CourseClickCountDAO, CourseSearchClickCountDAO}
import com.taipark.spark.project.domian.{ClickLog, CourseClickCount, CourseSearchClickCount}
import com.taipark.spark.project.utils.DateUtils
import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable.ListBuffer

/**
  * 使用Spark Streaming消费Kafka的数据
  */
object WebStatStreamingApp {
  def main(args: Array[String]): Unit = {

    if(args.length != 2){
      System.err.println("Userage:WebStatStreamingApp <brokers> <topics>");
      System.exit(1);
    }
    val Array(brokers,topics) = args

    val sparkConf = new SparkConf()
      .setAppName("WebStatStreamingApp")
      .setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf,Seconds(60))

    val kafkaParams = Map[String,String]("metadata.broker.list"-> brokers)
    val topicSet = topics.split(",").toSet
    val messages = KafkaUtils
      .createDirectStream[String,String,StringDecoder,StringDecoder](
      ssc,kafkaParams,topicSet
    )

    //messages.map(_._2).count().print()

    //ETL
//    30.163.55.7  2020-03-10 14:32:01  "GET /class/112.html HTTP/1.1"  404  http://www.baidu.com/s?wd=Hadoop
    val logs = messages.map(_._2)
    val cleanData = logs.map(line => {
      val infos = line.split("\t")
      //infos(2) = "GET /class/112.html HTTP/1.1"
      val url = infos(2).split(" ")(1)
      var courseId = 0

      //拿到课程编号
      if(url.startsWith("/class")){
        val courseIdHTML = url.split("/")(2)
        courseId = courseIdHTML.substring(0,courseIdHTML.lastIndexOf(".")).toInt
      }

      ClickLog(infos(0),DateUtils.parseToMinute(infos(1)),courseId,infos(3).toInt,infos(4))
    }).filter(clicklog => clicklog.courseId != 0)

//    cleanData.print()

    //需求一
    cleanData.map(x => {
      //HBase rowkey设计：20200311_9
      ((x.time.substring(0,8)) + "_" + x.courseId,1)
    }).reduceByKey(_+_).foreachRDD(rdd =>{
      rdd.foreachPartition(partitionRecords =>{
        val list = new ListBuffer[CourseClickCount]

        partitionRecords.foreach(pair =>{
          list.append(CourseClickCount(pair._1,pair._2))
        })

        CourseClickCountDAO.save(list)
      })
    })

    //需求二
    cleanData.map(x =>{
      //http://www.baidu.com/s?wd=Spark+Streaming
      val referer = x.referer.replaceAll("//","/")
      //http:/www.baidu.com/s?wd=Spark+Streaming
      val splits = referer.split("/")
      var host = ""
      //splits.length == 1 => -
      if(splits.length > 2){
        host = splits(1)
      }

      (host,x.courseId,x.time)
    }).filter(_._1 != "").map(x =>{
      (x._3.substring(0,8) + "_" + x._1 + "_" + x._2,1)
    }).reduceByKey(_+_).foreachRDD(rdd =>{
      rdd.foreachPartition(partitionRecords =>{
        val list = new ListBuffer[CourseSearchClickCount]

        partitionRecords.foreach(pair =>{
          list.append(CourseSearchClickCount(pair._1,pair._2))
        })

        CourseSearchClickCountDAO.save(list)
      })
    })

    ssc.start()
    ssc.awaitTermination()

  }
}

测试：

5.生产环境部署

不要硬编码，把setAppName和setMaster注释掉：

  val sparkConf = new SparkConf()
//      .setAppName("WebStatStreamingApp")
//      .setMaster("local[2]")

Maven打包部署前，需要将pom中指定build目录的两行注释掉，以防报错：

    <!--
    <sourceDirectory>src/main/scala</sourceDirectory>
    <testSourceDirectory>src/test/scala</testSourceDirectory>
    -->

Maven打包传到服务器：

利用spark-submit提交：

./spark-submit \
--master local[5] \
--name WebStatStreamingApp \
--class com.taipark.spark.project.spark.WebStatStreamingApp \
/home/hadoop/tplib/sparktrain-1.0.jar \
hadoop000:9092 streamingtopic

报错：

Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/spark/streaming/kafka/KafkaUtils$

修改，添加jar包spark-streaming-kafka-0-8_2.11：

./spark-submit \
--master local[5] \
--name WebStatStreamingApp \
--class com.taipark.spark.project.spark.WebStatStreamingApp \
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 \
/home/hadoop/tplib/sparktrain-1.0.jar \
hadoop000:9092 streamingtopic

报错：

java.lang.NoClassDefFoundError: org/apache/hadoop/hbase/client/HBaseAdmin

修改，增加HBASE的jar包：

./spark-submit \
--master local[5] \
--name WebStatStreamingApp \
--class com.taipark.spark.project.spark.WebStatStreamingApp \
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 \
--jars $(echo /home/hadoop/app/hbase-1.2.0-cdh5.7.0/lib/*.jar | tr ' ' ',') \
/home/hadoop/tplib/sparktrain-1.0.jar \
hadoop000:9092 streamingtopic

运行：

后台运行成功

6.Spring Boot开发

6.1 测试ECharts

新建一个Spring Boot项目，下载ECharts，利用其在线编译，获得echarts.min.js，放在resources/static/js下

pox.xml添加一个依赖：

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-thymeleaf</artifactId>
        </dependency>

resources/templates里做一个test.html：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>test</title>
    <!-- 引入 ECharts 文件 -->
    <script src="js/echarts.min.js"></script>
</head>
<body>
<!-- 为 ECharts 准备一个具备大小（宽高）的 DOM -->
<div id="main" style="width: 600px;height:400px;"></div>
<script type="text/javascript">
    // 基于准备好的dom，初始化echarts实例
    var myChart = echarts.init(document.getElementById('main'));

    // 指定图表的配置项和数据
    var option = {
        title: {
            text: 'ECharts 入门示例'
        },
        tooltip: {},
        legend: {
            data:['销量']
        },
        xAxis: {
            data: ["衬衫","羊毛衫","雪纺衫","裤子","高跟鞋","袜子"]
        },
        yAxis: {},
        series: [{
            name: '销量',
            type: 'bar',
            data: [5, 20, 36, 10, 10, 20]
        }]
    };

    // 使用刚指定的配置项和数据显示图表。
    myChart.setOption(option);
</script>
</body>
</html>

新建java文件：

package com.taipark.spark.web;

import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.ModelAndView;

/**
 * 测试
 */
@RestController
public class HelloBoot {
    @RequestMapping(value = "/hello",method = RequestMethod.GET)
    public String sayHello(){
        return "HelloWorld!";
    }

    @RequestMapping(value = "/first",method = RequestMethod.GET)
    public ModelAndView firstDemo(){
        return new ModelAndView("test");
    }
}

测试一下：

成功

6.2 动态实现ECharts

添加repository：

    <repositories>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
    </repositories>

添加依赖：

        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>1.2.0-cdh5.7.0</version>
        </dependency>

创建HBaseUtils.java：

package com.taipark.spark.web.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;


import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class HBaseUtils {
    HBaseAdmin admin = null;
    Configuration configuration = null;

    //私有构造方法(单例模式)
    private HBaseUtils(){
        configuration = new Configuration();
        configuration.set("hbase.zookeeper.quorum",
                "hadoop000:2181");
        configuration.set("hbase.rootdir",
                "hdfs://hadoop000:8020/hbase");

        try {
            admin = new HBaseAdmin(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    private static HBaseUtils instance = null;

    public static synchronized HBaseUtils getInstance(){
        if(instance == null){
            instance = new HBaseUtils();
        }
        return instance;
    }

    //根据表名获取HTable实例
    public HTable getTable(String tableName){
        HTable table = null;
        try {
            table = new HTable(configuration,tableName);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return table;
    }

    /**
     * 根据表名和输入条件获取HBASE的记录数
     * @param tableName
     * @param dayCourse
     * @return
     */
    public Map<String,Long> query(String tableName,String condition) throws Exception{
        Map<String,Long> map = new HashMap<>();

        HTable table = getTable(tableName);
        String cf ="info";
        String qualifier = "click_count";

        Scan scan = new Scan();
        Filter filter = new PrefixFilter(Bytes.toBytes(condition));
        scan.setFilter(filter);
        ResultScanner rs = table.getScanner(scan);
        for(Result result:rs){
            String row = Bytes.toString(result.getRow());
            long clickCount = Bytes.toLong(result.getValue(cf.getBytes(), qualifier.getBytes()));
            map.put(row,clickCount);
        }
        return map;
    }

    public static void main(String[] args) throws Exception{
        Map<String, Long> map = HBaseUtils.getInstance().query("web_course_clickcount", "20200311");

        for(Map.Entry<String,Long> entry:map.entrySet()){
            System.out.println(entry.getKey() + ":" + entry.getValue());
        }
    }
}

测试通过：

定义网页访问数量Bean：

package com.taipark.spark.web.domain;

import org.springframework.stereotype.Component;

/**
 * 网页访问数量实体类
 */
@Component
public class CourseClickCount {

    private String name;
    private long value;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public long getValue() {
        return value;
    }

    public void setValue(long value) {
        this.value = value;
    }
}

DAO层：

package com.taipark.spark.web.dao;

import com.taipark.spark.web.domain.CourseClickCount;
import com.taipark.spark.web.utils.HBaseUtils;
import org.springframework.stereotype.Component;


import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 网页访问数量数据访问层
 */
@Component
public class CourseClickDAO {
    /**
     * 根据天查询
     * @param day
     * @return
     * @throws Exception
     */
    public List<CourseClickCount> query(String day) throws Exception{

        List<CourseClickCount> list = new ArrayList<>();
        //去HBase表中根据day获取对应网页的访问量
        Map<String, Long> map = HBaseUtils.getInstance().query("web_course_clickcount", "20200311");
        for(Map.Entry<String,Long> entry:map.entrySet()){
            CourseClickCount model = new CourseClickCount();
            model.setName(entry.getKey());
            model.setValue(entry.getValue());

            list.add(model);
        }
        return list;
    }

    public static void main(String[] args) throws Exception{
        CourseClickDAO dao = new CourseClickDAO();
        List<CourseClickCount> list = dao.query( "20200311");

        for(CourseClickCount model:list){
            System.out.println(model.getName() + ":" + model.getValue());
        }
    }
}

使用JSON需要引入：

        <dependency>
            <groupId>net.sf.json-lib</groupId>
            <artifactId>json-lib</artifactId>
            <version>2.4</version>
            <classifier>jdk15</classifier>
        </dependency>

Web层：

package com.taipark.spark.web.spark;

import com.taipark.spark.web.dao.CourseClickDAO;
import com.taipark.spark.web.domain.CourseClickCount;
import net.sf.json.JSONArray;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.ModelAndView;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * web层
 */
@RestController
public class WebStatApp {

    private static Map<String,String> courses = new HashMap<>();
    static {
        courses.put("112","某些外国人对中国有多不了解？");
        courses.put("128","你认为有哪些失败的建筑？");
        courses.put("145","为什么人类想象不出四维空间？");
        courses.put("146","有什么一眼看上去很舒服的头像？");
        courses.put("131","男朋友心情不好时女朋友该怎么办？");
        courses.put("130","小白如何从零开始运营一个微信公众号？");
        courses.put("821","为什么有人不喜欢极简主义？");
        courses.put("825","有哪些书看完后会让人很后悔没有早看到？");
    }

//    @Autowired
//    CourseClickDAO courseClickDAO;
//    @RequestMapping(value = "/course_clickcount_dynamic",method = RequestMethod.GET)
//    public ModelAndView courseClickCount() throws Exception{
//        ModelAndView view = new ModelAndView("index");
//        List<CourseClickCount> list = courseClickDAO.query("20200311");
//
//        for(CourseClickCount model:list){
//            model.setName(courses.get(model.getName().substring(9)));
//        }
//        JSONArray json = JSONArray.fromObject(list);
//
//        view.addObject("data_json",json);
//
//        return view;
//    }

    @Autowired
    CourseClickDAO courseClickDAO;
    @RequestMapping(value = "/course_clickcount_dynamic",method = RequestMethod.POST)
    @ResponseBody
    public List<CourseClickCount> courseClickCount() throws Exception{
        ModelAndView view = new ModelAndView("index");
        List<CourseClickCount> list = courseClickDAO.query("20200311");
        for(CourseClickCount model:list){
            model.setName(courses.get(model.getName().substring(9)));
        }

        return list;
    }

    @RequestMapping(value = "/echarts",method = RequestMethod.GET)
    public ModelAndView echarts(){
        return new ModelAndView("echarts");
    }
}

下载JQuery，并放到static/js下，新建echarts.html：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>web_stat</title>
    <!-- 引入 ECharts 文件 -->
    <script src="js/echarts.min.js"></script>
    <script src="js/jquery-3.4.1.min.js"></script>
</head>
<body>
<!-- 为 ECharts 准备一个具备大小（宽高）的 DOM -->
<div id="main" style="width: 960px;height:540px;position: absolute;top:50%;left:50%;margin-top: -200px;margin-left: -300px;"></div>
<script type="text/javascript">
    // 基于准备好的dom，初始化echarts实例
    var myChart = echarts.init(document.getElementById('main'));

    option = {
        title: {
            text: '某站点实时流处理访问量统计',
            subtext: '网页访问次数',
            left: 'center'
        },
        tooltip: {
            trigger: 'item',
            formatter: '{a} <br/>{b} : {c} ({d}%)'
        },
        legend: {
            orient: 'vertical',
            left: 'left'
        },
        series: [
            {
                name: '访问次数',
                type: 'pie',
                radius: '55%',
                center: ['50%', '60%'],
                data: (function () {
                   var datas = [];
                   $.ajax({
                       type: "POST",
                       url: "/taipark/course_clickcount_dynamic",
                       dataType: "json",
                       async: false,
                       success: function (result) {
                           for(var i=0;i<result.length;i++){
                               datas.push({"value":result[i].value,
                               "name":result[i].name})
                           }
                       }
                   })
                    return datas;
                })(),
                emphasis: {
                    itemStyle: {
                        shadowBlur: 10,
                        shadowOffsetX: 0,
                        shadowColor: 'rgba(0, 0, 0, 0.5)'
                    }
                }
            }
        ]
    };
    // 使用刚指定的配置项和数据显示图表。
    myChart.setOption(option);
</script>
</body>
</html>

测试一下：

6.3 Spring的服务器部署

Maven打包并上传服务器

java -jar web-0.0.1.jar

完成~

欢迎点赞+收藏+转发朋友圈素质三连

文章不错？点个【在看】吧！ 👇