基于Spark的實時視頻分析系統①

鄭 健,馮 瑞

1(復旦大學 計算機科學技術學院,上海 201203)

2(復旦大學 上海市智能信息處理重點實驗室,上海 201203)

3(復旦大學 上海視頻技術與系統工程研究中心,上海 201203)

基于Spark的實時視頻分析系統①

鄭 健1,馮 瑞2,3

1(復旦大學 計算機科學技術學院,上海 201203)

2(復旦大學 上海市智能信息處理重點實驗室,上海 201203)

3(復旦大學 上海視頻技術與系統工程研究中心,上海 201203)

視頻監控技術在交通管理、公共安全、智慧城市等方面有著廣泛的應用前景,且向著智能識別、實時處理、大數據分析的方向發展. 本文針對大規模實時視頻監控提出了新的解決方案. 基于Spark streaming流式計算、分布式存儲及OLAP框架,使多路視頻處理在可擴展性、容錯性及數據多維聚合分析上具有明顯的優勢. 系統根據視頻處理算法劃分為單機處理與分布式處理. 并將視頻圖像處理與數據分析耦合,利用Kafka消息隊列與Spark streaming完成對多路視頻輸出數據的進一步操作. 結合分布式存儲方案,并利用OLAP框架實現對海量數據實時多維聚合分析與高效實時查詢.

Spark; 視頻分析; 數據分析; 實時計算

當前社會對視頻監控的需求日益強烈,在公共安全、交通管理、智慧城市等領域有著廣泛的應用. 相對于傳統的單純視頻監控,結合圖像算法與機器學習算法的智能視頻監控有著顯著的優勢[1]. 當前的視頻監控系統,在視頻流傳輸、解碼、圖像處理與識別等方面有著大量的研究,但在分布式處理與多路視頻數據關聯分析等方面的研究較少.

在實際生活中,視頻監控一般應用在較為重要的場合,這對視頻監控系統在高可用、低延遲上提出了更高的要求. 由于視頻圖像這種非結構化數據處理較為復雜,導致圖像處理算法的處理時間相對較長. 如果要實現視頻分析的實時性,需要在圖像分析算法或者處理架構上進行改進. 現有的視頻監控以單路視頻流為單位進行分析,每路視頻流處理后的數據之間完全沒有關聯. 而在一些場景,如智能交通領域,往往同一片區域的視頻數據之間能夠挖掘出更多的信息[2],可針對交通數據進行交互式分析、實時挖掘、實時研判,為千萬上億級別的過車記錄實現關聯匯總以及實時預警. 如整個交通網絡的流量數據分析、交通擁塞時合理路線的推送、套牌車輛識別、黑名單車輛鄰近監控點預警等. 而且每路視頻處理后的歷史數據往往包含著重要信息,能夠從中分析出有價值的信息以進行統計與決策,如過去1年的時間里車輛的行駛軌跡分析等. 在海量數據面前,傳統的關系型數據庫已無法實時進行多維數據分析,這需要采用專門的OLAP解決方案[3].

目前,有些針對視頻的分布式解決方案,如Natarajan等人[2]提出的基于MapReduce的交通視頻分析系統,但其將視頻流先存儲至HDFS后再利用MapReduce進行分布式處理,時延較大. 而且Hadoop適合批處理,而不適合實時流計算. Huang等人[4]雖提出利用Spark進行圖像處理的方法,但在視頻流動態接入、系統高可用與多路視頻關聯分析等方面沒有提出解決方案.

本文基于Spark[5]、Kafka消息中間件[6]、分布式存儲[7,8]及OLAP框架[9],使多路視頻處理在可擴展性、容錯性及數據多維聚合分析上具有明顯的優勢. Spark streaming相對于其他流式計算框架的突出特點是能夠高效并行的實現容錯恢復及其對數據分析、機器學習、圖計算等工具的無縫集成.

1 系統概述

本系統分為采集層、視頻處理層、消息分析層、存儲與查詢層等. 系統總體框圖如圖1所示.

圖1 系統總體框圖

采集層: 用于接收來自各個攝像頭傳入的RTSP視頻流數據. 利用 YARN 與 Spark streaming,根據集群資源負載情況,動態的在相關節點中啟動接收器,接收視頻流數據并且解碼.

視頻處理層: 用于處理解碼后的視頻幀序列. 將視頻分析算法分為兩種: 幀間相關與幀間無關. 并對兩種情況進行分別處理. 然后調用opencv及其他圖像處理庫進行圖像預處理、特征提取與圖像識別等操作,并將處理結果輸出至Kafka消息隊列以進行進一步分析.

消息處理層: 采用 Kafka與 Spark streaming 進行消息流實時分析. 根據具體場景,可對單路視頻消息獨自分析,或對多路視頻消息進行關聯分析. 該層針對消息數據的精確一次消費語義進行了實現. 消息處理結果輸出至存儲層,以對數據進行實時查詢或離線分析.

存儲與查詢層: 以HDFS、HBase及Postgresql作為結構化數據及非結構化數據的存儲組件,利用Elasticsearch與Kylin完成對數據的全文檢索及多維聚合分析. 該層能夠向外提供實時查詢、統計分析、圖片檢索等服務.

完成視頻處理算法的各個子模塊后,在Spark Job中利用PipedRDD進行操作. pipe為RDD及Dstream中的算子,用于調用外部程序對分區集合中的數據進行處理. 系統調用流程如下:

sc.fromCameraStream(“rtsp://10.27.31.5/road?fps=15”)

.pipe(“feature_extraction”)

.pipe(“object_detection”)

2 各模塊的算法設計與實現

2.1 采集層

2.1.1 視頻流動態接入

這里將從攝像頭傳入的視頻流信息集中于分布式平臺上處理,由于各個機器資源配置不同,導致處理能力各異. 在實際應用中,可能隨時增加或減少攝像頭,所以處理平臺需要實現攝像頭的動態接入. 本系統基于YARN資源管理器,這里ReceiverTracker將根據集群資源狀況進行動態調度. 在新增攝像頭時,將Receiver接收器分配至空閑的集群節點上,或在減少攝像頭時,將資源壓力較大的節點上的Receiver接收器重新分配至其他空閑的集群節點上. 在YARN中,每個NodeManager每隔一段時間會通過心跳機制向Resource Manager發送心跳信息,其中包括該節點上的資源使用狀況. 通過調用相關接口,能夠獲取集群中各個節點的資源使用信息. Spark在啟動 Receiver時,先將其封裝成一個RDD,作為一個job進行任務提交,等待Resource Manager為其分配合適的container,然后在相應Node Manager節點上啟動. 本系統對原生的ReceiverTracker組件進行了二次開發,在封裝Receiver成RDD時,根據YARN接口返回的數據修改其preferredLocation變量以選擇分配節點. 視頻流接入流程如圖2所示.

圖2 視頻流接入流程

2.1.2 視頻流解碼

對于從攝像頭流入RTSP視頻流數據,需要先進行解碼,然后才能進行下一步處理. 這里通過調用opencv庫中的VideoCapture方法進行視頻流解碼(需要ffmpeg與gstreamer庫的支持). 在Receiver接收器的onStart方法中,調用VideoCapture方法將不斷流入的rtsp視頻流數據解碼并生成一個個Mat數據結構,這里用frame表示,該結構是opencv中進行圖像處理操作的基本對象. Spark streaming的執行流程如圖3所示.Spark streaming 中的 Receiver接收器負責接收數據,然后解碼生成圖像幀序列并緩存至currentBuffer中.Timer定時器每隔設定的間間隔回調BlockGenerator,將currentBuffer中緩存的數據流封裝成一系列Block,放入blocksForPushing中,作為Dstream中分區的數據記錄. blockPushThread周期性的從blocksForPushing取出Block存入存儲體系. 由JobGenerator為每一批Block 生成 Job,交由 Spark 引擎處理. 本系統中,Dstream中RDD的分區數據由圖像幀序列組成,每幀圖像frame的數據格式如下所示.

其中,

streamID: 視頻流編號,作為每個視頻流的唯一標識.

frameSID: 幀序列編號,用于標記視頻流中各個幀序列的順序關系.

data: 視頻幀數據,包含圖像數據的字節數組,由Mat轉化而來.

圖3 Spark streaming 執行流程

2.2 視頻流處理

解碼后的視頻數據是由圖像幀序列組成. 現有的視頻分析算法根據每幀圖像之間的關系可分為幀間無關與幀間相關. 對于前者,可以將各幀圖像分布至不同節點上并行處理; 對于后者,由于幀序列之間相關聯,所以需要進行單節點處理. 視頻幀處理過程包括圖片預處理、特征提取、目標檢測與識別等. 在圖像處理過程中,需要調用opencv庫或者其他圖像處理庫. 在Spark中,可以通過利用RDD上的pipe算子,能夠以管道形式調用外部程序對RDD中的數據進行處理.

2.2.1 幀間相關

幀間相關是指視頻分析算法需要使用連續的視頻幀序列才能完成處理,即當前幀的處理需要依賴之前的幀序列,如煙霧檢測、目標跟蹤等. 這里將每路視頻流分配在一個節點中處理. 接收器Receiver中的store方法將接收到的數據保存在本地節點中,如果將RDD中的分區數置為1,將保證數據只在本地節點處理. 由于在處理時當前RDD中的視頻幀數據時可能需要用到過去時間段RDD中的視頻幀數據信息,所以需要使用Dstream中的window算子及updateStateByKey算子進行處理.

2.2.2 幀間無關

幀間無關是指視頻分析算法可以針對每幀圖像進行處理,而不需要依賴以前的視頻數據,如車牌識別、人臉識別等. 可以將接收到的視頻幀數據分配到若干節點中并行處理,如圖4所示. 由于視頻幀數據集合作為一個RDD,所以能夠以分區為單位分配到不同節點中處理,并發度即為分區數. 處理完成后,通過 collect算子收集至driver端,使用Restful用于前端顯示. 由于Spark具有容錯性,所以框架本身能夠保證不會出現RDD分區中的幀數據丟失.

圖4 任務執行圖

視頻處理層專注于對圖像幀數據的分析識別等工作,而對于識別結果的進一步處理,以消息數據形式流入消息處理層. 將兩個階段進行耦合,有利于系統的靈活性與魯棒性.

2.3 消息處理層

這里采用Kafka與Spark streaming進行消息流實時分析,Kafka用于接收視頻流處理過程中所產生的消息數據,Spark streaming用于對 Kafka中的消息數據進行實時處理. 由于Kafka具有高并發、高吞吐量及高可用等優勢,所以經常在流式計算中充當消息中間件.對視頻分析過程中所產生的輸出數據集中處理,有較大的靈活性和實用價值. 如多路視頻間的關聯分析,在交通監控場景下,可對交通數據進行實時挖掘、實時研判,為千萬上億級別的過車記錄實現關聯匯總以及實時預警,如套牌車分析、黑名單車輛預警、道路流量統計等. 消息分析的框架如圖5所示. 該模塊主要技術細節如下:

① topic 設置: 每路視頻對應一個 topic,每個 topic包含的partition數目依賴具體場景而定. 對于需要順序消費的場景,partition數目為1. 否則可以設置多個分區進行并行消費.

② 消息語義的實現: 在消息處理場景中,可分為最多一次消費、最少一次消費及精確一次消費. 由于Kafka在接收生產者發送的消息時,可以選擇使用確認機制,所以容易完成生產者對Kafka的語義實現. 下面討論消費者對Kafka的精確一次消費語義的實現. 依據是否采用 Receiver,Spark streaming 提供的與 Kafka集成的接口,可分為createStream與createDirectStream.由于后者能夠提供更強的語義保證與更高的效率,所以這里使用createDirectStream. 這里Kafka中需要消費的分區與RDD包含的分區一一對應. 消費Kafka中的消息時,主要包括以下步驟: 接收消息、處理消息、輸出結果并修改Kafka中相應分區的offset. create DirectStream能夠保證接收消息與處理消息精確一次,輸出結果與修改offset的先后順序將決定最終的語義.如果先修改offset,可能造成部分數據未處理; 如果先輸出結果,可能造成部分數據輸出多次. 實現精準一次消費語義,需要滿足: 輸出結果冪等或者保證二者作為一個事務操作提交. 以Postgresql為例,每個分區開始消費時需要先從數據庫中讀取相應的offset值放入TopicAndPartition 中,并傳入 createDirectStream. 處理完畢時需要將兩者放在一個事務中提交.

本程序采用scala語言編寫,數據庫模塊采用ScalikeJDBC庫. 關鍵代碼如下:

圖5 消息分析框架

2.4 視頻信息存儲

在存儲層,系統采用HDFS+HBase+Po stgresql+Elasticsearch+kylin方案. 其中HBase用于存儲結構化數據,包括視頻數據屬性信息. Elasticsearch 與 kylin 用于海量數據的實時檢索與多維聚合分析. Elasticsearch是一種基于lucence的高效的分布式全文檢索引擎框架. 這里用于多標簽查找與全文搜索. Kylin是一種優秀的OLAP框架,在這里用于對歷史數據的實時多維聚合分析. 這對于視頻監控領域的分析型場景具有很大實用價值. 視頻信息檢索分為兩個部分: 基于文本的數據查詢與基于內容的圖像搜索(如圖6所示).

圖6 圖像搜索架構圖

2.5 容錯性分析

視頻采集層,由于Receiver具有容錯性,在節點失效或Receiver崩潰時,Spark會在當前節點或是其他節點重新啟動.

消息隊列層,由于Kafka以分區為單位進行多結點主從分配,主分區所在節點宕機時會切換至從分區.

生產者發送消息至Kafka時會有確認機制,主從分區全部存儲完畢后才對消息進行確認.

在利用Spark進行數據計算時,由于Spark引擎本身能夠利用Linearge及checkpoint進行容錯恢復,所以流入的數據在計算過程中不會丟失.

3 實驗結果

本實驗所運行的服務器集群包含5個節點,集群的硬件配置如下表所示.

實驗對比了原系統與本系統的處理性能. 由于日志處理架構為公用部分,兩種系統均可進行接入. 故本實驗著重檢驗了視頻處理部分,并根據實時性、容錯性及擴展性對兩者進行了對比.

角色 數量 內存 CPU Master /RM 1 32 GB Intel(R) Xeon(R)CPU E5-2650 @2.00 GHz Work /NM 4 8 GB Intel Core(TM)CPU i3-2120 CPU@3.3 GHz

(1) 實時性

在實驗中,進行4路視頻實時分析. 包括2路幀間相關分析(包括camera3與camera4)與2路幀間無關分析(包括camera1與camera2). 分別從rtsp流中解碼并轉換成opencv能夠處理的Mat格式,調用視頻分析算法進行處理. 在幀間相關算法的選擇上,本實驗中選取了煙霧檢測算法,該算法根據視頻幀序列中的煙霧信息分析其運動特征,進行煙霧的判別. 在幀間相關算法的選擇上,實驗中選取了車牌識別算法,根據提取的hog特征,依據SVM+adaboost分類器進行識別.

原系統的視頻流處理模塊是采用單機進行處理,雖然也配置了多個節點,但節點之間沒有關聯,僅僅用于多路視頻流的擴展. 不僅無法對幀間無關視頻流進行分布式處理,而且在單點故障上也沒有解決.

當算法對單幀圖像的處理時間比較長時,為了保證對流入視頻流處理的實時性,往往采用跳幀的方式進行處理. 這種方式雖然可以緩解這一情況,但以犧牲視頻流的信息為代價. 本系統的分布式處理方案則解決了這一問題. 在實驗中,以每秒能夠處理的視頻幀作為衡量標準,實驗結果如圖7所示. 對于幀間無關算法,由于采用分布式處理方案,所以每秒處理的幀數很高,與實時傳入的幀數相同. 對于幀間相關算法,本系統的每秒處理幀數也高于原系統.

圖7 處理幀數對比圖

(2) 容錯性

原系統中,采用守護進程監控所在節點的視頻流處理進程的的運行狀態,如果失效則進行重新啟動. 但存在單點故障的情況.

本系統中,由于Receiver具有容錯性,在節點失效或Receiver崩潰時,Spark會在當前節點或是其他節點重新啟動. 整個恢復過程只需1～3秒.

(3) 擴展性

當新增攝像頭時,原系統需要人工選擇節點接入.對于不同機器的處理能力,人工往往無法比對. 如果隨機選擇節點進行接入,則有可能導致節點負載不均造成處理時延的進一步增加.

而本系統,基于資源管理平臺YARN,能夠實時獲取集群的資源狀態,自動選擇合適的節點進行接入并處理,極大減少了接入時延,提升了處理效率.

4 結語

本文對基于Spark的實時視頻分析系統的實現過程進行了詳細介紹,該系統利用當前開源社區主流的分布式組件,涉及資源管理、計算、存儲、消息中間件及大數據即時分析等技術,具有高可用、實時性等特點. 將傳統的對視頻數據的單路分析轉換為對多路視頻信息的關聯分析,通過添加消息分析層,進一步實時的挖掘不同數據之間的價值,為更有效的構建分析型業務提供了支撐.

由于YARN資源管理器中沒有將GPU作為資源對象,而且Spark框架也沒有提供GPU的相關接口,故這里沒有引入. 后續將會進行相關工作以進一步提升處理效率.

1黃凱奇,陳曉棠,康運鋒,等. 智能視頻監控技術綜述. 計算機學報,2015,38(6): 1093–1118. [doi: 10.11897/SP.J.1016.2015.01093]

2Natarajan VA,Jothilakshmi S,Gudivada VN. Scalable traffic video analytics using hadoop mapreduce. The First International Conference on Big Data,Small Data,Linked Data and Open Data. Barcelona,Spain. 2015. 11–15.

3Apache Software Foundation. Kylin: Extreme OLAP engine for big data. http://kylin.apache.org/.

4黃文輝,馮瑞. 基于 Spark Streaming 的視頻/圖像流處理與新的性能評估方法. 計算機工程與科學,2015,37(11):2055–2060. [doi: 10.3969/j.issn.1007-130X.2015.11.010]

5Apache Software Foundation. Apache spark,lightning-fast cluster computing. http://spark.apache.org.

6Apache Software Foundation. Kafka: A distributed streaming platform. https://kafka.apache.org.

7Apache Software Foundation. HBase: A distributed,scalable,big data store. https://hbase.apache.org.

8Apache Software Foundation. HDFS: A distributed,scalable File System. https://hadoop.apache.org.

9Apache Software Foundation. Apache Kylin: Extreme OLAP engine for big data. http://kylin.apache.org.

10Zaharia M,Chowdhury M,Das T,et al. Resilient distributed datasets: A fault-tolerant abstraction for in-memory cluster computing. 9th USENIX Conference on Networked Systems Design and Implementation. San Jose,CA,USA. 2012.

11Apache Software Foundation. Spark streaming + Kafka integration guide. http://spark.apache.org/docs/latest/streaming-kafka-integration.html.

Scalable Real-Time Video Analysis System Based on Spark

ZHENG Jian1,FENG Rui2,3

1(School of Computer Science,Fudan University,Shanghai 201203,China)
2(Shanghai Key Laboratory of Intelligent Information Processing,Fudan University,Shanghai 201203,China)
3(Shanghai Engineering Research Center for Video Technology and System,Fudan University,Shanghai 201203,China)

The video surveillance technology has a wide application prospect in traffic management,public safety,intelligent city,and is developing towards intelligent recognition,real-time processing,and large data analysis. In this paper,we propose a new system for large-scale real-time video surveillance. The system is based on Spark streaming,distributed storage and OLAP framework so that multi-channel video processing has obvious advantages in scalability,fault tolerance and data analysis of the multi-dimensional polymer. According to video processing algorithm,the processing module is divided into single machine processing and distributed processing. The video processing is separated from the data analysis,and the further operation of the multi-channel video output data is completed by using Kafka message queue and Spark streaming. Combining the distributed storage technology with OLAP framework,the system achieves real-time multi-dimensional data analysis and high-performance real-time query.

Spark; video analysis; data analysis; realtime computation

鄭健,馮瑞.基于 Spark 的實時視頻分析系統.計算機系統應用,2017,26(12):51–57. http://www.c-s-a.org.cn/1003-3254/6112.html

國家科技支撐計劃(2013BAH09F01);臨港地區智能制造產業專項(ZN2016020103)

2017-03-20; 采用時間: 2017-04-10