基于Kafka和Storm的車輛套牌實時分析存儲系統(tǒng)①

任培花,蘇 銘

(山西大同大學(xué) 計算機與網(wǎng)絡(luò)工程學(xué)院,大同 037009)

“套牌車”是指未在交管部門辦理手續(xù),偽造、冒用他人合法車輛和行駛駕照的車輛[1].近年來,隨著機動車數(shù)量的增加,車輛套牌現(xiàn)象屢見不鮮,嚴重損害了交通參與者的合法權(quán)益,增加了交管部門的工作難度,甚至增添了社會不穩(wěn)定因素等.因此,套牌行為已成為當(dāng)前交通和治安管理中的一個難點問題.

現(xiàn)今城市的各個重要路口,均有車輛監(jiān)控系統(tǒng)可以記錄車牌號碼、經(jīng)過地點、時間、現(xiàn)場圖片等.傳統(tǒng)的車輛套牌監(jiān)測大多基于車輛監(jiān)控系統(tǒng)的日志信息采用人工識別、牌照識別、或射頻識別等方法識別套牌車輛.然而伴隨車輛數(shù)量和出行量地持續(xù)增加,城市交通采集范圍在逐步擴大,按現(xiàn)有監(jiān)控系統(tǒng),每天會產(chǎn)生以億為單位的日志記錄,不管用哪種監(jiān)測方法面對海量交通數(shù)據(jù)集,必然會有成本高、效率低、實時性差等問題.因此,為了解決這些難題,國內(nèi)外很多人研究將大數(shù)據(jù)技術(shù)引入到車輛套牌稽查系統(tǒng)中,對這些數(shù)據(jù)進行實時分析和存儲.

1 相關(guān)研究

隨著并行計算框架的產(chǎn)生,關(guān)于車輛套牌的大數(shù)據(jù)研究正在興起,很多學(xué)者專家紛紛展開相關(guān)研究.以“套牌”、“大數(shù)據(jù)”為檢索詞,從 2014-2019年期間,對中國知網(wǎng)、萬方數(shù)據(jù)庫、維普網(wǎng)公開發(fā)表的有關(guān)車輛套牌大數(shù)據(jù)文章進行檢索.發(fā)現(xiàn)車輛套牌監(jiān)測的研究主要停留在監(jiān)測方法方面,大數(shù)據(jù)技術(shù)在車輛套牌方面的應(yīng)用研究不足,采用的大數(shù)據(jù)分析框架也比較老.另外,研究內(nèi)容主要集中在數(shù)據(jù)分析方面,數(shù)據(jù)存儲方面的研究很少.

代表性的如文獻[1]提出了一種基于歷史車牌識別數(shù)據(jù)(ANPR)集的套牌車并行檢測方法TP-Finder,實現(xiàn)了基于整數(shù)劃分的數(shù)據(jù)分塊策略,可準(zhǔn)確呈現(xiàn)所有疑似套牌車輛的歷史行車軌跡.文獻[2]提出一種基于路段閾值表和時間滑動窗口的套牌計算模型,可以實時甄別交通數(shù)據(jù)流中的套牌嫌疑車.文獻[3]將實際車輛記錄遷移到Hadoop集群的HBase中,然后從Hive從HBase中獲取同一車牌號碼的時空分布情況,通過校正因子獲取最終的嫌疑套牌車.文獻[4,5]提出一種針對大規(guī)模數(shù)據(jù)集的分布式計算模型MapReduce.文獻[6]提出一種新的基于Hadoop的MapReduce算法模型,可以有效地解決處理海量數(shù)據(jù)時面臨的性能瓶頸問題,該算法通過引入多臺硬件計算資源協(xié)同處理大規(guī)模數(shù)據(jù)下的套牌車檢測.文獻[7]提出一種比Hadoop實時性好的分布式實時計算框架Storm.該框架具有健壯性、容錯性、動態(tài)調(diào)整并行度等特性.

通過對文獻研究可知,目前普遍做法將實際的海量數(shù)據(jù)導(dǎo)入或并行連接到大數(shù)據(jù)平臺上,然后再進行數(shù)據(jù)分析,進而監(jiān)測套牌車輛.但這些做法的實時性普遍不足,如文獻[1-3]均采用對歷史記錄的分析,車輛運行每時每刻在發(fā)生,這種做法顯然實時性很差.文獻[4-6]中雖然考慮了實時性,但MapReduce屬于批處理算法,等數(shù)據(jù)集到一個量的時候才啟動,勢必會有一個時間延遲.因此,本文充分考慮車輛套牌監(jiān)測的實時性要求,借鑒文獻[7],引入Kafka (分布式消息隊列)和Storm(分布式實時大數(shù)據(jù)處理系統(tǒng))來解決海量車牌數(shù)據(jù)的實時分析和信息存儲問題,Kafka作為中間件可以為日志信息提供緩沖機會,有效緩解了數(shù)據(jù)采集和數(shù)據(jù)分析的不同步問題,提高了數(shù)據(jù)的高可用性和實時性,避免了由于服務(wù)器故障而造成數(shù)據(jù)丟失的問題.Storm中的運行的是拓撲(topology)算法相對于MapReduce而言,進程是永駐的,只要有數(shù)據(jù)就可以進行實時處理,從而可以實現(xiàn)實時分析,實現(xiàn)套牌監(jiān)測的實時性、準(zhǔn)確性.

2 Kafka和Storm的車輛套牌實時分析存儲系統(tǒng)分析

為了實現(xiàn)實時分析、存儲車輛套牌信息,本文提出一個基于Kafka和Storm的車輛套牌實時分析存儲系統(tǒng)(簡稱車輛套牌實時分析存儲系統(tǒng)).從系統(tǒng)功能模塊的邏輯結(jié)構(gòu)和劃分兩個角度分別進行描述.

2.1 邏輯層次結(jié)構(gòu)

該系統(tǒng)包含的邏輯執(zhí)行過程:日志獲取、日志緩沖、數(shù)據(jù)分析和數(shù)據(jù)存儲.圖1是邏輯過程分層結(jié)構(gòu)圖.

圖1 邏輯層次結(jié)構(gòu)圖

數(shù)據(jù)采集主要用來收集用戶操作產(chǎn)生的車輛日志信息.數(shù)據(jù)緩存減少了流量超峰給系統(tǒng)帶來的壓力,該模塊使用Flume[8](分布式日志收集系統(tǒng))、Kafka、Storm、Redis[9](云數(shù)據(jù)庫)等大數(shù)據(jù)框架搭建后端服務(wù)架構(gòu).從結(jié)構(gòu)圖(如圖1)可知,道路監(jiān)控系統(tǒng)產(chǎn)生的日志信息,先通過Kafka進行備份緩存,然后將Kafka、Storm進行整合,將數(shù)據(jù)導(dǎo)入Storm運行框架中.Storm中Spout (Storm消息源)會源源不斷地從Kafka上某個主題獲取數(shù)據(jù),并對數(shù)據(jù)進行封裝發(fā)送到下游的Bolt (Storm消息處理者)計算節(jié)點,Bolt節(jié)點對數(shù)據(jù)進行實時計算,判斷是否出現(xiàn)套牌車輛,算法邏輯都會在Bolt節(jié)點中進行運算處理.最后系統(tǒng)將實時計算后的結(jié)果數(shù)據(jù)存儲到服務(wù)器的文件中,最后一次的車輛信息存入Redis中.

2.2 模塊功能分析

車輛套牌實時分析存儲系統(tǒng)的功能包括日志緩存與獲取、日志信息切分、套牌監(jiān)測和信息存儲.

(1)日志緩存與獲取

通過連接道路監(jiān)控系統(tǒng),在數(shù)據(jù)實時采集與數(shù)據(jù)實時處理之間搭建一個Kafka消息隊列進行緩存,解決數(shù)據(jù)實時采集與數(shù)據(jù)實時處理之間速度不同步和數(shù)據(jù)丟失的問題,進一步將道路車輛監(jiān)控系統(tǒng)產(chǎn)生的實時日志數(shù)據(jù)通過Kafka導(dǎo)入系統(tǒng).其次通過Kafka與Storm進行整合,然后將數(shù)據(jù)導(dǎo)入Storm運行框架中.

(2)日志信息切分

對采集到數(shù)據(jù)進行切分,獲取系統(tǒng)需要使用的有效數(shù)據(jù).

(3)套牌監(jiān)測

通過車輛id從Redis數(shù)據(jù)庫中快速讀取道路車輛監(jiān)控系統(tǒng)對應(yīng)的最近監(jiān)控歷史記錄,計算當(dāng)前車輛的區(qū)間速度,如果車輛速度超過區(qū)間速度值就將對應(yīng)車牌號碼標(biāo)記為套牌號牌.

(4)信息存儲

從實時的道路監(jiān)控記錄中提取出必要的交通信息,并將實時交通信息存入數(shù)據(jù)庫中.

2.3 類的設(shè)計

本系統(tǒng)涉及的主要實體類包括:KafkaTopo、SplitBolt、SpeedBolt、StorageBolt、JedisPoolUtils.下面是整個系統(tǒng)的類圖結(jié)構(gòu),如圖2所示.

實體類介紹:

(1)KafkaTopo類主要是將Kafka與Storm進行整合,這個類既具備緩沖特點又具備實時計算的特點.

(2)JedisPoolUtils類主要是編寫對Redis進行讀寫的Java客戶端代碼,讀取連接池配置文件,從而提供訪問Redis的接口.

(3)SplitBolt類主要是對收到的日志信息進行拆分,提取有用信息,發(fā)送到SpeedBolt.

(4)SpeedBolt類主要是對SplitBolt發(fā)送來的信息與Redis中的信息進行對比,計算出動態(tài)車速,并且檢測出套牌車輛,將套牌車輛信息發(fā)送到StorageBolt.

(5)StorageBolt類主要是對SpeedBolt發(fā)送來的套牌車輛信息進行存儲.

圖2 系統(tǒng)類圖

KafkaTopo類是整個系統(tǒng)的主要控制部分,通過將數(shù)據(jù)分析與計算流程串聯(lián)起來;首先是設(shè)置Kafka的主題與配置其Broker的主機地址,通過配置可以將日志數(shù)據(jù)獲取進來;接下來設(shè)置3個Bolt組件,分別完成數(shù)據(jù)切分、動態(tài)車速計算與數(shù)據(jù)存儲的功能;最后將整個工程打成jar包提交到配置好的服務(wù)器上,來完成實時的車輛套牌檢測功能.其具體的功能時序圖如圖3所示.

3 Kafka和Storm的車輛套牌實時分析存儲系統(tǒng)實現(xiàn)

本系統(tǒng)實現(xiàn)按照日志緩存與獲取、日志信息切分、套牌監(jiān)測和信息存儲4個功能展開.

3.1 日志緩存與獲取

日志緩存與獲取是通過Kafka來實現(xiàn),首先將Kafka分別安裝到3臺Linux服務(wù)器上,由于Kafka的高可用是通過Zookeeper[10]來實現(xiàn)的,因此需要在3臺服務(wù)器上安裝Zookeeper集群,Zookeeper集群的IP將作為Kafka的Broker的配置地址.接下來通過Storm提供的SpoutConfig來將Broker的主機地址、Kafka的主題、Zookeeper集群的服務(wù)器地址、Storm的SpoutId進行設(shè)置,從而將整個數(shù)據(jù)的接收所需要依賴的環(huán)境搭建起來.接下來對Storm從kafka獲取數(shù)據(jù)的方式進行設(shè)置,將其設(shè)置成從數(shù)據(jù)流的起始位置開始讀取數(shù)據(jù);這些配置設(shè)置完成后,最終實例化一個Topology Builder對象來將之前的配置信息進行整合,從而使其成功獲取從車輛監(jiān)控系統(tǒng)獲取的數(shù)據(jù).

圖3 系統(tǒng)時序圖

整個數(shù)據(jù)流的緩存是通過Kafka來實現(xiàn)的.因為Kafka的broker結(jié)點上會有消費者機制與生產(chǎn)者機制,通過設(shè)置生產(chǎn)者所生產(chǎn)消息的長度來控制消費者的消費.使得整個數(shù)據(jù)流經(jīng)過Kafka都會有一個緩存的過程,如果數(shù)據(jù)出現(xiàn)丟失,也可以通過設(shè)置ACK來實現(xiàn)回滾,使得消息不會產(chǎn)生丟失.整個Kafka的機制實現(xiàn)了日志信息數(shù)據(jù)流傳輸?shù)母呖捎眯?

其整個獲取與緩存的流程圖如圖4所示.

3.2 日志信息的切分

通過從Storm的Spout將信息發(fā)送到指定的SplitBolt進行切分字段,SplitBolt繼承BaseBasicBolt,通過對其execute()方法進行重寫來實現(xiàn)日志信息的切分.通過其方法的Tuple參數(shù)接收到Spout發(fā)送來的日志信息,將接收到的信息通過toString()方法轉(zhuǎn)換成字符串類型,從而進一步使用split()方法來對整個字符串信息進行切分,獲取車輛的id、車牌、坐標(biāo)、行進方向、拍攝時間等字段信息.之后調(diào)用參數(shù)中Collector對象的emit()方法來將切分出來的字段信息進行封裝,發(fā)送到下一個Bolt.

整個日志信息切分過程時序圖如圖5所示.

圖4 日志信息獲取與緩存流程圖

圖5 日志信息切分時序圖

3.3 套牌監(jiān)測

文獻[1]中提出一種利用車輛時空矛盾關(guān)系判斷套牌的算法,借鑒該算法.本文認為正常行駛的車輛在區(qū)間內(nèi)的動態(tài)車速在一個限速范圍內(nèi),如果某輛車出現(xiàn)了套牌情況,必然會出現(xiàn)在相同時間點,坐標(biāo)位置不同的情況,而且算出的區(qū)間動態(tài)車速遠高于標(biāo)準(zhǔn)車速,因此可斷定該車輛為可疑套牌車輛.

算法步驟:

Step 1.創(chuàng)建Jedis客戶端,配置相關(guān)數(shù)據(jù),建立連接池,連接Redis數(shù)據(jù)庫接口.

Step 2.獲取當(dāng)前車輛id、車牌、坐標(biāo)、運行方向、獲取時間等信息

Step 3.從Redis中通過id獲取車輛信息,若有,拿出來通過動態(tài)車速對比,看是否發(fā)生套牌.若沒有,將信息存入Redis中

其主要業(yè)務(wù)邏輯是:從上一Bolt中獲取“id”,“registId”,“hangId”,“x”,“y”,“dir”,“time”和“info”信息,通過id判斷Redis中是否有該車輛信息,若沒有,則將該車輛整條信息保存進入Redis中;若有,則將車輛信息拿出來切分,從而獲取“x”,“y”,“dir”,“time”等字段信息,將兩次的信息進行對比計算出動態(tài)車速;若此車速遠大于區(qū)間車速,則該車為套牌車輛.

(1)動態(tài)車速計算

從上一Bolt中獲取車輛信息,通過車輛id判斷Redis中是否有該車輛信息,若沒有,則將該車輛整條信息保存進入Redis中;若有,則將車輛信息拿出來切分,從而獲取坐標(biāo)、時間等字段信息,將兩次的信息進行對比計算出動態(tài)車速;若此車速遠大于區(qū)間車速,則該車為套牌車輛.

具體的業(yè)務(wù)處理是在execute()方法中,通過其Tuple參數(shù)來接收SplitBolt發(fā)送過來的各個字段信息和車輛的整條info日志信息.通過創(chuàng)建carInfo、hangId、x、y、dir、time等字段來獲取車輛信息;之后通過調(diào)用Jedis對象的get()方法,將last_hangId傳入,看返回結(jié)果是否為空,若可以獲取到車輛信息,則將獲取到的車輛信息進行再次切分,拿到上次記錄到的字段信息,分別設(shè)置為 last_x、last_y、last_dir、last_time;之后調(diào)用String對象的equals()方法來判斷相同車輛兩次的行駛方向是否相同,若相同并且為x方向,則將x于last_x相減取絕對值,其結(jié)果就是車輛的行駛路徑;將兩次時間相減取絕對值并進行單位換算,從而獲得時間.路程與時間相除即可得到車輛的動態(tài)車速.若行駛方向為y,計算方法相同.最后將車輛信息存入Redis中,替換掉之前的車輛信息.若從Redis中獲取數(shù)據(jù)為空,則說明此車輛還沒有被車輛監(jiān)控系統(tǒng)記錄過,因此直接存入Redis中.

(2)套牌判定

獲取動態(tài)車速后,接下來就是與此路段的區(qū)間標(biāo)準(zhǔn)車速進行比較,若獲取的動態(tài)車速遠大于區(qū)間車速,則懷疑出現(xiàn)了兩輛相同車牌的車輛,將其判定為套牌車輛.并將此車輛信息通過調(diào)用Collector的emit()方法將套牌車輛信息發(fā)送出去.最后仍然要調(diào)用declare OutputFields()方法,來指定發(fā)送字段為info,即套牌車輛信息.

對以上整個車速計算和套牌車的檢測所做出的功能時序圖如圖6所示.

圖6 套牌監(jiān)測功能時序圖

3.4 信息存儲

將檢測出的套牌車輛信息通過編寫輸出代碼和修改相關(guān)配置文件,存入分布式服務(wù)器上指定的文檔中,從而可以完成信息的分布式存儲.

首先,通過Tuple對象獲取SpeedBolt發(fā)送過來的套牌車輛信息,調(diào)用FileWriter的write()方法,來對日志信息進行寫操作;然后調(diào)用其flush()方法來刷新數(shù)據(jù)流,從而使得之前寫入的數(shù)據(jù)能完整輸出到指定的文件中.信息存儲的功能時序圖如圖7所示.

圖7 套牌車輛信息存儲時序圖

通過對4個功能模塊的實現(xiàn)過程敘述與部分功能時序圖的介紹,對系統(tǒng)的整個功能進行了詳細的實現(xiàn).對接收到的日志信息進行了詳細的分析與計算,整個系統(tǒng)的業(yè)務(wù)邏輯實現(xiàn)完畢,接下來工程的部署模塊可以直接將工程打成jar包來進行實時的運行.至此,系統(tǒng)實現(xiàn)已全部完成.

4 實驗

本系統(tǒng)的實驗環(huán)境包括集群搭建和數(shù)據(jù)庫服務(wù)器安裝,主要包括Zookeeper集群、Kafka集群、Storm集群、Redis安裝等.實驗環(huán)境配置完畢,將工程打成jar包上傳到搭建好的集群運行,通過連接車輛模擬系統(tǒng)產(chǎn)生實時數(shù)據(jù),可以看見Redis數(shù)據(jù)庫上存儲所有收集到的車輛信息,如圖8所示.

圖8 Redis測試數(shù)據(jù)圖

/home/hadoop/stormoutput目錄下存儲有套牌車輛嫌疑的車輛信息,如圖9所示.以這些信息為基礎(chǔ),交管人員只需后期和車輛具體核實,即可按照相關(guān)交通法規(guī)進行處理.

圖9 套牌車輛信息圖

5 總結(jié)

針對當(dāng)前海量車輛數(shù)據(jù)套牌檢測實時性差的難點,引入了大數(shù)據(jù)技術(shù),經(jīng)過文獻對比研究,發(fā)現(xiàn)Kafka作為中間件進行緩存,不僅保證了數(shù)據(jù)采集的效率而且還保證了數(shù)據(jù)的高可用性.再加上,Storm比Hadoop實時性處理能力強,所以Kafka結(jié)合Storm提高了車輛套牌監(jiān)測的實時性.最后通過集群環(huán)境的搭建,實驗分析發(fā)現(xiàn)系統(tǒng)性能已達到設(shè)計目標(biāo).在今后的研究工作中,將探索研究套牌識別的精準(zhǔn)度,進而更有效地幫助交管部門完成套牌識別工作.