/TrafficTeach

大数据项目:车流量监控

Primary LanguageJava

车流量监控

前言

目的 1.对业务场景抽象,熟练Spark编码 2.增加自定义累加器,自定义UDF 3.Spark 优化方式

项目 数据处理架构

模块介绍

  • 卡扣流量分析 Spark Core

  • 卡扣车流量转化率 Spark Core

  • 各区域车流量最高top5的道路统计 SparkSQL

  • 稽查布控,道路实时拥堵统计 SparkStreaming

hive表

monitor_flow_action表 – date 日期 天 – monitor_id 卡口号 – camera_id 摄像头编号 – car 车牌 – action_time 某个摄像头拍摄时间 s – speed 通过卡扣的速度 – road_id 道路id – area_id 区域ID

monitor_camera_info表 某一个卡扣对应的摄像头编号 – monitor_id:卡扣编号 – camera_id:摄像头编号

数据加载hive中 1). 创建表,加载数据load data Data2File

hive -f createHiveTab.sql

2). 集群中提交spark作业,使用代码生成到hive Data2Hive

大数据开发流程

  • 数据调研(对底层的数据的表结构进行调研,分析和研究)
  • 需求分析(与PM讨论需求,画原型图 axure)
  • 基于讨论出来的结果做出技术方案(某个难点用什么技术,数据库选型)
  • 具体实施

功能点

  • 根据使用者(平台使用者)指定的某些条件,筛选出指定的一批卡扣信息(比如根据区域、时间筛选)

  • 检测卡扣状态,对于筛选出来的所有的卡口(不代表一个摄像头)信息统计 • 卡口正常数 • 异常数 • camera的正常数 • camera的异常数 • camera的详细信息( monitor_id:camera_id)

  • 车流量最多的TonN卡扣号 • 获取每一个卡扣的详细信息( Top5 )

  • 随机抽取N个车辆信息,对这些数据可以进行多维度分析(因为随机抽取出来的N个车辆信息可以很权威的代表整个 区域的车辆)

  • 计算出经常高速通过的TopN卡口 (查看哪些卡扣经常被高速通过,高速,中速,正常,低速 根据三个速度段进行四 次排序,高速通过的车辆数相同就比较中速通过的车辆数,以此来推)

  • 跟车分析

需求分析

按条件筛选卡扣信息

• 可以指定 不同的条件,时间范围、区域范围、卡扣号等 可以灵活的分析不同区域的卡扣信息

监测卡扣状态

• 对符合条件的卡扣信息,可以动态的检查每一个卡扣的状态,查看卡扣是否正常工作,也可以查看摄像头

车流量最多的TonN卡扣

• 查看哪些卡扣的车流量最高,为什么会出现这么高的车流量。分析原因,例如今天出城的车辆非常多,啥原因,今天进 城的车辆非常多,啥原因? 要造反? 这个功能点里面也会拿到具体的车辆的信息,分析一下是京牌车造成的还是外地 车牌?

在符合条件的卡扣信息中随机抽取N个车辆信息

• 随机抽取N辆车的信息,可以权威的代表整个区域的车辆,这时候可以分析这些车的轨迹,看一下在不同的时间点车辆 的流动方向。以便于道路的规划。

计算出经常高速通过的TopN卡口

• 统计出是否存在飙车现象,或者经常进行超速行驶,可以在此处安装违章拍摄设备

跟车分析

• 计算出所有车是否被跟踪过,然后将结果存储在MySQL中,以便后期进行查询

项目分析

monitor_flow_action 监控数据表

monitor_camera_info 卡扣与摄像头基本关系表

1.卡扣监控

统计: 正常的卡扣个数,异常的卡扣个数,正常的摄像头个数,异常的摄像头个数,异常的摄像头详细信息

正常卡扣个数:

monitor_camera_info 基本关系表中卡扣与摄像头的关系与在monitor_flow_action 监控数据表 中,卡扣与摄像头的关系完全对应上
0001:11111,22222
0001 11111 xxx
0001 22222 xxx
	
RDD思路-正常的卡扣数为例:
		monitor_flow_action表 -> RDD<Monitor_id,Camera_id> -> RDD<Monitor_id,[camera_ids]> - RDD<Monitor_id,camera_ids>
		monitor_camera_info表 -> RDD<Monitor_id,Camera_id> -> RDD<Monitor_id,[camera_ids]>

异常的卡扣个数:

1.monitor_camera_info 基本关系表中 卡扣 与摄像头的关系,在监控的数据表中 一条都没有对应。

2.monitor_camera_info 基本关系表中 卡扣 与摄像头的关系,在监控的数据表中 部分数据有对应。

正常的摄像头个数:

异常的摄像头个数:

异常的摄像头详细信息:0001:11111,22222,33333

~0004:76789,27449,87911,61106,45624,37726,09506
~0001:70037,23828,34361,92206,76657,26608
~0003:36687,99260,49613,97165
~0006:82302,11645,73565,36440
~0002:60478,07738,53139,75127,16494,48312
~0008:34144,27504,83395,62222,49656,18640
~0007:19179,72906,55656,60720,74161,85939,51743,40565,13972,79216,35128,27369,84616,09553
~0000:67157,85327,08658,57407,64297,15568,31898,36621
~0005:09761,12853,91031,33015,52841,15425,45548,36528

注意:

求个数: 累加器实现(并行 分布式)

异常的摄像头信息,用累加器实现,无非拼的是字符串

更新累加器与take使用时,take算子可以触发多个job执行,可以造成累加器重复计算。

./spark-submit --master spark://node1:7077,node2:7077 --jars ../lib/fastjson-1.2.11.jar,../lib/mysql-connector-java-5.1.6.jar  --class MonitorFlowAnalyze  ../lib/Test.jar 1


 ~0001:13846,54785,51995,64341,45994,32228,82054,87746
 ~0003:38780,08844,03281,07183,50318,87000,16722,11604,26508,45523,46380
 ~0007:61833,19140,38387
 ~0005:63920,23464,37389,01219,96765,24844,32101,24141~ 
 ~0004:60778,35444,35403,68811,73819,81893
 ~0006:09621,67028,96375,60036,91237,53743,10305
 ~0002:24694,01172,25945,79625,83215,72235,26855
 ~0008:24630,40432,96808,78708,28294
 ~0000:68070,12865,49505,26035,36931,38053,91868

2.通过车辆数最多的topN卡扣

3.统计topN卡扣下经过的所有车辆详细信息

4.车辆通过速度相对比较快的topN卡扣

车速: 120=<speed 高速 90<=speed<120 中速 60<=speed<90 正常 0<speed<60 低速

5.卡扣“0001”下所有车辆轨迹

1.过滤日期范围内 卡扣“0001”下 有哪些车辆?
2.过滤日期范围内 这些车辆经过卡扣的时间,按照时间升序排序

6.车辆碰撞

01,02中这一天同时出现的车辆
01:(car,row) 02:(car,row)
(car,row).join(car,row)

01:car  02:car
car02.intersection(car02)

7.随机抽取车辆

在一天中要随机抽取100辆车,抽取的车辆可以权威代表当天交通运行情况。
假如这天一共有10000辆车,要随机抽取100辆车:
sample(true,0.1,seed)

00~01 100    100/10000*100 = 1 
01~02 100	1
02~03 100	1
04~05 200	2
05~06 200	2
06~07 300	3
08~09 500	5
09~10 200	2
10~11 200	2
11~12 300	3
12~13 500	5
13~14 700	7
。。
。。
。。

8.卡扣流量转换率

卡扣流量转换率,是指车辆运行连续卡口的转换率 如:一车辆经过卡口的轨迹为:0001,0002,0004,0001,0002,0003,0001,0002。 0001,0002:卡扣0001到卡扣0002的转换率为:经过卡扣0001,又经过卡扣0002的次数/经过卡扣0001的次数,针对上面的例子, 经过卡口0001的次数为:3次,经过卡口0001,又经过卡口0002的次数为:3次,那么卡扣0001到卡扣0002的转换率为:3/3=100%,同理: 0001,0002,0003 代表卡扣0001,0002到卡扣0003的转换率。 0001,0002,0003,0004代表卡扣0001,0002,0003到卡扣00004的转换率。 MonitorOneStepConvertRateAnalyze.java

一辆车的轨迹: 0001->0002->0003->0001->0002->0004->0005->0001 0001,0002----卡扣0001到卡扣0002 的车流量转化率:通过卡扣0001又通过卡扣0002的次数/通过卡扣0001的次数 2/3 0001,0002,0003 ---- 卡扣0001,0002到0003的车辆转换率:通过卡扣0001,0002,0003的次数 /通过卡扣0001,0002 0001,0002,0003,0004 -----卡扣0001,0002,0003到0004的车辆转换率:通过卡扣0001,0002,0003,0004的次数 /通过卡扣0001,0002,0003 0001,0002,0003,0004,0005 -----卡扣0001,0002,0003,0004到0005的车辆转换率:通过卡扣0001,0002,0003,0004,0005的次数 /通过卡扣0001,0002,0003,0004的次数 手动输入卡扣号: 0001,0002,0003,0004,0005 求: 0001,0002 0001,0002,0003 0001,0002,0003,0004 0001,0002,0003,0004,0005

粤A11111: ("0001",100) ("0001,0002",30) ("0001,0002,0003",10) 粤B22222: ("0001",200) ("0001,0002",100) ("0001,0002,0003",70) ("0001,0002,0003,0004",10)

9.实时道路拥堵情况

计算一段时间内卡扣下通过的车辆的平均速度。
这段时间不能太短,也不能太长。就计算当前时间的前五分钟 当前卡扣下通过所有车辆的平均速度。
每隔5s 计算一次当前卡扣过去5分钟 所有车辆的平均速度。

SparkStreaming 窗口函数
window lenth:5min
slide interval:5s

10.动态改变广播变量

`transform` `foreachRDD`

11.统计每个区域中车辆最多的前3道路

道路车辆:道路中的每个卡扣经过的车辆累加

天河区	元岗路1		0001=30,0002=50,0003=100,0004=20	200	
天河区	元岗路2		0005=50,0006=100 					150
天河区	元岗路3		100
越秀区	xxx1		200
越秀区	xxx2		150
越秀区	xxx3		100

SparkSQL
Hive 表 --t1 :
	monitor_id		car		road_id		area_id	
	
	
	-----
	
	areaId	area_name	road_id		monitor_id	car ------	tmp_car_flow_basic
	
		
sql:
	select area_name,road_id,count(car) as car_count,UDAF(monitor_id) as monitor_infos from t1 group by area_name,road_id		---- tmp_area_road_flow_count
	
开窗函数:row_number() over (partition by xxx order by xxx ) rank

	select area_name,road_id,car_count,monitor_infos, row_number() over (partition by area_name order by car_count desc ) rank from tmp_area_road_flow_count  ---- tmp
	
	select area_name,road_id,car_count,monitor_infos from tmp where rank <=3
	
	
-----------------------------------------------------------------------	
总sql:
	select 
		area_name,road_id,car_count,monitor_infos 
	from 
		(
		select 
			area_name,road_id,car_count,monitor_infos, row_number() over (partition by area_id order by carCount desc ) rank 
		from 
			(
				select 
					area_name,road_id,count(car) as car_count ,UDAF(monitor_id) as monitor_infos
				from 
					t1 
				group by area_name,road_id
			) t2
		) t3
	where rank <=3

=================================================================================================================== sql: select prefix_area_name_road_id,count(car) as car_count,UDAF(monitor_id) as monitor_infos from t1 group by prefix_area_name_road_id ---- tmp_area_road_flow_count

	select area_name,road_id,car_count,monitor_infos, row_number() over (partition by area_name order by car_count desc ) rank from tmp_area_road_flow_count  ---- tmp
	
	select area_name,road_id,car_count,monitor_infos from tmp where rank <=3
	
	
-----------------------------------------------------------------------	
总sql:
	select 
		area_name,road_id,car_count,monitor_infos 
	from 
		(
		select 
			area_name,road_id,car_count,monitor_infos, row_number() over (partition by area_id order by carCount desc ) rank 
		from 
			(
				select 
					area_name,road_id,count(car) as car_count ,UDAF(monitor_id) as monitor_infos
				from 
					t1 
				group by area_name,road_id
			) t2
		) t3
	where rank <=3

车辆轨迹

统计卡扣0001下所有车辆的轨迹 -- take(20)

各区域车流量最高topN的道路统计 1.会将小于spark.sql.autoBroadcastJoinThreshold值(默认为10M)的表广播到executor节点,不走shuffle过程,更加高效。 sqlContext.setConf("spark.sql.autoBroadcastJoinThreshold", "20971520"); //单位:字节 2.在Hive中执行sql文件: hive –f sql.sql 3.提交命令: --master spark://node1:7077,node2:7077 --jars ../lib/mysql-connector-java-5.1.6.jar,../lib/fastjson-1.2.11.jar --driver-class-path ../lib/mysql-connector-java-5.1.6.jar:../lib/fastjson-1.2.11.jar ../lib/Test.jar 4

3.缉查布控,道路实时拥堵统计 动态改变广播变量的值:可以通过transform和foreachRDD

屏蔽过多黄色警告,忽略java类方法的参数 与注释; File -> Settings -> Editor -> Inspections -> java ->javadoc: 参数不一致的屏蔽: Declaration has problems in Javadoc refere 红色 改成 waring黄色 参数没有注释: Dangling Javadoc comment 去掉勾选 Declaration has Javadoc problems 去掉勾选