大数据技术之_26_交通状态预测项目_01

2019-05-30 09:00:25 浏览数 (1)

一 项目背景

该项目以车辆预测为基础，学习业务解决的方法论。 学习完本项目后，可以解决如下问题或适用于如下业务场景：   1、公路堵车预测   2、地铁人流量预测   3、共享单车聚集点预测等等

扩展知识：

代码语言：javascript复制
spark-shell --master yarn --conf spark.eventLog.dir=hdfs://hadoop/tmp/spark2 --jars /home/hadoop-SNAPSHOT.jar

智能判断：
    轨迹：将某一辆车在所有监测点留下的踪迹聚合连线，则为该车的轨迹。
    跟车：判断某一辆车是否被跟踪（此处“跟踪”为中性词汇）等。比如：婚车（判断是否属于一个车队）
    碰撞：这里不是撞车分析，而是在几个电子围栏内（比如，监测点1，监测点2），同一辆车，在某一个时间范围内，检测出该车出现在不同的监测点。电子围栏：比如OA打卡。

技侦支持：
    布控：警方输入布控的车辆信息（车牌号，车型，颜色等等）
    实时报警：符合布控标准，则报警
    套牌分析：相同车牌号，车辆信息不一致
    落脚点：车辆在哪个区域停留时间长

统计分析：
    流量统计：哪几个监测点的车流量比较高
    外来车辆统计等等

数据结构示例：
日期            关卡id       摄像id       车辆          发生时间          速度    公路id   区域id（维度=特征值向量）

2017-04-25      0001        09203       京W47147     2017-04-25 20:58:17     138     49      04
2017-04-25      0005        06975       京W47147     2017-04-25 20:12:39     50      10      06
2017-04-25      0001        02846       京W47147     2017-04-25 20:20:20     214     21      00
2017-04-25      0003        06044       京W47147     2017-04-25 20:15:58     78      47      01
2017-04-25      0000        01599       京W47147     2017-04-25 20:40:58     59      32      01
2017-04-25      0002        09260       京M91266     2017-04-25 09:09:57     105     15      00

一个 Event（事件）至少包含一行数据。
因为 Kafka 是基于事件的。

为什么一个 Event（事件）包含多行数据？
答：我们将多行数据封装(打包)成一个 Event，发送给 Kafka，这样的好处是减少网络IO。
如何打包呢？
答：使用 json 格式，如下：

{
    "monitor_arr":
    [
        {
            "time": "2017-04-25",
            monitor_id:"0001",
            ...
        },
        {
            "time": "2017-04-25",
            monitor_id:"0005",
            ...
        },
        ......
    ]
}

有监督学习：有标签(label)的训练 --> 建模的过程 --> 求通项公式的过程 --> 求拟合函数的过程 --> 求参数的过程 --> 连续数据，常用算法：回归算法 --> 线性回归、逻辑斯特回归
无监督学习：没有标签(label)的训练 --> 离散数据 --> 比如归类问题，常用算法：聚类算法 --> 支持向量机、随机森林（起源于决策树，万能药）、K-means 算法
半监督学习：一部分有标签，一部分无标签。

老罗的锤子手机不赚钱，为了交个朋友--情怀，卖配件、T恤等赚钱。

平民化的最接近科学/科研 --> 计算机

维度认知：

二 项目架构与环境

2.1 项目架构

2.2 项目环境

以下环境为本次项目使用的情景，并非只有在此环境下才可以完成整体业务需求。请灵活变动。（本例已在以下环境中完成测试）

三 项目实现

我们新建 java 项目 tf，之后的每一个项目模块都建立于该项目之下。然后删除掉 src 目录。

3.1 数据模拟

  请确保 zookeeper 和 kafka 的正确配置。   如果之前安装的 scala 版本不是 2.11.8 请替换之：

代码语言：javascript复制
$ tar -zxf /opt/software/scala-2.11.8.tgz -C /opt/module/

使用 root 用户，配置环境变量：
[atguigu@hadoop102 module]$ sudo vim /etc/profile

#SCALA_HOME
export SCALA_HOME=/opt/module/scala/scala-2.11.8
export PATH=$PATH:$SCALA_HOME/bin

使环境变量生效：
[atguigu@hadoop102 module]$ sudo source /etc/profile

  我们需要产生一些监测点的模拟车速数据，并将这些数据实时的发送给 kafka，保存到 traffic 主题中，以供后续的 Spark 读取数据并加工之后存放于 redis。

3.1.1 数据结构
3.1.2 编写代码

思路：   a) 新建模块 maven 工程：tf_producer   b) 配置 maven 依赖。   c) 因为要把数据发送给 kafka，所以配置 kafka 属性，保存于某个配置文件中。   d) 编写 kafka 加载属性的工具类。   e) 每隔 5 分钟，切换一次模拟状态，例如第一个五分钟，车速都在 30km/h 以上，下一个五分钟，车速都在 10km/h 以下，往复模拟公路一会堵车，一会不堵车的情况。   f) 启动 zookeeper 集群和 kafka 集群，并创建 kafka 主题，检查主题存在性。   g) 将数据发送至 kafka 并使用 kafka console-consumer 进行检测。

1) 新建项目：tf_producer

2) maven 的 pom.xml 文件配置：

代码语言：javascript复制
<?xml version="1.0" encoding="UTF-8"?>
<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/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.atguigu</groupId>
    <artifactId>tf_producer</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.11.0.2</version>
        </dependency>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.41</version>
        </dependency>
    </dependencies>
</project>

3) kafka 属性配置文件：kafka.properties（生产者）

代码语言：javascript复制
# 设置 kafka 的 brokerlist
bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092

# 生产者序列化
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer

acks=all
retries=0

# 设置消费者所属的消费组
group.id=g_traffic1

# 设置是否自动确认 offset
enable.auto.commit=true

# 设置自动确认 offset 的时间间隔
auto.commit.interval.ms=30000

# 设置本次消费的主题
kafka.topics=traffic

# 设置 zookeeper 中 follower 和 leader 之间的关于 kafka 的信息同步时间间隔
zookeeper.sync.time.ms=250
num.io.threads=12
batch.size=65536
buffer.memory=524288

# kafka 中消息保存的时间（单位是小时），企业开发中是 7 天
log.retention.hours=2

4) 编写 kafka 属性加载工具类：PropertyUtil.scala

代码语言：javascript复制
package com.atguigu.utils

import java.util.Properties

object PropertyUtil {
  val properties = new Properties()
  // 加载配置属性
  try {
    val inputStream = ClassLoader.getSystemResourceAsStream("kafka.properties")
    properties.load(inputStream)
  } catch {
    case ex: Exception => println(ex)
  } finally {

  }

  // 定义通过键得到属性值的方法
  def getProperty(key: String): String = properties.getProperty(key)
}

5) 开始模拟数据，每隔 5 分钟切换一次模拟状态，文件：Producer.scala

代码语言：javascript复制
package com.atguigu.producer

import java.text.DecimalFormat
import java.util.Calendar

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import com.atguigu.utils.PropertyUtil

import scala.util.Random
import java._

import com.alibaba.fastjson.JSON

/**
  * 模拟产生数据，同时把数据实时的发送到 kafka
  * 随机产生 监测点id 以及 速度
  * 序列化为 json
  * 发送给 kafka
  */
object Producer {

  def main(args: Array[String]): Unit = {
    // 读取配置文件信息
    val properties = PropertyUtil.properties
    // 创建 kafka 生产者对象
    val producer = new KafkaProducer[String, String](properties)

    // 模拟产生实时数据，单位为：秒
    var startTime = Calendar.getInstance().getTimeInMillis() / 1000

    // 数据模拟，堵车状态切换的周期单位为：秒
    val trafficCycle = 300

    val df = new DecimalFormat("0000")
    // 开始不停的实时产生数据
    while (true) {
      // 模拟产生监测点 id：1~20
      val randomMonitorId = df.format(Random.nextInt(20)   1)
      // 模拟车速
      var randomSpeed = "000"

      // 得到本条数据产生时的当前时间，单位为：秒
      val currentTime = Calendar.getInstance().getTimeInMillis() / 1000
      // 每 5 分钟切换一次公路状态
      if (currentTime - startTime > trafficCycle) {
        randomSpeed = new DecimalFormat("000").format(Random.nextInt(16))
        if (currentTime - startTime > trafficCycle * 2) {
          startTime = currentTime
        }
      } else {
        randomSpeed = new DecimalFormat("000").format(Random.nextInt(31)   30)
      }

      // 该 Map 集合用于存放生产出来的数据
      val jsonmMap = new util.HashMap[String, String]()
      jsonmMap.put("monitor_id", randomMonitorId)
      jsonmMap.put("speed", randomSpeed)

      // 因为 kafka 是基于事件的，在此，我们每一条产生的数据都序列化为一个 json 事件
      val event = JSON.toJSON(jsonmMap)

      // 发送事件到 kafka 集群中
      producer.send(new ProducerRecord[String, String](PropertyUtil.getProperty("kafka.topics"), event.toString))

      Thread.sleep(500)

      // 测试
      // println("监测点id："   randomMonitorId   ","   "车速："   randomSpeed)
      println(event)
    }
  }
}

6) 启动集群中的其他相关节点（zookeeper，hadoop 等），启动 kafka，并创建 kafka 主题，检查主题存在性

代码语言：javascript复制
[atguigu@hadoop102 ~]$ start-cluster.sh 

Linux 集群服务群起脚本 (1) 启动脚本：start-cluster.sh

代码语言：javascript复制
#!/bin/bash
echo "================        开始启动所有节点服务      ==========="
echo "================        正在启动 Zookeeper      ==========="
for i in atguigu@hadoop102 atguigu@hadoop103 atguigu@hadoop104
do
    ssh $i 'source /etc/profile;/opt/module/zookeeper-3.4.10/bin/zkServer.sh start'
done

echo "================        正在启动 HDFS           ==========="
ssh atguigu@hadoop102 '/opt/module/hadoop-2.7.2/sbin/start-dfs.sh'

echo "================        正在启动 YARN           ==========="
ssh atguigu@hadoop103 '/opt/module/hadoop-2.7.2/sbin/start-yarn.sh'

echo "================    hadoop102 节点正在启动 JobHistoryServer   ==========="
ssh atguigu@hadoop102 '/opt/module/hadoop-2.7.2/sbin/mr-jobhistory-daemon.sh start historyserver'

(2) 停止脚本：stop-cluster.sh

代码语言：javascript复制
#!/bin/bash
echo "================        开始停止所有节点服务      ==========="
echo "================    hadoop102 节点正在停止 JobHistoryServer ==========="
ssh atguigu@hadoop102 '/opt/module/hadoop-2.7.2/sbin/mr-jobhistory-daemon.sh stop historyserver'

echo "================        正在停止 YARN           ==========="
ssh atguigu@hadoop103 '/opt/module/hadoop-2.7.2/sbin/stop-yarn.sh'

echo "================        正在停止 HDFS           ==========="
ssh atguigu@hadoop102 '/opt/module/hadoop-2.7.2/sbin/stop-dfs.sh'

echo "================        正在停止 Zookeeper      ==========="
for i in atguigu@hadoop102 atguigu@hadoop103 atguigu@hadoop104
do
    ssh $i 'source /etc/profile;/opt/module/zookeeper-3.4.10/bin/zkServer.sh stop'
done

(3) 查看进程脚本：util.sh

代码语言：javascript复制
#!/bin/bash
for i in atguigu@hadoop102 atguigu@hadoop103 atguigu@hadoop104
do
    echo "================      $i 的所有进程       ==========="
    ssh $i '/opt/module/jdk1.8.0_144/bin/jps'
done

尖叫提示：脚本学会之后，如果后续再有新的节点需要添加到群起任务中，可以自行解决之。 尖叫提示：启动与停止注意脚本的执行顺序，而且停止脚本的停止过程应该是启动过程的倒序。

zookeeper 集群群起脚本：

代码语言：javascript复制
[atguigu@hadoop102 ~]$ zkstart.sh

(1) 启动脚本：zkstart.sh

代码语言：javascript复制
#!/bin/bash
echo "==========  正在启动 zookeeper 集群  =========="
for i in atguigu@hadoop102 atguigu@hadoop103 atguigu@hadoop104
do
    ssh $i 'source /etc/profile;/opt/module/zookeeper-3.4.10/bin/zkServer.sh start'
done

(2) 停止脚本：zkstop.sh

代码语言：javascript复制
#!/bin/bash
echo "==========  正在停止 zookeeper 集群  =========="
for i in atguigu@hadoop102 atguigu@hadoop103 atguigu@hadoop104
do
    ssh $i 'source /etc/profile;/opt/module/zookeeper-3.4.10/bin/zkServer.sh stop'
done

(3) 状态脚本：zkstatus.sh

代码语言：javascript复制
#!/bin/bash
echo "==========  正在查看 zookeeper 集群状态  =========="
for i in atguigu@hadoop102 atguigu@hadoop103 atguigu@hadoop104
do
    ssh $i 'source /etc/profile;/opt/module/zookeeper-3.4.10/bin/zkServer.sh status'
done

kafka 集群脚本：

代码语言：javascript复制
[atguigu@hadoop102 ~]$ kafka-start.sh

(1) 启动脚本：kafka-start.sh

代码语言：javascript复制
#!/bin/bash
echo "================        正在启动 Kafka 集群       ==========="
for i in atguigu@hadoop102 atguigu@hadoop103 atguigu@hadoop104
do
    ssh $i 'source /etc/profile;/opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties'
done

(2) 停止脚本：kafka-stop.sh

代码语言：javascript复制
#!/bin/bash
echo "================        正在停止 Kafka 集群       ==========="
for i in atguigu@hadoop102 atguigu@hadoop103 atguigu@hadoop104
do
    ssh $i 'source /etc/profile;/opt/module/kafka/bin/kafka-server-stop.sh -daemon'
done

创建 kafka 主题：traffic

代码语言：javascript复制
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --create --zookeeper hadoop102:2181 --replication-factor 1 --partitions 3 --topic traffic

删除 kafka 主题：traffic

代码语言：javascript复制
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --delete --zookeeper hadoop102:2181 --topic traffic

检查 kafka 的 traffic 主题是否正常：

代码语言：javascript复制
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --list --zookeeper hadoop102:2181
3.1.3 测试

将数据发送至 kafka 并使用 kafka console-consumer 进行检测，持续运行若干分钟后，查看数据是否稳定输入输出。 启动 kafka 控制台消费者：

代码语言：javascript复制
// kafka-console-consumer
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic trafffic

kafka 控制台消费者消费数据如下图所示;

3.2 数据采集

  我们将实时模拟出来的数据，放置于 redis 中。

3.2.1 编写代码

思路：   a) 新建工程：tf_consumer   b) 配置 maven 依赖并添加 scala 框架的支持。   c) 配置 redis 并测试。   d) 将刚才 kafka.properties 以及 PropertyUtil 拷贝过来，并进行相应的修改。   e) 编写 redis 操作工具类：RedisUtil   f) 读取 kafka 中的数据，实时保存到 redis 中，并且按照分钟和监测点聚合车速和车辆个数。

1) 新建工程：tf_consumer

2) 配置 maven 的 pom.xml 文件以及 kafka.properties： pom.xml：

代码语言：javascript复制
<?xml version="1.0" encoding="UTF-8"?>
<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/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.atguigu</groupId>
    <artifactId>tf_consumer</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.11.0.2</version>
        </dependency>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.41</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka_2.11</artifactId>
            <version>1.6.3</version>
        </dependency>

        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.9.0</version>
        </dependency>
    </dependencies>
</project>

3) 修改 kafka.properties 配置文件（消费者）：

代码语言：javascript复制
# 设置 kafka 的 brokerlist
bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092

# 消费者反序列化
key.deserializer=org.apache.kafka.common.serialization.StringDeSerializer
value.deserializer=org.apache.kafka.common.serialization.StrinDegSerializer

acks=all
retries=0

# Kafka 老版本中的元数据服务列表
metadata.broker.list=hadoop102:9092,hadoop103:9092,hadoop104:9092

# 设置消费者所属的消费组
group.id=g_traffic1

# 设置是否自动确认 offset
enable.auto.commit=true

# 设置自动确认 offset 的时间间隔
auto.commit.interval.ms=30000

# 设置本次消费的主题
kafka.topics=traffic

# 设置 zookeeper 中 follower 和 leader 之间的关于 kafka 的信息同步时间间隔
zookeeper.sync.time.ms=250
num.io.threads=12
batch.size=65536
buffer.memory=524288

# kafka 中消息保存的时间（单位是小时），企业开发中是 7 天
log.retention.hours=2

3) 配置 Redis（单节点）环境并测试

代码语言：javascript复制
// 通过 wget 下载 Redis 的源码
[atguigu@hadoop102 software]$ wget http://download.redis.io/releases/redis-4.0.2.tar.gz

// 将源代码解压到指定目录 /opt/module 下
[atguigu@hadoop102 software]$ tar -zxf redis-4.0.2.tar.gz -C /opt/module

// 进入 Redis 源代码目录，编译安装（因为 redis 是用 C 语言写的）
[atguigu@hadoop102 module]$ cd redis-4.0.2/

// 安装 GCC
[atguigu@hadoop102 module]$ sudo yum install gcc

// 编译源代码
[atguigu@hadoop102 redis-4.0.2]$ make MALLOC=libc

如果报错
zmalloc.h:50:31: error: jemalloc/jemalloc.h: No such file or directory
zmalloc.h:55:2: error: #error "Newer version of jemalloc required"
make[1]: *** [adlist.o] Error 1
make[1]: Leaving directory `/opt/module/redis-4.0.2/src'
make: *** [all] Error 2
解决办法是：
make MALLOC=libc

注意：Redis 并没有自己实现内存池，没有在标准的系统内存分配器上再加上自己的东西。
redis-2.4 以上自带 jemalloc，你不需要加任何参数，通过 zmalloc.c 源码中我们可以看到，Redis 在编译时，会先判断是否使用 tcmalloc，如果是，会用 tcmalloc 对应的函数替换掉标准的 libc 中的函数实现。其次会判断 jemalloc 是否使用，最后如果都没有使用才会用标准的 libc 中的内存管理函数。所以用 tcmalloc 优化请谨慎使用，这两个分配器碎片率相差不大，建议用自带 jemalloc。

如果要安装 tcmalloc 可以这样：
make USE_TCMALLOC=yes

// 编译安装（注意：要使用 root 用户权限）
[atguigu@hadoop102 redis-4.0.2]$ sudo make install

// 创建配置文件，放入指定的目录
[atguigu@hadoop102 redis-4.0.2]$ sudo cp /opt/module/redis-4.0.2/redis.conf /opt/module/redis-4.0.2/myredis

// 修改配置文件中以下内容（注意 redis 新版的 4.x 与 老版本 3.x 上配置的细微差别）
[atguigu@hadoop102 redis-4.0.2]$ sudo vim /opt/module/redis-4.0.2/myredis/redis.conf

bind 0.0.0.0                                            #69行       #绑定主机 IP，默认值为127.0.0.1，我们是跨机器运行，所以需要更改，表示任意机器集群均可访问，实际开发是中不建议这样改
daemonize yes                                           #136行      #是否以后台 daemon 方式运行，默认不是后台运行
pidfile /var/run/redis/redis_6379.pid                   #158行      #redis 的 PID 文件路径（可选）
logfile "/opt/module/redis-4.0.2/myredis/redis.log"     #171行      #定义 log 文件位置，模式 log 信息定向到 stdout，输出到 /dev/null（可选）
dir "/opt/module/redis-4.0.2/myredis"                   #263行      #本地数据库存放路径，默认为./（可选）

// 编译安装默认存在在 /usr/local/bin 目录下，如下
[atguigu@hadoop102 redis-4.0.2]$ cd /usr/local/bin/
[atguigu@hadoop102 bin]$ ll
总用量 9572
-rw-r--r-- 1 root root      83 5月   8 01:27 dump6379.rdb
-rw-r--r-- 1 root root      83 5月   8 01:27 dump6380.rdb
-rw-r--r-- 1 root root      83 5月   8 01:27 dump6381.rdb
lrwxrwxrwx 1 root root       6 4月  28 17:17 nc -> netcat
-rwxr-xr-x 1 root root  103479 4月  28 17:17 netcat
-rwxr-xr-x 1 root root  290454 5月  23 12:37 redis-benchmark
-rwxr-xr-x 1 root root 2971304 5月  23 12:37 redis-check-aof
-rwxr-xr-x 1 root root   45443 5月   6 17:27 redis-check-dump
-rwxr-xr-x 1 root root 2971304 5月  23 12:37 redis-check-rdb
-rwxr-xr-x 1 root root  419907 5月  23 12:37 redis-cli
lrwxrwxrwx 1 root root      12 5月  23 12:37 redis-sentinel -> redis-server
-rwxr-xr-x 1 root root 2971304 5月  23 12:37 redis-server

在安装完 Redis 之后，启动 Redis

代码语言：javascript复制
// 启动 Redis 服务器
[atguigu@hadoop102 redis-4.0.2]$ redis-server /opt/module/redis-4.0.2/myredis/redis.conf

// 连接 Redis 服务器
[atguigu@hadoop102 redis-4.0.2]$ redis-cli -h 192.168.25.102 -p 6379
192.168.25.102:6379> set k1 123
OK
192.168.25.102:6379> get k1
"123"
192.168.25.102:6379> keys *
1) "k1"
2) "uid:2"
192.168.25.102:6379> lrange uid:2 0 -1      #查看列表的某个范围的数据
1) "150:5.0"
2) "144:3.0"
3) "110:4.0"
192.168.25.102:6379> lpush uid:1 3671:3.0 2968:1.0 2455:2.5     #存一组列表数据
192.168.25.102:6379> flushall       #清空所有数据
192.168.25.102:6379> select 1       #选择数据库

// 查看 Redis 的启动情况
[atguigu@hadoop102 redis-4.0.2]$ ps -ef | grep redis
atguigu    6033      1  0 13:08 ?        00:00:00 redis-server 0.0.0.0:6379                              
atguigu    6046   4336  0 13:12 pts/0    00:00:00 grep redis 

// 停止 Redis 服务器
[atguigu@hadoop102 redis-4.0.2]$ redis-cli shutdown

4) 将刚才 kafka.properties 以及 PropertyUtil 拷贝过来，kafka.properties 需要进行相应的修改 5) 编写 redis 操作工具类：RedisUtil.scala

代码语言：javascript复制
package com.atguigu.utils

import redis.clients.jedis._

// 代码写在半生对象中，这些代码会在类加载的时候，自动的进行初始化
object RedisUtil {
  // 配置 redis 基本连接参数
  val host = "192.168.25.102"
  val port = 6379
  val timeout = 30000

  val config = new JedisPoolConfig

  // 设置连接池允许最大的连接个数
  config.setMaxTotal(200)
  // 设置最大空闲连接数
  config.setMaxIdle(50)
  // 设置最小空闲连接数
  config.setMinIdle(8)

  // 设置连接时的最大等待的毫秒数
  config.setMaxWaitMillis(10000)
  // 设置在获取连接时，检查连接的有效性
  config.setTestOnBorrow(true)
  // 设置在释放连接时，检查连接的有效性
  config.setTestOnReturn(true)

  // 设置在连接空闲时，检查连接的有效性
  config.setTestWhileIdle(true)

  // 设置两次扫描之间的时间间隔毫秒数
  config.setTimeBetweenEvictionRunsMillis(30000)
  // 设置每次扫描的最多的对象数
  config.setNumTestsPerEvictionRun(10)
  // 设置逐出连接的最小时间间隔，默认是 1800000 毫秒 = 30 分钟
  config.setMinEvictableIdleTimeMillis(60000)

  //  连接池
  lazy val pool = new JedisPool(config, host, port, timeout)

  // 释放资源
  lazy val hook = new Thread{ // 钩子函数：执行一些善后操作，正常退出
    override def run() = {
      pool.destroy()
    }
  }

  sys.addShutdownHook(hook.run())
}

6) 在 SparkConsumer.scala 中读取 kafka 中的数据，实时保存到 redis 中，并且按照分钟和监测点聚合车速和车辆个数。用到 Spark Streaming 的时间窗口函数进行聚合。

代码语言：javascript复制
package com.atguigu.consumer

import java.text.SimpleDateFormat
import java.util.Calendar

import com.alibaba.fastjson.{JSON, TypeReference}
import com.atguigu.utils.{PropertyUtil, RedisUtil}
import kafka.serializer.StringDecoder
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 堵车预测：处理实时数据，消费数据到 redis
  */
object SparkConsumer {
  def main(args: Array[String]): Unit = {
    // 初始化 Spark
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("TrafficStreaming")
    val sc = new SparkContext(sparkConf)
    val ssc = new StreamingContext(sc, Seconds(5))

    // 设置检查点目录
    ssc.checkpoint("./ssc/checkpoint")

    // 配置 kafka 参数，使用的是 spark 为我们封装的一套操作 kafka coonsumer 的工具包
    val kafkaParam = Map("metadata.broker.list" -> PropertyUtil.getProperty("metadata.broker.list"))

    // 配置 kafka 主题
    val topics = Set(PropertyUtil.getProperty("kafka.topics"))

    // 读取 kafka 主题 中的每一个事件 event
    val kafkaLineDStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParam, topics)
      .map(_._2) // 由于我们 event 中的键是 null，所以我们需要把值映射出来

    // 解析 json 字符串
    val event = kafkaLineDStream.map(line => { // {"monitor_id":"0001","speed":"038"}
      // 使用 fastjson 来解析当前事件中封装的数据信息，由于该 json 字符串不支持 Scala Map，所以需要先将 json 字符串解析为 Java Map
      val lineJavaMap = JSON.parseObject(line, new TypeReference[java.util.Map[String, String]]() {})
      // 将 Java Map 转换成 Scala Map
      import scala.collection.JavaConverters._
      val lineScalaMap: collection.mutable.Map[String, String] = mapAsScalaMapConverter(lineJavaMap).asScala
      println(lineScalaMap) // Map[String, String] = ("monitor_id" -> "0001", "speed" -> "038")
      lineScalaMap
    })

    // 将每一条数据根据 monitor_id 聚合，聚合每一条数据中的 “车辆速度” 叠加
    // 例如：聚合好的数据形式：(monitor_id, (speed, 1))  ("0001", (038, 1))
    // 最終結果举例：("0001", (1365, 30))
    val sumOfSpeedAndCount = event
      .map(e => (e.get("monitor_id").get, e.get("speed").get)) // ("0001", "038")、("0001", "048")、("0002", "015")
      .mapValues(s => (s.toInt, 1)) // ("0001", (038, 1))、("0001", (048, 1))、("0002", (015, 1))
      .reduceByKeyAndWindow( // reduce 表示从左边开始执行将得到的结果返回给第一个参数
      (t1: (Int, Int), t2: (Int, Int)) => (t1._1   t2._1, t1._2   t2._2),
      Seconds(60), // 滑动窗口大小 60 秒，误差最大 59 秒，即上一分钟的数据当成下一分钟的数据来用了。
      Seconds(60)) // 滑动步长 60 秒，对我们实际建模的影响忽略不计，因为：实际中，不可能1分钟内就造成大量拥堵，或者堵车不可能1分钟之内就缓解了！！！后面建模的时候会进行线性滤波。

    // 定义 redis 数据库中的数据库索引 index
    val dbIndex = 1
    // 将采集到的数据，按照每分钟放置于redis 中，将用于后边的数据建模
    sumOfSpeedAndCount.foreachRDD(rdd => {
      rdd
        .foreachPartition(partitionRecords => {
          partitionRecords
            .filter((tuple: (String, (Int, Int))) => tuple._2._1 > 0) // 过滤掉元组数据中的速度小于0的数据
            .foreach(pair => {
            // 开始取出这 60 秒的 windows 中所有的聚合数据进行封装，准备存入 redis 数据库
            val jedis = RedisUtil.pool.getResource

            val monitorId = pair._1
            val sumOfCarSpeed = pair._2._1
            val sumOfCarCount = pair._2._2

            // 模拟数据为实时流入
            // 两种情况：
            // 1、数据生产时，会产生时间戳字段，流入到 kafka 的事件中
            // 2、数据消费时，数据消费的时间，就当做数据的生产时间（会有一些小小误差），本业务选择这种方式

            val dateSDF = new SimpleDateFormat("yyyyMMdd") // 用于 redis 中的 key
            val hourMinuteSDF = new SimpleDateFormat("HHmm") // 用于 redis 中的 fields

            val currentTime = Calendar.getInstance().getTime

            val dateTime = dateSDF.format(currentTime) // 20190528
            val hourMinuteTime = hourMinuteSDF.format(currentTime) // 1617

            // 选择存入的数据库
            jedis.select(dbIndex)
            jedis.hset(dateTime   "_"   monitorId, hourMinuteTime, sumOfCarSpeed   "_"   sumOfCarCount)

            println(dateTime   "_"   monitorId, hourMinuteTime, sumOfCarSpeed   "_"   sumOfCarCount)

            // RedisUtil.pool.returnResource(jedis) // 老的 API
            jedis.close() // 新的 API
          })
        })
    })

    // Spark 开始工作
    ssc.start()
    ssc.awaitTermination()
  }
}

// 复习 Scala 中 Map 的取值方式：

// 方式1-使用 map(key)
//   1、如果 key 存在，则返回对应的值。
//   2、如果 key 不存在，则抛出异常 [java.util.NoSuchElementException]。
//   3、在 Java 中，如果 key 不存在则返回 null。
// 方式2-使用 contains 方法检查是否存在 key
//  使用 containts 先判断再取值，可以防止异常，并加入相应的处理逻辑。
//   1、如果 key 存在，则返回 true。
//   2、如果 key 不存在，则返回 false。
// 方式3-使用 map.get(key).get 取值
//   1、如果 key 存在，则 map.get(key) 就会返回 Some(值)，然后 Some(值).get 就可以取出。
//   2、如果 key 不存在，则 map.get(key) 就会返回 None。
// 方式4-使用 map.getOrElse(key, defaultvalue) 取值
//   底层是：def getOrElse[V1 >: V](key: K, default: => V1)
//   1、如果 key 存在，则返回 key 对应的值。
//   2、如果 key 不存在，则返回默认值。在 java 中底层有很多类似的操作。
// 如何选择取值方式建议
//   如果我们确定 map 有这个 key，则应当使用 map(key)，速度快。
//   如果我们不能确定 map 是否有 key，而且有不同的业务逻辑，使用 map.contains() 先判断再加入逻辑。
//   如果只是简单的希望得到一个值，使用 map4.getOrElse("ip", "127.0.0.1")
3.2.2 测试

我们使用集群的群起脚本：

开启 zookeeper 集群：

代码语言：javascript复制
[atguigu@hadoop102 ~]$ zkstart.sh

开启 kafka 集群：

代码语言：javascript复制
[atguigu@hadoop102 ~]$ kafka-start.sh

开启 redis，在 redis 根目录执行：

代码语言：javascript复制
// 启动 Redis 服务器
[atguigu@hadoop102 redis-4.0.2]$ redis-server /opt/module/redis-4.0.2/myredis/redis.conf

运行数据生产 运行数据消费 查看运行结果： 在 redis 根目录中，举个例子依次执行：

代码语言：javascript复制
[atguigu@hadoop102 redis-4.0.2]$ redis-cli -h 192.168.25.102 -p 6379
192.168.25.102:6379> select 1
OK
192.168.25.102:6379[1]> keys *
 1) "20190528_0014"
 2) "20190528_0005"
 3) "20190528_0019"
 4) "20190528_0009"
 5) "20190528_0004"
 6) "20190528_0013"
 7) "20190528_0016"
 8) "20190528_0020"
 9) "20190528_0015"
10) "20190528_0010"
11) "20190528_0018"
12) "20190528_0008"
13) "20190528_0001"
14) "20190528_0003"
15) "20190528_0007"
16) "20190528_0012"
17) "20190528_0002"
18) "20190528_0011"
19) "20190528_0017"
20) "20190528_0006"
192.168.25.102:6379[1]> hgetall 20190528_0001
 1) "1646"
 2) "279_7"
 3) "1647"
 4) "239_6"
 5) "1648"
 6) "240_5"
 7) "1649"
 8) "318_7"
 9) "1650"
10) "184_6"
11) "1651"
12) "54_8"
13) "1652"
14) "81_10"
15) "1653"
16) "69_9"
17) "1654"
18) "69_9"
19) "1655"
20) "57_8"
21) "1656"
22) "262_6"
23) "1657"
24) "149_3"
25) "1659"
26) "168_4"
27) "1700"
28) "134_4"
29) "1701"
30) "65_8"
31) "1702"
32) "81_10"

注意：不要直接复制，每次操作有些内容是有变动的。比如时间相关的，比如 IP 相关的。

小结：

代码语言：javascript复制
堵车内容回顾：
一、数据生产
    目的：能够让我们清楚数据结构是什么样子的，实际开发中这部分不是我们做；实际开发中：已有数据结构，已有目标，要做的就是目前手中已有的资料如何实现目标
    数据结构：卡口id，车速（没有包含数据生产时的时间戳）
    堵车状态的转换逻辑（if else），为的是生产的数据尽可能的贴近现实情况

二、数据消费
    kafka（高级 API,spark 提供的工具包） --> redis
    时间窗口的大小为 60 秒
    时间窗口的滑动步长为 60 秒
    数据存储在 redis 中，使用的是数据类型是 Hash(即 Map 类型)：KV 模式不变，但是 V 也是一个键值对
        key : 20190528_0001
        field : 1754
        value : 1365_30

天猫双十一（使用 Storm   Flink 实现）
1、如果我们使用 SparkStreaming 实现，时间窗口的宽度不能设置太大，可能会出现内存溢出。
2、5秒内聚合的数据该如何处理呢？答：保存到 redis 中（即落盘）。
3、那么下一个时间窗口的新的数据该如何处理呢？答：先将 redis 中前一个 5 秒的数据读出来，然后和这次的 5 秒数据进行相加后，再放回到 redis 中（即落盘）。小结：所有的流式框架都是这样做的。

流式框架的根本的哲学意义是：仅仅处理中间逻辑，即是进行运算(计算)的，不负责数据存储的。
如果在内存中想进行长期的数据累加，就相当于一个不断微分再积分的过程，把时间微分到足够细，细到不会导致内存溢出为止，然后再微分的基础上求和，再把所有的微分结果进行积分。

某一个小时间段内的数据量越小，则时间窗口的宽度就可以设置的越大，那么数据展示的延迟就会变长，但是整体数据处理的效率就会变得越高。--> 不像流式处理了！

3.3 数据建模

  在此我们选择通过有监督学习中的手段建立可以预测下一时刻堵车状态的模型。 拟牛顿图解：

线性滤波图解：

目标卡口与相关卡口关系：

模型图解：

3.3.1 编写代码

思路：   a) 确定要对哪个监测点进行建模，我们称之为目标监测点。   b) 找到目标监测点的其他相关监测点（比如相关监测点与目标监测点属于一条公路的）。   c) 从 redis 中访问得到以上所有监测点若干小时内的历史数据信息（一部分作为训练数据，一部分作为测试数据）。   d) 提取组装特征向量与目标向量，训练参数集，训练模型。   e) 测试模型吻合度，将符合吻合度的模型保存到 HDFS 中，同时将模型的保存路径放置于 redis 中。

1) 新建 module：tf_modeling 2) 编写 maven 的 pom.xml 文件，添加 scala 框架支持

代码语言：javascript复制
<?xml version="1.0" encoding="UTF-8"?>
<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/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.atguigu</groupId>
    <artifactId>tf_modeling</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-mllib_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>

        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.9.0</version>
        </dependency>
    </dependencies>
</project>

3) 创建 Train.scala 实现上述思路：

代码语言：javascript复制
package com.atguigu.train

import java.io.{File, PrintWriter}
import java.text.SimpleDateFormat
import java.util.{Calendar, Date}

import com.atguigu.utils.RedisUtil
import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS
import org.apache.spark.mllib.evaluation.MulticlassMetrics
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

/**
  * 堵车预测：建模，不同的卡口不同的模型（函数）
  */
object Train {
  def main(args: Array[String]): Unit = {
    // 写入文件的输出流，将本次评估结果保存到下面这个文件中
    val writer = new PrintWriter(new File("model_train.txt"))

    // 初始化 Spark
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("TrafficTrainModel")
    val sc = new SparkContext(sparkConf)

    // 定义 redis 的数据库相关参数
    val dbIndex = 1
    // 获取 redis 连接
    val jedis = RedisUtil.pool.getResource
    jedis.select(dbIndex)

    // 设定 目标监测点：你要对哪几个监测点进行建模（本例中对 2 个检测点进行建模）
    val targetMonitorIDs = List("0005", "0015")
    // 取出 目标监测点的相关监测点：算法工程师告诉我们的（本例中我们随意写几个）
    val relationMonitors = Map[String, Array[String]](
      "0005" -> Array("0003", "0004", "0005", "0006", "0007"),
      "0015" -> Array("0013", "0014", "0015", "0016", "0017")
    )

    // 访问 redis 取出 目标监测点的相关监测点 的数据

    // 遍历 目标监测点的相关监测点 的 Map 集合
    targetMonitorIDs.map(targetMonitorID => { // 这个 map 执行 2 次
      // 初始化时间
      // 获取当前时间
      val currentDate = Calendar.getInstance().getTime

      // 格式化 当前时间 为 年月日 对象
      val dateSDF = new SimpleDateFormat("yyyyMMdd")
      // 格式化 当前时间 为 小时分钟数 对象
      val hourMinuteSDF = new SimpleDateFormat("HHmm")

      // 格式化当前时间
      val dateSDFString = dateSDF.format(currentDate) // 20190528

      // 获取 当前目标监测点的相关监测点
      val relationMonitorArray = relationMonitors(targetMonitorID)
      // 根据 当前目标监测点的相关监测点，取出当前时间的所有相关监测点的平均车速
      val relationMonitorInfo = relationMonitorArray.map(relationMonitorID => { // 这个 map 执行 5 次
        (relationMonitorID, jedis.hgetAll(dateSDFString   "_"   relationMonitorID))
        // ("0003", {"0900":"1356_30", "0901":"100_2", ..., "0959":"134_4"})
      })

      // 创建 3 个数组：因为要使用 拟牛顿法(LBFGS)进行建模，该方法需要
      // 第一个数组放 特征因子数据集，
      // 第二个数组放 label 标签向量（特征因子对应的结果数据集），
      // 第三个数组放 前两者之间的关联（即真正的特征向量）
      val dataX = ArrayBuffer[Double]() // 实际的每一分钟的平均车速
      val dataY = ArrayBuffer[Double]() // 第 4 分钟的平均车速

      // 用于存放 特征因子数据集 和 特征因子对应的结果数据集 的映射关系
      val dataTrain = ArrayBuffer[LabeledPoint]()

      // 确定使用多少时间内的数据进行建模（本例中取 1 小时）
      val hours = 1

      // 将时间回退到当前时间的 1 小时之前，时间单位：分钟
      // 遍历 目标监测点的数据（外循环）
      for (i <- Range(60 * hours, 2, -1)) { // 本例中是 60 到 2(不包括2)，步长是 -1，即 60, 59, 58, ..., 5, 4,
        dataX.clear()
        dataY.clear()

        // 遍历 目标监测点的所有相关监测点 的数据（内循环）
        for (index <- 0 to 2) {
          // 当前for循环 的时间 = 当前时间的毫秒数 - 1 个小时的毫秒数   0分钟的毫秒数，1分钟的毫秒数，2分钟的毫秒数  （第3分钟作为监督学习的结果向量--label 向量）
          val oneMoment = currentDate.getTime - 60 * i * 1000   60 * index * 1000
          // 获取 当前for循环 的时间的小时分钟数
          val oneHM = hourMinuteSDF.format(new Date(oneMoment))

          // 获取当前小时分钟数的数据
          for ((k, v) <- relationMonitorInfo) { // ("0003", {"0900":"1356_30", "0901":"100_2", ..., "0959":"134_4"})

            // hours 个小时前的后 3 分钟的数据，组装到 dataX 中
            if (v.containsKey(oneHM)) { // 判断本次时刻的数据是否存在，如果存在，则取值，否则，则取 -1(表示数据缺失)
              val speedAndCarCount = v.get(oneHM).split("_")
              val valueX = speedAndCarCount(0).toFloat / speedAndCarCount(1).toFloat // 得到当前这一分钟的平均车速
              dataX  = valueX
            } else {
              dataX  = -59.0F
            }

            // 如果 index == 2，说明已经得到 hours 个小时前的后 3 分钟的数据，并组装到了 dataX 中；如果是目标卡口，则说明下一分钟数据是 label 向量的数据，ze存放 dataY 中
            if (index == 2 && targetMonitorID == k) {
              val nextMoment = oneMoment   60 * 1000
              val nextHM = hourMinuteSDF.format(new Date(nextMoment))
              if (v.containsKey(nextHM)) { // 判断本次时刻的数据是否存在，如果存在，则取值，否则，则不管它（有默认值 0）
                val speedAndCarCount = v.get(nextHM).split("_")
                val valueY = speedAndCarCount(0).toFloat / speedAndCarCount(1).toFloat // 得到第 4 分钟的平均车速
                dataY  = valueY
              }
            }

          }
        }

        // 准备训练模型
        // 先将 dataX 和 dataY 映射到一个 LabeledPoint 对象中
        if (dataY.toArray.length == 1) { // 说明结果集中有数据了
          val label = dataY.toArray.head
          val record = LabeledPoint(
            // 因为使用的是 拟牛顿法(LBFGS) 进行建模，该方法需要 特征结果 有几种情况（不能是无穷种情况）
            // label 范围为 0~6（7个类别），越大则道路越通畅
            if (label / 10 < 6) (label / 10).toInt else 6, Vectors.dense(dataX.toArray)
          )
          dataTrain  = record
        }
      }

      // 将特征数据集写入到文件中方便查看，至此，我们的特征数据集已经封装完毕
      dataTrain.foreach(record => {
        println(record)
        writer.write(record.toString()   "rn")
      })

      // 将特征数据集转为 rdd 数据集
      val rddData = sc.parallelize(dataTrain)
      // 随机封装训练集和测试集
      val randomSplits = rddData.randomSplit(Array(0.6, 0.4), 11L)
      val trainData = randomSplits(0)
      val testData = randomSplits(1)

      if (!rddData.isEmpty()) {
        // 使用训练数据集进行训练模型
        val model = new LogisticRegressionWithLBFGS().setNumClasses(7).run(trainData)

        // 使用测试数据集测试训练好的模型
        val predictAndLabel = testData.map {
          case LabeledPoint(label, feature) =>
            val predict = model.predict(feature)
            (predict, label)
        }

        // 得到当前 目标监测点 的评估值
        val metrics = new MulticlassMetrics(predictAndLabel)
        val accuracy = metrics.accuracy
        println("评估值："   accuracy)
        writer.write(accuracy.toString   "rn")

        // 设置评估阈值，评估值范围为[0.0, 1.0]，越大 model 越优秀，我们保存评估值大于 0 的评估模型
        if (accuracy > 0.6) {
          // 将模型保存到 hdfs 中，并将模型路径保存到 redis 中
          val hdfsPath = "hdfs://hadoop102:9000/traffic/model/"   targetMonitorID   "_"   new SimpleDateFormat("yyyyMMddHHmmss").format(new Date
          (currentDate.getTime))
          model.save(sc, hdfsPath)

          jedis.hset("model", targetMonitorID, hdfsPath)
        }
      }
    })

    // 释放 redis 连接
    // RedisUtil.pool.returnResource(jedis) // 老的 API
    jedis.close() // 新的 API
    writer.close()
  }
}
3.3.2 测试

  运行数据模拟与数据采集，等待一会之后，开始进行预测，查看 http://hadoop102:50070 中是否产生对应的模型样本。同时查看 redis 中是否有保存训练好的模型存放路径。

3.4 数据预测

3.4.1 编写代码

思路：   a) 用户传入想要进行预测的时间节点，读取该时间节点之前 3 分钟，2 分钟和 1 分钟的数据。   b) 此时应该已经得到了历史数据集，通过该历史数据集预测传入时间点的车流状态。   尖叫提示：为了方便观察测试，建议传一个历史时间点，这样可以很直观的看到预测结果是否符合期望值。

1) 新建 module：tf_prediction 2) 配置 maven 的 pom.xml 文件

代码语言：javascript复制
<?xml version="1.0" encoding="UTF-8"?>
<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/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.atguigu</groupId>
    <artifactId>tf_prediction</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-mllib_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>

        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.9.0</version>
        </dependency>
    </dependencies>
</project>

3) 新建 Prediction.scala 文件，实现上述思路

代码语言：javascript复制
package com.atguigu.predict

import java.text.SimpleDateFormat
import java.util.Date

import com.atguigu.utils.RedisUtil
import org.apache.spark.mllib.classification.LogisticRegressionModel
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

/**
  * 堵车预测：根据训练出来的模型进行堵车预测
  */
object Prediction {
  def main(args: Array[String]): Unit = {
    // 初始化 Spark
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("TrafficPrediction")
    val sc = new SparkContext(sparkConf)

    // 时间设置：为了拼凑出 redis 中的 key 和 field 的字段

    // 格式化 时间 为 年月日 对象
    val dateSDF = new SimpleDateFormat("yyyyMMdd")
    // 格式化 时间 为 小时分钟数 对象
    val hourMinuteSDF = new SimpleDateFormat("HHmm")

    // 2019-05-29 13:00
    val userSDF = new SimpleDateFormat("yyyy-MM-dd HH:mm")

    // 定义用户传入的日期：想要预测是否堵车的日期
    val inputDateString = "2019-05-29 10:29"
    val inputDate = userSDF.parse(inputDateString)

    // 得到 redis 中的 key
    val dateSDFString = dateSDF.format(inputDate) // 20180529

    val dbIndex = 1
    val jedis = RedisUtil.pool.getResource
    jedis.select(dbIndex)

    // 想要预测的监测点
    // 设定 目标监测点：你要对哪几个监测点进行建模（本例中对 2 个检测点进行建模）
    val targetMonitorIDs = List("0005", "0015")
    // 取出 目标监测点的相关监测点：算法工程师告诉我们的（本例中我们随意写几个）
    val relationMonitors = Map[String, Array[String]](
      "0005" -> Array("0003", "0004", "0005", "0006", "0007"),
      "0015" -> Array("0013", "0014", "0015", "0016", "0017")
    )

    // 遍历 目标监测点的相关监测点 的 Map 集合
    targetMonitorIDs.map(targetMonitorID => { // 这个 map 执行 2 次
      // 获取 当前目标监测点的相关监测点
      val relationMonitorArray = relationMonitors(targetMonitorID)
      // 根据 当前目标监测点的相关监测点，取出当前时间的所有相关监测点的平均车速
      val relationMonitorInfo = relationMonitorArray.map(relationMonitorID => { // 这个 map 执行 5 次
        (relationMonitorID, jedis.hgetAll(dateSDFString   "_"   relationMonitorID))
        // ("0003", {"0900":"1356_30", "0901":"100_2", ..., "0959":"134_4"})
      })

      // 装载目标时间点之前的 3 分钟的历史数据
      val dataX = ArrayBuffer[Double]() // 实际的每一分钟的平均车速

      // 组装数据
      for (index <- Range(3, 0, -1)) {
        val oneMoment = inputDate.getTime - 60 * index * 1000
        val oneHM = hourMinuteSDF.format(new Date(oneMoment)) // 1257

        for ((k, v) <- relationMonitorInfo) {
          if (v.containsKey(oneHM)) {
            val speedAndCarCount = v.get(oneHM).split("_")
            val valueX = speedAndCarCount(0).toFloat / speedAndCarCount(1).toFloat
            dataX  = valueX
          } else {
            dataX  = -59.0F
          }
        }
      }

      // 加载模型
      val modelPath = jedis.hget("model", targetMonitorID)
      val model = LogisticRegressionModel.load(sc, modelPath)

      // 进行预测
      val predict = model.predict(Vectors.dense(dataX.toArray))

      // 打印展示
      println(targetMonitorID   "，堵车评估值："   predict   "，是否通畅："   (if (predict > 3) "通畅" else "拥堵"))

      // 结果保存
      jedis.hset(inputDateString, targetMonitorID, predict.toString)
    })

    // 释放 redis 连接
    // RedisUtil.pool.returnResource(jedis) // 老的 API
    jedis.close() // 新的 API
  }
}
3.4.2 测试

  预测任务执行完毕后，进入redis，通过查看对应监测点，对应传入时间节点的具体车速值，来验证预测结果是否正确。

四 项目总结

  与该项目类似的需求还有很多很多，涵盖了生活各个方面。不同的业务，不同的逻辑，不同的思路，不同的数学模型，需要具体情况具体分析。有类似的、但不完全一样的需求，就需要多思考，灵活处理了。

kafka bash bash指令 scala 腾讯云测试服务

0 人点赞