一、项目简介
这是一个车辆监控项目。主要实现了三个功能：
1.计算每一个区域车流量最多的前3条道路。
2.计算道路转换率
3.实时统计道路拥堵情况（当前时间，卡口编号，车辆总数，速度总数，平均速度）
二、项目结构
├─TrafficBySparkAndKafka
├─data
└─src
├─main
│ ├─java
│ │ └─vip
│ │ ├─producedate2hive（模拟数据到文件和Hive）
│ │ ├─shuai7boy
│ │ │ └─trafficTemp
│ │ │ ├─areaRoadFlow（每个区域top3道路速度统计。道路转换率。）
│ │ │ ├─conf （获取配置文件帮助类）
│ │ │ ├─constant （接口静态类，防止硬编码）
│ │ │ ├─dao
│ │ │ │ ├─factory （工厂类）
│ │ │ │ └─impl （接口实现类）
│ │ │ ├─domain （属性定义类）
│ │ │ ├─jdbc （jdbc帮助类）
│ │ │ ├─rtmroad（实时统计道路拥堵情况）
│ │ │ ├─skynet
│ │ │ └─util （帮助类）
│ │ └─spark
│ │ └─spark
│ │ └─test （模拟实时数据）
│ ├─resources
│ └─scala
│ └─top
│ └─shuai7boy
│ └─trafficTemp
│ └─areaRoadFlow （利用scala和java互调用，实现top3道路速度统计）
│
└─test
└─java
三、数据源
数据源类型：
monitor_flow_action(每个摄像头的监控数据)
当天日期 卡口编号 摄像头编号 车牌号 拍摄时间 车速 道路编号 区域编号
2020-05-08 0001 34287 京M80025 2020-05-08 05:35:58 57 25 03
2020-05-08 0005 99132 京M80025 2020-05-08 05:51:28 149 50 04
monitor_camera_info（卡口和摄像头对应编号）
0006 00443
0006 25745
0006 98681
0006 36400
存储介质：
如果在本地运行的话，这里读取的是本地文件。
如果在集群运行，对于批处理读取的是Hive，对于流处理这里读取Kafka。
四、数据转换流程
1.计算每一个区域车流量最多的前3条道路。
-
从表traffic.monitor_flow_action根据日期获取车流量监控日志信息。
挡在集群中时，traffic.monitor_flow_action代表的是hive中的表，当在本地运行时，traffic.monitor_flow_action是本地创建的临时表。
-
从area_info表中获取区域信息。
area_info是MySql中的表。
-
根据步骤二获取的区域信息，补全监控日志名称。根据join，map即可拼接一个新的RDD，并将RDD转换为DataFrame的临时表tmp_car_flow_basic。
-
统计各个区域的道路车流量。
使用Spark SQL根据区域名称，道路ID进行分组。即可统计出每个区域，每条道路对应的车流量。
-
统计每个区域top3车流量。
利用开窗函数进行统计。row_number() over(partition by area_name order by road_id desc)
用到的技术：Hive，Spark SQL，临时表，MySql，JDBC，join，map，RDD转换DataFrame。
2.计算道路转换率
-
从MySql拿出我们要对比的转换路段
-
从日志拿出指定日期的监控数据
-
将监控数据转换为键值对（car，row）格式
-
计算每个路段的匹配情况。
逻辑：将第三步拿到的数据，根据car进行分组，映射键值对。将轨迹信息根据时间进行排序，然后拼接。
将我们指定的路段（第一步获取到的）和上面拼接的数据进行比对，得出匹配情况。（路段，匹配次数）
-
因为上面求的是多辆车的 （路段，匹配次数）。这步使用reduceByKey进行聚合，将相同路段进行汇总。
-
获取转化率。
转换率=（这次路段的匹配度）/（上次路段的匹配度）即可得到。、
这次路段的匹配度=（聚合数据.get(路段)）
用到的技术：mapToPair，groupByKey，flatMapToPair（进来一辆车，出去多个对应路段信息），reduceByKey。
3.实时统计道路拥堵情况（根据车辆和车速判断）
- 根据日志获取（卡口ID，汽车速度）格式数据
- 获取（卡口ID，（汽车速度，1））格式数据，后面的1代表车辆数
- 获取（卡口ID，（汽车总速度，总车辆数））
- 打印车辆（卡口，总速度，总车辆，平均速度）
用到的技术:map,mapToPair,mapValues(仅仅针对value进行map，(key,(value,1))格式数据)，reduceByKeyAndWindow。