本站小編為你精心準備了流式計算在交通管理中應用參考范文，愿這些范文能點燃您思維的火花，激發您的寫作靈感。歡迎深入閱讀并收藏。

【摘要】

近年來，為強化路面動態監控，各級公安交通管理部門開始大規模建設車輛智能監測記錄系統（以下簡稱卡口系統），隨著卡口系統接入的卡口數量日益增多，當卡口的過車數據量大到一定時候，基于傳統關系型數據庫的比對預警方式，會出現預警時間延遲，無法滿足實戰的實時性需求。本文針對卡口過車的實時預警要求，對大數據流式計算組件進行了分析和研究，提出了過車信息實時接入、實時比對預警的大數據流式計算技術解決方案，為基層路面民警的及時攔截查處爭取了時間。

【關鍵詞】

大數據；流式計算；SparkStreaming；智能交通；卡口

引言

近年來，為強化路面動態監控，各級公安交通管理部門開始大規模建設車輛智能監測記錄系統(以下簡稱卡口系統)，根據公安部“金盾工程”總體建設以及公安機關圖像信息聯網應用要求，以公安交通管理綜合應用平臺為依托，整合共享各地車輛智能監測記錄系統信息資源，建立橫向聯網、縱向貫通的交通安全主動防控云平臺，滿足各級公安交通管理部門車輛緝查布控和預警攔截、車輛軌跡和交通流量分析研判、交通違法行為甄別查處等業務應用。隨著各地卡口系統接入的卡口數量日益增多，基于傳統關系型數據庫的實時比對預警，無法滿足實戰的實時性的需求。自從Google了基于云計算的分布式大數據處理編程模型，大數據技術得到了廣泛的應用，開源的Hadoop分布式計算軟件框架更是將大數據應用推向了極限，網頁搜索、精準營銷等典型應用的成功使Hadoop、MapReduce成為大數據的象征。MapReduce是一種離線的批處理方式，可以成功處理TB、PB級海量數據，但無法應對實時數據分析需求和對消息事件的實時響應，大數據處理需要支持實時處理和迭代計算技術作為補充，因此流式計算成為大數據技術研究的新熱點。流式計算來自于一個信念：數據的價值隨著時間的流逝而降低，所以事件出現后必須盡快對它們進行處理，而不是緩存起來成批處理。基于卡口海量的實時過車信息，如何與黑名單信息快速的比對預警，成為當前主動防控云平臺應用的關鍵技術。本文提出了基于大數據流式計算的快速比對的解決方案，實現嫌疑車輛快速比對預警，為基層路面民警的及時攔截查處爭取了時間。

1主動防控平臺概況

按照公安部公路交通安全防控體系建設要求，基于全國機動車緝查布控系統[1]，應用大數據、云計算技術，實現了卡口機動車過車信息匯聚，實現海量過車信息查詢、軌跡分析、套牌分析[2]、伴隨分析、碰撞分析、區間測速等功能，實現跨區域、跨警鐘的信息共享、深度挖掘，為監測公路運行情況、快速查緝違法行為、打擊涉車犯罪、提升公路安全管控水平和社會安全服務水平。

1.1軟件架構

（1）如圖1所示，在分布式消息總線集群服務器上構建基于Kafka的分布式消息總線，前端卡口將過車信息臨時存儲在Kafka消息隊列中；（2）在流計算集群服務器上構建基于SparkStreaming的實時流式計算，實現過車信息與機動車登記信息、黑名單信息等實時關聯分析；（3）在分布式存儲集群服務器上構建基于HBase[3]的分布式數據庫，實現過過車信息、流量統計信息、嫌疑車輛信息，黑名單信息等存儲；（4）構建基于Hadoop大數據引擎[4]，實現關系型數據庫和分布式數據庫的數據關聯應用和可視化展示。1.2平臺功能按照面向基層，貼近實現、重在應用的工作思路，基于卡口過車信息，研發了車輛監控、緝查布控、執勤執法、分析研判等四大功能模塊，具體功能如下。1.2.1車輛監控功能實現基于GIS卡口、視頻實時監控、目標車輛實時追蹤、機動車軌跡查詢等功能。

1.2.2緝查布控功能

采用大數據流式計算技術，實現過車的實時比對預警、重點人員車輛的實時比對預警、假套牌車輛的實時比對、區間測速、流量統計的實時運算。

1.2.3執勤執法功能

實現執法服務站管理、重點車輛檢查登記、現場違法非現場攔截查處等功能。1.2.4分析研判功能實現對公路客運、旅游客運、危險品運輸車輛、逾期未檢驗、逾期未報廢、凌晨2時至5時客運車輛違規上路行駛、重要路段區間測速、道路交通流量等分析功能，實現了對嫌疑假牌、套牌、伴隨車輛、碰撞車輛等分析研判功能。

2數據處理架構

數據處理架構由數據采集、數據接入、流式計算、數據輸出等四部分構成.數據采集：車輛智能監測記錄系統的前端卡口負責過車信息采集，包括文本和圖片信息。數據接入：車輛智能監測記錄系統調用全國機動車緝查布控系統提供Webservice接入服務，將過車信息寫入，過車信息使用Kafka分布式消息隊列作為緩沖，接入服務不再負責比對。流式計算：使用Storm或SparkStreaming等流式計算技術，從Kafka分布式消息隊列中取數據進行實時比對處理。數據輸出：比對結果通過JDBC方式輸出至關系型數據庫ORACLE。

2.1流式計算技術

流式計算技術和批量處理技術有著本質的差別，流式計算技術需要維護消息隊列并進行實時消息的及時處理。分布式流式計算技術雖然處于起步發展階段，但由于市場廣泛需求的驅動，成為關注和研究熱點。當前具有代表性的流式計算技術有Storm、SparkStreaming[2]。Storm是Twitter支持開發的一款分布式、開源的、實時的、高容錯的大數據流式計算系統。Storm集群主要由一個主節點和一群工作節點構成，通過Zookeeper進行協調。如圖3所示，在Storm中，先要設計一個用于實時計算的圖狀結構，我們稱之為拓撲（topology）。這個拓撲將會被提交給集群，由集群中的主控節點（masternode）分發代碼，將任務分配給工作節點（workernode）執行。一個拓撲中包括spout和bolt兩種角色，其中spout發送消息，負責將數據流以tuple元組的形式發送出去；而bolt則負責轉換這些數據流，在bolt中可以完成計算、過濾等操作，bolt自身也可以隨機將數據發送給其他bolt。由spout發射出的tuple是不可變數組，對應著固定的鍵值對[5,6]。如圖4所示，SparkStreaming是核心SparkAPI的一個擴展，它并不會像Storm那樣一次一個地處理數據流，而是在處理前按時間間隔預先將其切分為一段一段的批處理作業。Spark針對持續性數據流的抽象稱為DStream（DiscretizedStream），一個DStream是一個微批處理（micro-batching）的RDD（彈性分布式數據集）；而RDD則是一種分布式數據集，能夠以兩種方式并行運作，分別是任意函數和滑動窗口數據的轉換。表1給出了Storm、SparkStreaming的功能，性能等對比，基于下述對比，選擇了SparkStreaming流式計算技術。

2.2分布式消息隊列

Kafka也是Apache[7]下的開源消息系統項目，是一種高吞吐量的分布式消息訂閱系統，在普通的服務器上每秒也能處理幾十萬條消息，可用于低時延的收集和發送大量的事件和日志數據。Kafka也是Apache下的開源消息系統項目，是一種分布式的，基于/訂閱的消息系統。它以時間復雜度為O(1)的方式提供消息持久化能力，即使對TB級以上數據也能保證常數時間復雜度的訪問性能。具有高吞吐量，即使在非常普通的硬件機器上也能做到單機支持每秒十萬條以上消息的傳輸。支持KafkaServer間的消息分區及分布式消費，同時能保證每個Partition內的消息順序傳輸。同時支持離線數據處理和實時數據處理，并且支持在線水平擴展。Kafka包括以下四個組件：一是話題（Topic），它是特點類型的消息流，消息是字節的有效負載，話題是消息的分類名；二是生產者（Producer），它是能夠消息到話題的任何對象；三是（Broker）或Kafka集群，已的消息保存在其中；四是消費者（Consumer），它可以訂閱一個或多個話題，并從拉取數據，從而消費這些已的消息。Kafka的整體架構如圖5所示。因為Kafka內在就是分布式的，一個Kafka集群通常包括多個。為了均衡負載，將話題分成多個分區，每個存儲一或多個分區。多個生產者和消費者能夠同時生產和獲取消息。

2.3SparkStreaming與Kafka集成

Kafka[6]是一個分布式的消息-訂閱系統，下面介紹如何使用SparkStreaming從Kafka中接收數據，具體包括兩種方法：一是使用Receivers和Kafka高層次的API；二是使用DirectAPI，這是使用低層次的KafkaAPI，并沒有使用到Receivers，是Spark1.3.0中開始引入的。

2.3.1基于Receivers的方法

這個方法使用了Receivers來接收數據。如圖6，Receivers的實現使用到Kafka高層次的消費者API。對于所有的Receivers，接收到的數據將會保存在Sparkexecutors中，然后由SparkStreaming啟動的Job來處理這些數據。然而，在默認的配置下，這種方法在失敗的情況下會丟失數據，為了保證零數據丟失，你可以在SparkStreaming中使用WAL日志，這是在Spark1.2.0才引入的功能，這使得我們可以將接收到的數據保存到WAL中（WAL日志可以存儲在HDFS上），所以在失敗的時候，我們可以從WAL中恢復，而不至于丟失數據。

2.3.2基于DirectAPI的方法和基于Receiver

接收數據不一樣，這種方式定期地從Kafka的topic+partition中查詢最新的偏移量，再根據定義的偏移量范圍在每個batch里面處理數據。當作業需要處理的數據來臨時，spark通過調用Kafka的簡單消費者API讀取一定范圍的數據。如圖7和基于Receiver方式相比，這種方式主要有幾個優點：（1）簡化并行。我們不需要創建多個Kafka輸入流，然后union他們。而使用directStream，SparkStreaming將會創建和Kafka分區一樣的RDD分區個數，而且會從Kafka并行地讀取數據，也就是說Spark分區將會和Kafka分區有一一對應的關系，這對我們來說很容易理解和使用。（2）高效。第一種實現零數據丟失是通過將數據預先保存在WAL中，這將會復制一遍數據，這種方式實際上很不高效，因為這導致了數據被拷貝兩次：一次是被Kafka復制；另一次是寫到WAL中。（3）恰好一次語義（Exactly-oncesemantics）。通過Kafka低層次的API，并沒有使用到Zookeeper，偏移量僅僅被SparkStreaming保存在Checkpoint中。這就消除了SparkStreaming和Zookeeper中偏移量的不一致，而且可以保證每個記錄僅僅被SparkStreaming讀取一次，即使是出現故障。

3流式計算解決方案

隨著前端卡口接入數量的不斷增加，過車數據規模的不斷擴大，使用傳統的邏輯架構會造成以下兩個問題：一是過車數據上傳積壓問題，傳統的傳輸機制已不能滿足大數據量的過車信息上傳；二是實時比對效率降低問題，通過接入服務程序提供的Webservice或Servlet接口，實現過車信息接入，接入時進行比對預警，當數據量大的時候，無法及時預警。基于以上問題我們采用大數據庫流式計算技術，使用Kafka分布式消息總線作為緩沖，接入服務不再負責比對，只負責提供接口寫入數據至Kafka，然后由SparkStreaming從Kafka中取數據進行實時比對預警，并將結果輸出到交通安全主動防控平臺中。

3.1比對預警示意圖

3.2過車等9種信息接入

通過接入服務器過實現過車信息、流量檢測信息、氣象檢測信息、交通事件信息、交通誘導信息、停車場車輛停車信息、警車定位信息、警員定位信息、非現場違法信息9種數據接入，Kafka以Topic來進行消息管理，在系統中按每一數據類型設定相應的Topic，然后由相應的Consumer去負責消費需要的Topic數據。

3.3基礎信息內存加載

為了更快的信息加載速度，系統先定期將機動車登記信息、黑名單信息裝載至HBase分布式數據庫，然后SparkStreaming再從分布式數據庫加載機動車登記信息、黑名單信息。SparkStreaming信息加載時分為全項信息加載和根據hash算法部分信息加載。其中全項信息的加載由后臺任務定時加載；根據hash算法的加載由Consumer任務在拉取partitions數據時觸發加載根據partitions的hash算法決定要加載那部分車輛的基礎信息和布控車輛信息（目前沿用山東項目的算法，根據號牌號碼信息和partitions的個數）。

3.4比對預警信息生成

在系統中，對于從Kafka中實時獲取到的Topic數據，SparkStreaming作為Consumer負責動靜態信息的實時碰撞、分析和預警，區間測速、旅行時間計算等。系統已實現人工布控黑名單信息、機動車登記信息、駕駛人管理信息、全國客運車輛和危險品運輸車、逾期未年檢、逾期未報廢等重點車輛信息數據和卡口過車信息實時碰撞分析，對嫌疑車輛在秒級發出實時預警信息，指揮中心民警在接收到預警信息后可及時指揮路面民警對嫌疑車輛進行攔截查處。流式計算的最終結果，對于海量的布控黑名單軌跡信息、重點車輛軌跡信息等根據業務類型存放到HBase中相應的業務表中，對于預警信息，存放到Oracle關系型數據庫，便于后續業務處理。

4結論

大數據流式計算和批量計算適用于不同的應用場景。批處理匯聚海量數據分析出的結果可能更精確，但對數據時效性要求嚴格而對歷史數據積累并不非常關注的場景，流式計算具有明顯的優勢。批量計算和流式計算是有優劣互補的，因此在多種應用場合下可以將兩者結合起來使用。目前，山東全省已實現了1023套卡口上傳過車信息100毫秒內接收并預警，日接入過車信息超過1550萬。實現了各類動、靜態信息的實時比對，嫌疑車輛300毫秒內發出預警。與傳統采用關系型數據庫相比，采用SparkStreaming流式計算技術的比對預警更快、監測的種類更多。山東已實現逾期未檢驗、逾期未報廢、強制注銷、車主駕照滿分、暫扣等實時比對預警，為基層民警的應用提供了數據支撐。通過采用SparkStreaming流式計算技術，解決了數據的積壓問題、保證了業務數據的有效性，解決前端卡口接入性能、保證了比對預警時效性、從而大大提高了交管部門的管控能力。基于Hadoop的大數據云計算平臺擴展性強，存儲和計算能力可以不斷提升，充分運用大數據云計算技術，讓交通管理變得更加“智慧”。

作者：周建寧 徐曉東 蔡崗 單位：公安部交通管理科學研究所