授權協議 Readme

開發語言 Java

操作系統 跨平臺

軟件類型 開源軟件

所屬分類 云計算、 云原生

軟件簡介

車流量監控

前言

目的 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 -> RDD - RDD monitor_camera_info表 -> RDD -> RDD

?

異常的卡扣個數：

?

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=

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 去掉勾選

?