大數據平臺的視頻分析并行處理系統設計

南京郵電大學電子與光學工程學院 程思哲 錢國明

視頻智能分析在傳統工業的安全生產監測領域有著很大的發展空間與應用潛力。隨著監控視頻數據量飛速增長以及視頻分析算法模型精度要求越來越高，視頻分析系統的低延遲和高并發成為了一個重要的挑戰。本文采用Spark、Kafka、HDFS、HBase、Opencv和Yolov5等技術棧構建了一個海量視頻數據分析平臺，基于此平臺可以實現對海量視頻數據進行實時流計算。

近年來，在國家工業互聯網戰略的推動下，工業園區對視頻智能分析系統的需求日益強烈。在企業端進行邊緣計算，實時檢測值班人員脫崗睡崗、未戴安全帽、受限區域違規闖入、人員超限、異常火點、異常煙霧等異常狀態，便于及時進行異常狀態預警及主動干預，對提高企業安全生產監測具有重要的意義。但是隨著監控視頻數據量快速增長以及分析精度要求越來越高，必須解決視頻分析系統對低延遲和高并發的要求，目前已出現了將視頻分析算法移植到大數據分布式平臺的解決方案，如Jothilakshmi等人提出的使用Hadoop、MapReduce的視頻分析方案，其原理是將視頻流直接存儲至HDFS，然后再使用MapReduce做分布式處理。這樣做雖然解決了海量視頻的存儲問題，但是MapReduce只適合做批處理，只能定時對hdfs中的視頻文件做分析，時延較大，無法做到對海量視頻的實時分析。本文采用Spark、Kafka、hdfs、hbase、opencv、yolov5等構建了一個海量視頻數據分析平臺，基于此平臺可以實現對海量視頻數據進行實時流計算。

1 系統設計

1.1 系統總體架構

本系統主要分成四層分別是視頻采集層、視頻處理層、視頻緩沖層、視頻分析層、視頻存儲層。系統架構圖如圖1所示。

圖1 系統總體架構

1.2 系統組件

Spark是一個基于分布式的內存計算引擎，主要基于分布式彈性數據集，即RDD實現。Spark Streaming是Spark提供的對于大數據進行實時計算的一種流式處理框架。它的底層是基于Spark Core和內存的大數據實時計算模型，它的底層最核心組件是RDD。同時針對實時計算的特點，在RDD之上，進行了一層封裝，叫做DStream。Spark Streaming是采用將流數據分解成一系列批處理作業來進行流式計算的。Spark Streaming常接收Kafka、HDFS等各種來源的實時輸入數據，進行處理后，處理結果保存在Hbase、hdfs等結構型和非結 構型數據庫中。由于利用內存來保存基于MapReduce算法過程的中間輸出和結果，這對于圖像處理和機器學習方法中使用頻率較高的迭代計算而言大大提升了工作效率。

OpenCV是一個跨平臺計算機視覺和機器學習軟件庫，它輕量級而且高效由一系列C函數和少量C++類構成，實現了圖像處理和計算機視覺方面的很多通用算法。JavaCV 是一款基于JavaCPP調用方式對Opencv封裝的一個開源視覺處理庫，里面集成了Opencv、FFmpeg等多種開源計算機視覺庫。由于使用spark大數據平臺也需要使用java語言進行編程，所以最終決定選用javacv來調用opencv視頻處理庫。

Kafka消息隊列作為視頻數據的緩沖組件為視頻流數據提供了可持久化的存儲。Kafka對消息保存時根據Topic進行歸類，發送消息者稱為Producer，消息接受者稱為Consumer，同類型的消費者建立Consumer group分組，此外Kafka集群有多個Kafka實例組成，每個實例稱為broker。使用Kafka消息隊列可以實現異步處理視頻數據，將視頻采集層和視頻處理層解耦。提高了系統平臺的靈活性和整體性能。

這里選用yolov5訓練模型來做視頻分析的原因是雖然spark自帶的MLlib和GraphX也集成了機器學習算法庫，但是在視頻分析方面還不是很成熟，特別是在使用GPU做模型訓練和分析方面支持較少。而yolov5作為專門的對象識別算法庫具有速度快，模型小，準確度高等特點在視頻分析領域有著非常高的市場占有率。YOLO的原理是將對象檢測重新定義為一個回歸問題，所以不需要發雜的管道。這一特性保障了yolo處理實時視頻流的速度，延遲小于25ms。

2 系統功能

2.1 視頻采集層

國內工業園區的視頻監控系統主要由高清攝像頭、光纜、矩陣、硬盤錄像機和編解碼器等組成。使用SparkStreaming為每一個視頻數據源啟動對應的Reciver（接收器），接收器以任務的形式運行在應用的Executor進程中，從輸入源接收數據，把數據分組為小的批次（batch），并保存為RDD，最后提交Spark Job執行。同時為了合理分配集群資源，使用YARN資源調度器動態獲取每個節點的資源使用信息，智能的選擇在負載較輕的節點中啟動接收器。接收到視頻流數據后調用opencv視頻處理庫中的VideoCapture方法進行視頻流解碼，然后使用opencv將視頻流數據轉換為視頻幀，并進行圖像預處理、特征提取與圖像識別等操作然后根據預處理結果進行分類。由于OpenCV操作的基本對象為Mat數據結構，因此我們先將視頻幀轉換為Mat數據結構的數據，再將Mat結構數據轉換為連續字節數組的格式。最后將格式轉化完成的圖像數據與設備的id和url及需要使用的算法模型分類封裝成一個json對象，json格式如下：

其中：

equipmentId：視頻采集設備的唯一ID。

timestamp：獲取視頻幀的時間。

MatType：該視頻幀需要執行的算法模型類型。

imageData：Mat圖像轉換后的連續字節數組。

最后對這些處理后的數據根據不同的視頻檢測類型使用Kafka Producer將數據發送至不同類型的kafka topic。

2.2 視頻緩沖層

視頻數據緩沖層部署Kafka集群作為緩沖隊列。首先我們需要根據處理層使用的模型算法建立對應的topic，然后前端設備采集到的視頻數據經過處理后分成不同類型的producer將視頻數據發送到對應的topic。最后由視頻處理層根據模型算法類型進行consumer group分組消費topic中的視頻數據，使用對應的算法模型做視頻檢測。

2.3 視頻處理層

視頻處理層使用OpenCV、yolov5以及Spark Streaming進行構建。該組件會從Kafka消息隊列中獲取流式數據，然后在spark平臺上進行特征提取后形成RDD，接下來運用yolov5對視頻數據進行檢測最后將處理后的文件和數據存放到分布式數據庫HDFS和Hbase中。

2.4 視頻存儲層

本系統的視頻數據存儲層選用Hadoop HDFS分布式文件系統和HBsae nosql數據庫。并將分析后的視頻流數據壓縮后儲存在HDFS文件系統中。Hbase作為非關系型數據庫主要用作存儲視頻分析后提取出來的特征描述。

3 實驗結果展示及分析

為了驗證本系統對海量視頻并行分析的性能提升，現部署兩套環境：一套使用5臺浪潮服務器來搭建spark集群平臺做視頻檢測。另一套使用單節點部署做視頻檢測。

如圖2所示實驗將視頻路數作為自變量，視頻檢測時間（單位：ms）作為因變量.視頻全部采用rtsp視頻流文件。

圖2 檢測時間與集群節點數量的關系

從實驗結果可以看出隨著視頻路數的增多，使用spark集群處理視頻的效率優勢越大。

4 結論與展望

對于海量視頻流檢測需要大數據平臺來作為支撐。像Spark、Kafka等這樣的開源分布式技術能夠用來構建高可用、高性能的系統平臺。本文并基于此分布式平臺來進行視頻流分析。同時利用成熟的視頻分析模型yolov5來做視頻檢測，提高了視頻分析的準確度和及時性。最后使用分布式大數據文件系統hdfs存儲處理后的視頻文件，非關系型數據庫hbase存儲提取的圖片的特征json數據，解決了大數據存儲問題。綜上該系統為解決在工業互聯網如火如荼的大環境下，化工園區攝像頭數量暴增，視頻分析算法需求多樣化，檢測結果及時性要求高的問題提供了新的解決方案。但是該系統的部署對服務器資源要求較高且部署難度較大，所以接下來考慮將容器化部署作為下一階段的研究目標。