本站小編為你精心準備了流式計算在交通管理中應(yīng)用參考范文，愿這些范文能點燃您思維的火花，激發(fā)您的寫作靈感。歡迎深入閱讀并收藏。
【摘要】
近年來，為強化路面動態(tài)監(jiān)控，各級公安交通管理部門開始大規(guī)模建設(shè)車輛智能監(jiān)測記錄系統(tǒng)（以下簡稱卡口系統(tǒng)），隨著卡口系統(tǒng)接入的卡口數(shù)量日益增多，當卡口的過車數(shù)據(jù)量大到一定時候，基于傳統(tǒng)關(guān)系型數(shù)據(jù)庫的比對預(yù)警方式，會出現(xiàn)預(yù)警時間延遲，無法滿足實戰(zhàn)的實時性需求。本文針對卡口過車的實時預(yù)警要求，對大數(shù)據(jù)流式計算組件進行了分析和研究，提出了過車信息實時接入、實時比對預(yù)警的大數(shù)據(jù)流式計算技術(shù)解決方案，為基層路面民警的及時攔截查處爭取了時間。
【關(guān)鍵詞】
大數(shù)據(jù)；流式計算；SparkStreaming；智能交通；卡口
引言
近年來，為強化路面動態(tài)監(jiān)控，各級公安交通管理部門開始大規(guī)模建設(shè)車輛智能監(jiān)測記錄系統(tǒng)(以下簡稱卡口系統(tǒng))，根據(jù)公安部“金盾工程”總體建設(shè)以及公安機關(guān)圖像信息聯(lián)網(wǎng)應(yīng)用要求，以公安交通管理綜合應(yīng)用平臺為依托，整合共享各地車輛智能監(jiān)測記錄系統(tǒng)信息資源，建立橫向聯(lián)網(wǎng)、縱向貫通的交通安全主動防控云平臺，滿足各級公安交通管理部門車輛緝查布控和預(yù)警攔截、車輛軌跡和交通流量分析研判、交通違法行為甄別查處等業(yè)務(wù)應(yīng)用。隨著各地卡口系統(tǒng)接入的卡口數(shù)量日益增多，基于傳統(tǒng)關(guān)系型數(shù)據(jù)庫的實時比對預(yù)警，無法滿足實戰(zhàn)的實時性的需求。自從Google了基于云計算的分布式大數(shù)據(jù)處理編程模型，大數(shù)據(jù)技術(shù)得到了廣泛的應(yīng)用，開源的Hadoop分布式計算軟件框架更是將大數(shù)據(jù)應(yīng)用推向了極限，網(wǎng)頁搜索、精準營銷等典型應(yīng)用的成功使Hadoop、MapReduce成為大數(shù)據(jù)的象征。MapReduce是一種離線的批處理方式，可以成功處理TB、PB級海量數(shù)據(jù)，但無法應(yīng)對實時數(shù)據(jù)分析需求和對消息事件的實時響應(yīng)，大數(shù)據(jù)處理需要支持實時處理和迭代計算技術(shù)作為補充，因此流式計算成為大數(shù)據(jù)技術(shù)研究的新熱點。流式計算來自于一個信念：數(shù)據(jù)的價值隨著時間的流逝而降低，所以事件出現(xiàn)后必須盡快對它們進行處理，而不是緩存起來成批處理。基于卡口海量的實時過車信息，如何與黑名單信息快速的比對預(yù)警，成為當前主動防控云平臺應(yīng)用的關(guān)鍵技術(shù)。本文提出了基于大數(shù)據(jù)流式計算的快速比對的解決方案，實現(xiàn)嫌疑車輛快速比對預(yù)警，為基層路面民警的及時攔截查處爭取了時間。
1主動防控平臺概況
按照公安部公路交通安全防控體系建設(shè)要求，基于全國機動車緝查布控系統(tǒng)[1]，應(yīng)用大數(shù)據(jù)、云計算技術(shù)，實現(xiàn)了卡口機動車過車信息匯聚，實現(xiàn)海量過車信息查詢、軌跡分析、套牌分析[2]、伴隨分析、碰撞分析、區(qū)間測速等功能，實現(xiàn)跨區(qū)域、跨警鐘的信息共享、深度挖掘，為監(jiān)測公路運行情況、快速查緝違法行為、打擊涉車犯罪、提升公路安全管控水平和社會安全服務(wù)水平。
1.1軟件架構(gòu)
（1）如圖1所示，在分布式消息總線集群服務(wù)器上構(gòu)建基于Kafka的分布式消息總線，前端卡口將過車信息臨時存儲在Kafka消息隊列中；（2）在流計算集群服務(wù)器上構(gòu)建基于SparkStreaming的實時流式計算，實現(xiàn)過車信息與機動車登記信息、黑名單信息等實時關(guān)聯(lián)分析；（3）在分布式存儲集群服務(wù)器上構(gòu)建基于HBase[3]的分布式數(shù)據(jù)庫，實現(xiàn)過過車信息、流量統(tǒng)計信息、嫌疑車輛信息，黑名單信息等存儲；（4）構(gòu)建基于Hadoop大數(shù)據(jù)引擎[4]，實現(xiàn)關(guān)系型數(shù)據(jù)庫和分布式數(shù)據(jù)庫的數(shù)據(jù)關(guān)聯(lián)應(yīng)用和可視化展示。1.2平臺功能按照面向基層，貼近實現(xiàn)、重在應(yīng)用的工作思路，基于卡口過車信息，研發(fā)了車輛監(jiān)控、緝查布控、執(zhí)勤執(zhí)法、分析研判等四大功能模塊，具體功能如下。1.2.1車輛監(jiān)控功能實現(xiàn)基于GIS卡口、視頻實時監(jiān)控、目標車輛實時追蹤、機動車軌跡查詢等功能。
1.2.2緝查布控功能
采用大數(shù)據(jù)流式計算技術(shù)，實現(xiàn)過車的實時比對預(yù)警、重點人員車輛的實時比對預(yù)警、假套牌車輛的實時比對、區(qū)間測速、流量統(tǒng)計的實時運算。
1.2.3執(zhí)勤執(zhí)法功能
實現(xiàn)執(zhí)法服務(wù)站管理、重點車輛檢查登記、現(xiàn)場違法非現(xiàn)場攔截查處等功能。1.2.4分析研判功能實現(xiàn)對公路客運、旅游客運、危險品運輸車輛、逾期未檢驗、逾期未報廢、凌晨2時至5時客運車輛違規(guī)上路行駛、重要路段區(qū)間測速、道路交通流量等分析功能，實現(xiàn)了對嫌疑假牌、套牌、伴隨車輛、碰撞車輛等分析研判功能。
2數(shù)據(jù)處理架構(gòu)
數(shù)據(jù)處理架構(gòu)由數(shù)據(jù)采集、數(shù)據(jù)接入、流式計算、數(shù)據(jù)輸出等四部分構(gòu)成.數(shù)據(jù)采集：車輛智能監(jiān)測記錄系統(tǒng)的前端卡口負責過車信息采集，包括文本和圖片信息。數(shù)據(jù)接入：車輛智能監(jiān)測記錄系統(tǒng)調(diào)用全國機動車緝查布控系統(tǒng)提供Webservice接入服務(wù)，將過車信息寫入，過車信息使用Kafka分布式消息隊列作為緩沖，接入服務(wù)不再負責比對。流式計算：使用Storm或SparkStreaming等流式計算技術(shù)，從Kafka分布式消息隊列中取數(shù)據(jù)進行實時比對處理。數(shù)據(jù)輸出：比對結(jié)果通過JDBC方式輸出至關(guān)系型數(shù)據(jù)庫ORACLE。
2.1流式計算技術(shù)
流式計算技術(shù)和批量處理技術(shù)有著本質(zhì)的差別，流式計算技術(shù)需要維護消息隊列并進行實時消息的及時處理。分布式流式計算技術(shù)雖然處于起步發(fā)展階段，但由于市場廣泛需求的驅(qū)動，成為關(guān)注和研究熱點。當前具有代表性的流式計算技術(shù)有Storm、SparkStreaming[2]。Storm是Twitter支持開發(fā)的一款分布式、開源的、實時的、高容錯的大數(shù)據(jù)流式計算系統(tǒng)。Storm集群主要由一個主節(jié)點和一群工作節(jié)點構(gòu)成，通過Zookeeper進行協(xié)調(diào)。如圖3所示，在Storm中，先要設(shè)計一個用于實時計算的圖狀結(jié)構(gòu)，我們稱之為拓撲（topology）。這個拓撲將會被提交給集群，由集群中的主控節(jié)點（masternode）分發(fā)代碼，將任務(wù)分配給工作節(jié)點（workernode）執(zhí)行。一個拓撲中包括spout和bolt兩種角色，其中spout發(fā)送消息，負責將數(shù)據(jù)流以tuple元組的形式發(fā)送出去；而bolt則負責轉(zhuǎn)換這些數(shù)據(jù)流，在bolt中可以完成計算、過濾等操作，bolt自身也可以隨機將數(shù)據(jù)發(fā)送給其他bolt。由spout發(fā)射出的tuple是不可變數(shù)組，對應(yīng)著固定的鍵值對[5,6]。如圖4所示，SparkStreaming是核心SparkAPI的一個擴展，它并不會像Storm那樣一次一個地處理數(shù)據(jù)流，而是在處理前按時間間隔預(yù)先將其切分為一段一段的批處理作業(yè)。Spark針對持續(xù)性數(shù)據(jù)流的抽象稱為DStream（DiscretizedStream），一個DStream是一個微批處理（micro-batching）的RDD（彈性分布式數(shù)據(jù)集）；而RDD則是一種分布式數(shù)據(jù)集，能夠以兩種方式并行運作，分別是任意函數(shù)和滑動窗口數(shù)據(jù)的轉(zhuǎn)換。表1給出了Storm、SparkStreaming的功能，性能等對比，基于下述對比，選擇了SparkStreaming流式計算技術(shù)。
2.2分布式消息隊列
Kafka也是Apache[7]下的開源消息系統(tǒng)項目，是一種高吞吐量的分布式消息訂閱系統(tǒng)，在普通的服務(wù)器上每秒也能處理幾十萬條消息，可用于低時延的收集和發(fā)送大量的事件和日志數(shù)據(jù)。Kafka也是Apache下的開源消息系統(tǒng)項目，是一種分布式的，基于/訂閱的消息系統(tǒng)。它以時間復(fù)雜度為O(1)的方式提供消息持久化能力，即使對TB級以上數(shù)據(jù)也能保證常數(shù)時間復(fù)雜度的訪問性能。具有高吞吐量，即使在非常普通的硬件機器上也能做到單機支持每秒十萬條以上消息的傳輸。支持KafkaServer間的消息分區(qū)及分布式消費，同時能保證每個Partition內(nèi)的消息順序傳輸。同時支持離線數(shù)據(jù)處理和實時數(shù)據(jù)處理，并且支持在線水平擴展。Kafka包括以下四個組件：一是話題（Topic），它是特點類型的消息流，消息是字節(jié)的有效負載，話題是消息的分類名；二是生產(chǎn)者（Producer），它是能夠消息到話題的任何對象；三是（Broker）或Kafka集群，已的消息保存在其中；四是消費者（Consumer），它可以訂閱一個或多個話題，并從拉取數(shù)據(jù)，從而消費這些已的消息。Kafka的整體架構(gòu)如圖5所示。因為Kafka內(nèi)在就是分布式的，一個Kafka集群通常包括多個。為了均衡負載，將話題分成多個分區(qū)，每個存儲一或多個分區(qū)。多個生產(chǎn)者和消費者能夠同時生產(chǎn)和獲取消息。
2.3SparkStreaming與Kafka集成
Kafka[6]是一個分布式的消息-訂閱系統(tǒng)，下面介紹如何使用SparkStreaming從Kafka中接收數(shù)據(jù)，具體包括兩種方法：一是使用Receivers和Kafka高層次的API；二是使用DirectAPI，這是使用低層次的KafkaAPI，并沒有使用到Receivers，是Spark1.3.0中開始引入的。
2.3.1基于Receivers的方法
這個方法使用了Receivers來接收數(shù)據(jù)。如圖6，Receivers的實現(xiàn)使用到Kafka高層次的消費者API。對于所有的Receivers，接收到的數(shù)據(jù)將會保存在Sparkexecutors中，然后由SparkStreaming啟動的Job來處理這些數(shù)據(jù)。然而，在默認的配置下，這種方法在失敗的情況下會丟失數(shù)據(jù)，為了保證零數(shù)據(jù)丟失，你可以在SparkStreaming中使用WAL日志，這是在Spark1.2.0才引入的功能，這使得我們可以將接收到的數(shù)據(jù)保存到WAL中（WAL日志可以存儲在HDFS上），所以在失敗的時候，我們可以從WAL中恢復(fù)，而不至于丟失數(shù)據(jù)。
2.3.2基于DirectAPI的方法和基于Receiver
接收數(shù)據(jù)不一樣，這種方式定期地從Kafka的topic+partition中查詢最新的偏移量，再根據(jù)定義的偏移量范圍在每個batch里面處理數(shù)據(jù)。當作業(yè)需要處理的數(shù)據(jù)來臨時，spark通過調(diào)用Kafka的簡單消費者API讀取一定范圍的數(shù)據(jù)。如圖7和基于Receiver方式相比，這種方式主要有幾個優(yōu)點：（1）簡化并行。我們不需要創(chuàng)建多個Kafka輸入流，然后union他們。而使用directStream，SparkStreaming將會創(chuàng)建和Kafka分區(qū)一樣的RDD分區(qū)個數(shù)，而且會從Kafka并行地讀取數(shù)據(jù)，也就是說Spark分區(qū)將會和Kafka分區(qū)有一一對應(yīng)的關(guān)系，這對我們來說很容易理解和使用。（2）高效。第一種實現(xiàn)零數(shù)據(jù)丟失是通過將數(shù)據(jù)預(yù)先保存在WAL中，這將會復(fù)制一遍數(shù)據(jù)，這種方式實際上很不高效，因為這導(dǎo)致了數(shù)據(jù)被拷貝兩次：一次是被Kafka復(fù)制；另一次是寫到WAL中。（3）恰好一次語義（Exactly-oncesemantics）。通過Kafka低層次的API，并沒有使用到Zookeeper，偏移量僅僅被SparkStreaming保存在Checkpoint中。這就消除了SparkStreaming和Zookeeper中偏移量的不一致，而且可以保證每個記錄僅僅被SparkStreaming讀取一次，即使是出現(xiàn)故障。
3流式計算解決方案
隨著前端卡口接入數(shù)量的不斷增加，過車數(shù)據(jù)規(guī)模的不斷擴大，使用傳統(tǒng)的邏輯架構(gòu)會造成以下兩個問題：一是過車數(shù)據(jù)上傳積壓問題，傳統(tǒng)的傳輸機制已不能滿足大數(shù)據(jù)量的過車信息上傳；二是實時比對效率降低問題，通過接入服務(wù)程序提供的Webservice或Servlet接口，實現(xiàn)過車信息接入，接入時進行比對預(yù)警，當數(shù)據(jù)量大的時候，無法及時預(yù)警。基于以上問題我們采用大數(shù)據(jù)庫流式計算技術(shù)，使用Kafka分布式消息總線作為緩沖，接入服務(wù)不再負責比對，只負責提供接口寫入數(shù)據(jù)至Kafka，然后由SparkStreaming從Kafka中取數(shù)據(jù)進行實時比對預(yù)警，并將結(jié)果輸出到交通安全主動防控平臺中。
3.1比對預(yù)警示意圖
3.2過車等9種信息接入
通過接入服務(wù)器過實現(xiàn)過車信息、流量檢測信息、氣象檢測信息、交通事件信息、交通誘導(dǎo)信息、停車場車輛停車信息、警車定位信息、警員定位信息、非現(xiàn)場違法信息9種數(shù)據(jù)接入，Kafka以Topic來進行消息管理，在系統(tǒng)中按每一數(shù)據(jù)類型設(shè)定相應(yīng)的Topic，然后由相應(yīng)的Consumer去負責消費需要的Topic數(shù)據(jù)。
3.3基礎(chǔ)信息內(nèi)存加載
為了更快的信息加載速度，系統(tǒng)先定期將機動車登記信息、黑名單信息裝載至HBase分布式數(shù)據(jù)庫，然后SparkStreaming再從分布式數(shù)據(jù)庫加載機動車登記信息、黑名單信息。SparkStreaming信息加載時分為全項信息加載和根據(jù)hash算法部分信息加載。其中全項信息的加載由后臺任務(wù)定時加載；根據(jù)hash算法的加載由Consumer任務(wù)在拉取partitions數(shù)據(jù)時觸發(fā)加載根據(jù)partitions的hash算法決定要加載那部分車輛的基礎(chǔ)信息和布控車輛信息（目前沿用山東項目的算法，根據(jù)號牌號碼信息和partitions的個數(shù)）。
3.4比對預(yù)警信息生成
在系統(tǒng)中，對于從Kafka中實時獲取到的Topic數(shù)據(jù)，SparkStreaming作為Consumer負責動靜態(tài)信息的實時碰撞、分析和預(yù)警，區(qū)間測速、旅行時間計算等。系統(tǒng)已實現(xiàn)人工布控黑名單信息、機動車登記信息、駕駛?cè)斯芾硇畔ⅰ⑷珖瓦\車輛和危險品運輸車、逾期未年檢、逾期未報廢等重點車輛信息數(shù)據(jù)和卡口過車信息實時碰撞分析，對嫌疑車輛在秒級發(fā)出實時預(yù)警信息，指揮中心民警在接收到預(yù)警信息后可及時指揮路面民警對嫌疑車輛進行攔截查處。流式計算的最終結(jié)果，對于海量的布控黑名單軌跡信息、重點車輛軌跡信息等根據(jù)業(yè)務(wù)類型存放到HBase中相應(yīng)的業(yè)務(wù)表中，對于預(yù)警信息，存放到Oracle關(guān)系型數(shù)據(jù)庫，便于后續(xù)業(yè)務(wù)處理。
4結(jié)論
大數(shù)據(jù)流式計算和批量計算適用于不同的應(yīng)用場景。批處理匯聚海量數(shù)據(jù)分析出的結(jié)果可能更精確，但對數(shù)據(jù)時效性要求嚴格而對歷史數(shù)據(jù)積累并不非常關(guān)注的場景，流式計算具有明顯的優(yōu)勢。批量計算和流式計算是有優(yōu)劣互補的，因此在多種應(yīng)用場合下可以將兩者結(jié)合起來使用。目前，山東全省已實現(xiàn)了1023套卡口上傳過車信息100毫秒內(nèi)接收并預(yù)警，日接入過車信息超過1550萬。實現(xiàn)了各類動、靜態(tài)信息的實時比對，嫌疑車輛300毫秒內(nèi)發(fā)出預(yù)警。與傳統(tǒng)采用關(guān)系型數(shù)據(jù)庫相比，采用SparkStreaming流式計算技術(shù)的比對預(yù)警更快、監(jiān)測的種類更多。山東已實現(xiàn)逾期未檢驗、逾期未報廢、強制注銷、車主駕照滿分、暫扣等實時比對預(yù)警，為基層民警的應(yīng)用提供了數(shù)據(jù)支撐。通過采用SparkStreaming流式計算技術(shù)，解決了數(shù)據(jù)的積壓問題、保證了業(yè)務(wù)數(shù)據(jù)的有效性，解決前端卡口接入性能、保證了比對預(yù)警時效性、從而大大提高了交管部門的管控能力。基于Hadoop的大數(shù)據(jù)云計算平臺擴展性強，存儲和計算能力可以不斷提升，充分運用大數(shù)據(jù)云計算技術(shù)，讓交通管理變得更加“智慧”。
作者：周建寧 徐曉東 蔡崗 單位：公安部交通管理科學研究所