在非結(jié)構(gòu)化數(shù)據(jù)領(lǐng)域佣渴，技術(shù)帶來了前所未有的爆炸性變化辛润。移動(dòng)設(shè)備砂竖、Web站點(diǎn)鹃答、社交媒體挣跋、科學(xué)儀器、衛(wèi)星舟肉、IoT設(shè)備以及監(jiān)控?cái)z像頭這樣的數(shù)據(jù)源每秒鐘都會(huì)產(chǎn)生大量的圖片和視頻路媚。
核心要點(diǎn)
- 為了可靠且高效地處理大規(guī)模的視頻流數(shù)據(jù)整慎，需要有一個(gè)可擴(kuò)展裤园、能容錯(cuò)拧揽、松耦合的分布式系統(tǒng)腺占；
- 本文中的示例應(yīng)用使用開源的技術(shù)來構(gòu)建這樣的系統(tǒng)，這些技術(shù)包括OpenCV衰伯、Kafka和Spark意鲸。另外，還可以使用Amazon S3或HDFS進(jìn)行存儲(chǔ)教翩；
- 該系統(tǒng)包含了三個(gè)主要的組件：視頻流收集器（Video Stream Collector）贪壳、流數(shù)據(jù)緩沖（Stream Data Buffer）以及視頻流處理器（Video Stream Processor）闰靴；
- 視頻流收集器需要與一個(gè)網(wǎng)絡(luò)攝像機(jī)（IP camera）集群協(xié)同工作蚂且，這些攝像機(jī)提供視頻內(nèi)容的實(shí)時(shí)流數(shù)據(jù)杏死，并且還會(huì)使用OpenCV視頻處理庫把視頻流轉(zhuǎn)換為幀淑翼，將數(shù)據(jù)以JSON的格式傳遞給Kafka Broker玄括，供流數(shù)據(jù)緩沖組件使用遭京；
- 視頻流處理組件基于Apache Spark構(gòu)建哪雕，同樣會(huì)使用OpenCV進(jìn)行視頻流數(shù)據(jù)的處理鲫趁。
在非結(jié)構(gòu)化數(shù)據(jù)領(lǐng)域饮寞，技術(shù)帶來了前所未有的爆炸性變化幽崩。移動(dòng)設(shè)備、Web站點(diǎn)陌选、社交媒體咨油、科學(xué)儀器、衛(wèi)星赚爵、IoT設(shè)備以及監(jiān)控?cái)z像頭這樣的數(shù)據(jù)源每秒鐘都會(huì)產(chǎn)生大量的圖片和視頻冀膝。
管理和有效分析這些數(shù)據(jù)是一個(gè)很大的挑戰(zhàn)窝剖，我們可以考慮一下某個(gè)城市的視頻監(jiān)控網(wǎng)絡(luò)赐纱。試圖監(jiān)控每個(gè)攝像頭的視頻流來發(fā)現(xiàn)感興趣的對象和事件是不現(xiàn)實(shí)且低效的疙描。相反淫痰，計(jì)算機(jī)視覺（computer vision，CV）庫能夠處理這些視頻流并提供智能的視頻分析和對象探測結(jié)果烈评。
但是讲冠，傳統(tǒng)的CV系統(tǒng)有一定的局限性竿开。在傳統(tǒng)的視頻分析系統(tǒng)中疯攒，帶有CV庫的服務(wù)器會(huì)同時(shí)收集和處理數(shù)據(jù)列荔，所以服務(wù)器的故障將會(huì)丟失視頻流數(shù)據(jù)。探測節(jié)點(diǎn)故障并將處理進(jìn)程轉(zhuǎn)移到其他節(jié)點(diǎn)上可能會(huì)導(dǎo)致碎片化的數(shù)據(jù)署恍。
有很多的實(shí)際任務(wù)將大數(shù)據(jù)相關(guān)的技術(shù)推進(jìn)到了視頻流分析領(lǐng)域：并行且按需處理大規(guī)模的視頻流盯质、從視頻幀中抽取不同的信息集概而、使用不同的機(jī)器學(xué)習(xí)庫分析數(shù)據(jù)唤殴、將分析得到的數(shù)據(jù)以流的方式發(fā)送到應(yīng)用的不同組件中以便于后續(xù)處理、將處理后的數(shù)據(jù)以不同的格式進(jìn)行輸出等等到腥。
視頻流分析——?jiǎng)幼鞲袘?yīng)
為了可靠且高效地處理大規(guī)模的視頻流數(shù)據(jù)，需要有一個(gè)可擴(kuò)展蔚袍、能容錯(cuò)乡范、松耦合的分布式系統(tǒng)。在本文所討論的視頻流分析中啤咽，我們將會(huì)討論這些原則晋辆。
視頻流分析包括如下的類型：
- 對象跟蹤（object tracking）
- 動(dòng)作感應(yīng)（motion detection）
- 面部識(shí)別（face recognition）
- 手勢識(shí)別（gesture recognition）
- 增強(qiáng)現(xiàn)實(shí)（augmented reality）
- 圖像分割（image segmentation）
本文中示例應(yīng)用的使用場景將會(huì)是視頻流中的動(dòng)作感應(yīng)鳞青。
動(dòng)作感應(yīng)指的是查找一個(gè)物體（通常會(huì)是人）相對于其周邊環(huán)境位置變化的過程厚脉。它大多數(shù)用于持續(xù)監(jiān)視特定區(qū)域的視頻監(jiān)控系統(tǒng)。CV庫提供的算法會(huì)分析這種攝像機(jī)所提供的實(shí)時(shí)視頻并查找所發(fā)生的動(dòng)作中捆。如果感應(yīng)到動(dòng)作的話幼东，將會(huì)觸發(fā)一個(gè)事件脓杉，這個(gè)事件可以發(fā)送消息給應(yīng)用或提示用戶尿赚。
在本文中，用于視頻流分析的應(yīng)用由三個(gè)主要的組件組成：
- 視頻流收集器（video stream collector）
- 視頻數(shù)據(jù)緩沖（stream data buffer）
- 視頻流處理器（video stream processor）
視頻流收集器要接受來自網(wǎng)絡(luò)攝像機(jī)集群的視頻流數(shù)據(jù)。這個(gè)組件將視頻幀序列化為流數(shù)據(jù)緩沖斩芭，這是一個(gè)用于視頻流數(shù)據(jù)的可容錯(cuò)數(shù)據(jù)隊(duì)列（queue）。視頻流處理器消費(fèi)緩沖中的流數(shù)據(jù)并進(jìn)行處理。這個(gè)組件將會(huì)使用視頻處理算法在視頻流數(shù)據(jù)中探測動(dòng)作迷殿。最后翰苫，處理過的數(shù)據(jù)或圖片文件會(huì)存儲(chǔ)到S3 bucket或HDFS目錄中导披。這個(gè)視頻流處理系統(tǒng)在設(shè)計(jì)時(shí)使用了OpenCV、Apache Kafka以及Apache Spark框架模蜡。
OpenCV忍疾、Kafka和Spark簡介
如下簡單介紹了相關(guān)的框架字币。
OpenCV
OpenCV（Open Source Computer Vision Library）是一個(gè)基于BSD許可證開源的庫。這個(gè)庫使用C++編寫判没，但是也提供了Java API辟犀。OpenCV包含了數(shù)百個(gè)CV算法，能夠用來處理和分析圖片及視頻文件。請參考該文檔了解更多細(xì)節(jié)税稼。
Kafka
Apache Kafka是一個(gè)分布式的流平臺(tái)兜蠕，它提供了一個(gè)發(fā)布和訂閱流記錄（streams of records）的系統(tǒng)盗舰。這些記錄能夠按照可容錯(cuò)的方式進(jìn)行存儲(chǔ)，消費(fèi)者可以處理這些數(shù)據(jù)爷绘。關(guān)于Kafka的更多信息，請參見該文檔。
Spark
Apache Spark是一個(gè)快速、通用的集群計(jì)算系統(tǒng)烘苹。它提供了用于SQL和結(jié)構(gòu)化數(shù)據(jù)處理的模塊档悠、用于機(jī)器學(xué)習(xí)的MLlib、用于圖像處理的GraphX以及Spark Streaming。該文檔中包含了關(guān)于Spark的更多細(xì)節(jié)。
系統(tǒng)架構(gòu)
圖1展現(xiàn)了視頻流分析系統(tǒng)的架構(gòu)圖。
[圖片上傳中...(image-df0f69-1547015523382-1)]
<small>圖1 視頻流分析系統(tǒng)的架構(gòu)圖</small>
設(shè)計(jì)與實(shí)現(xiàn)
本示例應(yīng)用的代碼可以通過GitHub獲取炫刷。
如下的章節(jié)介紹了樣例中視頻流收集器噩咪、流數(shù)據(jù)緩沖以及視頻流處理器的設(shè)計(jì)與實(shí)現(xiàn)細(xì)節(jié)。
視頻流收集器
視頻流收集器會(huì)與一個(gè)網(wǎng)絡(luò)攝像機(jī)集群協(xié)同工作厕隧，這些攝像機(jī)會(huì)提供實(shí)時(shí)視頻建丧。該組件必須要從每個(gè)攝像機(jī)讀取數(shù)據(jù)并將視頻流轉(zhuǎn)換為一系列的視頻幀翎朱。為了區(qū)分每個(gè)網(wǎng)絡(luò)攝像機(jī)，收集器要通過camera.url和camera.id屬性維護(hù)攝像機(jī)ID與URL之間的映射，這兩個(gè)屬性會(huì)在stream-collector.properties文件中定義。這些屬性在定義時(shí)，可以按照逗號(hào)分隔的格式定義攝像機(jī)URL和ID的列表革砸。不同的攝像機(jī)可能會(huì)以不同的規(guī)格來提供數(shù)據(jù)册踩，比如編解碼器（codec）、分辨率或者每秒的幀數(shù)。在通過視頻流創(chuàng)建幀的時(shí)候，收集器必須要保留這些細(xì)節(jié)數(shù)據(jù)俯邓。
收集器使用OpenCV視頻處理庫將視頻流轉(zhuǎn)換為幀朦蕴。每幀都會(huì)調(diào)整為所需的分辨率（比如640x480）。OpenCV將每幀或每幅圖片存儲(chǔ)為Mat對象。Mat需要轉(zhuǎn)換為可連續(xù)的（字節(jié)數(shù)組）形式，在這個(gè)過程要保留幀的完整信息，比如rows、cols和type。視頻流收集器使用如下的JSON信息格式來存儲(chǔ)這些細(xì)節(jié)。
<pre style="margin: 0px; padding: 0px; overflow-wrap: break-word; white-space: pre-wrap; font-family: "courier new", courier, monospace;">
{"cameraId":"cam-01","timestamp":1488627991133,"rows":12,"cols":15,"type":16,"data":"asdfh"}</pre>
cameraId是攝像機(jī)的唯一ID盈滴。timestamp是幀生成的時(shí)間。rows、cols和type是OpenCV Mat特定的細(xì)節(jié)信息。data是基于base-64編碼的字符串胸囱，代表了幀的字節(jié)數(shù)組箕宙。
視頻流收集器使用Gson庫將數(shù)據(jù)轉(zhuǎn)換為JSON消息锅很，這些消息會(huì)被發(fā)送至video-stream-event topic上扔仓。它會(huì)使用KafkaProducer客戶端將JSON消息發(fā)送至Kafka broker。KafkaProducer會(huì)將每個(gè)key發(fā)送至相同的分區(qū)并保證消息的順序慷吊。
<pre style="margin: 0px; padding: 0px; overflow-wrap: break-word; white-space: pre-wrap; font-family: "courier new", courier, monospace;">
JsonObject obj = new JsonObject(); obj.addProperty("cameraId",cameraId); obj.addProperty("timestamp", timestamp); obj.addProperty("rows", rows); obj.addProperty("cols", cols); obj.addProperty("type", type); obj.addProperty("data", Base64.getEncoder().encodeToString(data)); String json = gson.toJson(obj); producer.send(new ProducerRecord<String, String>(topic,cameraId,json),new EventGeneratorCallback(cameraId));</pre>
<small>圖2 以JSON消息的格式發(fā)送圖片的代碼片段</small>
Kafka的設(shè)計(jì)場景主要是用來處理較小的文本信息，但是這里的JSON信息中包含了視頻幀的字節(jié)數(shù)組哼绑，它會(huì)比較大（比如能夠達(dá)到1.5MB）双谆，所以Kafka在處理較大的信息之前竟稳，需要進(jìn)行配置的變更。如下的KafkaProducer屬性需要進(jìn)行調(diào)整：
batch.sizemax.request.sizecompression.type
請參見Kafka文檔的Producer Configs章節(jié)以及本項(xiàng)目 GitHub上的代碼和屬性文件了解更多細(xì)節(jié)。
視頻數(shù)據(jù)緩沖
為了無丟失地處理大量的視頻流數(shù)據(jù)惜论，將這些視頻數(shù)據(jù)保存到臨時(shí)存儲(chǔ)中就是非常必要的了句喜。對于收集器所產(chǎn)生的數(shù)據(jù)，Kafka broker的作用就像是一個(gè)緩沖隊(duì)列（buffer queue）睡毒。Kafka使用文件系統(tǒng)來存儲(chǔ)信息违寞，對這些信息的保存時(shí)間是可以配置的摄悯。
如果在處理之前就將數(shù)據(jù)保存到存儲(chǔ)中管跺，那就能保證它的持久性豁跑，同時(shí)還能提升系統(tǒng)的整體性能，因?yàn)樘幚砥骺梢愿鶕?jù)負(fù)載在不同的時(shí)間按照不同的速度來處理數(shù)據(jù)碍讨。當(dāng)數(shù)據(jù)的生成速度超過數(shù)據(jù)的處理速度時(shí)，這種方式能夠提升系統(tǒng)的可靠性覆获。
Kafka能夠保證在單個(gè)分區(qū)中給定topic的消息順序凝果。如果數(shù)據(jù)的順序比較重要的話欧瘪，在處理這類數(shù)據(jù)時(shí)冗茸，該特性就是非常有用的。為了存儲(chǔ)較大的信息匹中，在Kafka服務(wù)器的server.properties文件中需要調(diào)整如下的配置：
message.max.bytesreplica.fetch.max.bytes
請參見Kafka文檔的Broker Configs章節(jié)來了解這些屬性的詳細(xì)信息夏漱。
視頻流處理器
視頻流處理器會(huì)執(zhí)行下面的三個(gè)步驟：
- 從Kafka broker中以
VideoEventDatadataset的形式讀取JSON信息； - 根據(jù)攝像機(jī)ID對
VideoEventDatadataset進(jìn)行分組并將其傳遞給視頻流處理器顶捷； - 根據(jù)JSON數(shù)據(jù)創(chuàng)建一個(gè)Mat對象并處理視頻流數(shù)據(jù)挂绰。
視頻流收集器是基于Apache Spark構(gòu)建的。Spark提供了Spark Streaming API服赎，該API能夠使用離散的流（discretized stream）或DStream葵蒂，并且還提供了基于dataset的新Structured Streaming API交播。本應(yīng)用中的視頻流收集器使用Structured Streaming API來消費(fèi)和處理來自Kafka的數(shù)據(jù)。需要注意的是践付，本應(yīng)用所處理的格式化數(shù)據(jù)是JSON消息的形式秦士，視頻流處理器所要處理的非結(jié)構(gòu)化視頻數(shù)據(jù)會(huì)作為JSON消息的屬性。Spark的文檔這樣寫道“Structured Streaming提供了快速荔仁、可擴(kuò)展、容錯(cuò)芽死、端到端且保證僅執(zhí)行一次的流處理功能乏梁，用戶無需考慮流的相關(guān)事宜”。這也是視頻流處理器圍繞Spark的 Structured Streaming設(shè)計(jì)的原因所在关贵。Structured Streaming為結(jié)構(gòu)化的文本數(shù)據(jù)提供了內(nèi)置的支持遇骑，并且支持聚合查詢（aggregation queries）的狀態(tài)管理。該引擎還提供了一些其他的特性揖曾，比如處理非聚合查詢以及datasets外部的狀態(tài)管理（Spark 2.2.0的新特性）落萎。
為了處理較大的信息，如下的Kafka消費(fèi)者配置必須要傳遞給Spark引擎：
max.partition.fetch.bytesmax.poll.records
請參見Kafka文檔的Consumer Configs章節(jié)了解這些屬性的更多信息炭剪。
該組件的主類是VideoStreamProcessor练链。這個(gè)類首先創(chuàng)建一個(gè)SparkSession對象，它是與Spark SQL引擎協(xié)作的入口點(diǎn)奴拦。下一步是定義傳入消息的模式（schema）媒鼓，這樣的話，Spark就能夠使用該模式將消息從字符串格式解析為JSON格式错妖。Spark的bean encoder能夠?qū)⑵滢D(zhuǎn)換為Dataset<VideoEventData>绿鸣。VideoEventData是一個(gè)Java bean類，它會(huì)持有JSON消息的數(shù)據(jù)暂氯。
<pre style="margin: 0px; padding: 0px; overflow-wrap: break-word; white-space: pre-wrap; font-family: "courier new", courier, monospace;">
Dataset<VideoEventData> ds = spark.readStream().format("kafka") .option("kafka.bootstrap.servers",prop.getProperty("kafka.bootstrap.servers")) .option("subscribe",prop.getProperty("kafka.topic")) .option("kafka.max.partition.fetch.bytes",prop.getProperty("kafka.max.partition.fetch.bytes")) .option("kafka.max.poll.records", prop.getProperty("kafka.max.poll.records")) .load().selectExpr("CAST(value AS STRING) as message") .select(functions.from_json(functions.col("message"),schema).as("json")) .select("json.*").as(Encoders.bean(VideoEventData.class)); </pre>
<small>3 spark streaming處理kafka消息的代碼片段</small>
接下來潮模，groupByKey會(huì)根據(jù)攝像機(jī)的ID對數(shù)據(jù)集進(jìn)行分組，得到KeyValueGroupedDataset<String, VideoEventData>痴施。它會(huì)使用一個(gè)mapGroupsWithState transformation并作用于一組VideoEventData (Iterator<VideoEventData>)擎厢，這些數(shù)據(jù)代表了本次批處理的視頻幀，會(huì)根據(jù)攝像機(jī)ID進(jìn)行分組辣吃。這個(gè)transformation會(huì)首先檢查上一條已被處理的VideoEventData（視頻幀）是否存在锉矢，并將其傳遞給視頻處理器用于下一步的處理。在視頻處理之后齿尽，上一條被處理的VideoEventData（視頻幀）會(huì)被返回沽损，而狀態(tài)已更新。為了啟動(dòng)流應(yīng)用循头，需要在console sink和update output模式下針對dataset調(diào)用writeStream方法绵估。
請閱讀GitHub上的屬性文件和代碼了解更多細(xì)節(jié)炎疆。
技術(shù)和工具 Technologies and Tools
如下的表格列出了該視頻流分析系統(tǒng)所用到的技術(shù)和工具
| <small>技術(shù)和工具</small> | <small>版本</small> | <small>下載URL</small> |
| <small>JDK</small> | 1.8 | <small>http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html</small> |
| <small>Maven</small> | <small>3.3.9</small> | <small>https://maven.apache.org/download.cgi</small> |
| <small>ZooKeeper</small> | <small>3.4.8</small> | <small>https://zookeeper.apache.org/releases.html</small> |
| <small>Kafka</small> | <small>2.11-0.10.2.0</small> | <small>http://kafka.apache.org/downloads.html</small> |
| <small>Spark</small> | <small>2.2.0</small> | <small>http://spark.apache.org/downloads.html</small> |
| <small>OpenCV</small> | <small>3.2.0</small> | <small>http://opencv.org/releases.html</small> |
請參考文檔了解這些工具的安裝和配置。Kafka文檔和Spark文檔詳細(xì)介紹了如何搭建環(huán)境以及如何以獨(dú)立模式和集群模式運(yùn)行應(yīng)用国裳。要安裝OpenCV的話形入，請參考OpenCV文檔。OpenCV也可以通過Anaconda安裝缝左。
構(gòu)建與部署
本節(jié)詳細(xì)介紹了如何構(gòu)建和運(yùn)行示例應(yīng)用的視頻流收集器和視頻流處理器組件亿遂。這個(gè)應(yīng)用既能用來處理離線的視頻文件，也能處理實(shí)時(shí)的攝像機(jī)數(shù)據(jù)渺杉，但是在這里我們配置它分析一個(gè)離線示例視頻文件蛇数。請按照下述的步驟構(gòu)建和運(yùn)行應(yīng)用。
1.下載并安裝上述表格中所列的工具是越。確保ZooKeeper和Kafka服務(wù)器已處于啟動(dòng)運(yùn)行的狀態(tài)耳舅；
2. 該應(yīng)用會(huì)使用OpenCV原生庫（.dll或.so），所以使用System.loadLibrary()加載它們倚评。在系統(tǒng)環(huán)境變量中設(shè)置這些原生庫的目錄路徑或者將路徑作為命令行參數(shù)傳遞進(jìn)來浦徊。例如，對于64位的Windows機(jī)器天梧，原生庫文件（opencv_java320.dll和opencv_ffmpeg320_64.dll）的路徑將會(huì)是{OpenCV Installation Directory} \build\java\x64盔性。
3.stream-collector.properties文件會(huì)將Kafka topic作為video-stream-event。在Kafka中創(chuàng)建該topic和分區(qū)（partitions）呢岗。我們可以使用kafka-topic命令來創(chuàng)建topic和分區(qū)纯出；
4. stream-processor.properties文件有一個(gè)processed.output.dir屬性，它指定了處理后圖片的保存路徑敷燎。創(chuàng)建文件并為該屬性設(shè)置目錄路徑暂筝；
5.stream-collector.properties文件有一個(gè)camera.url屬性，它保存了視頻文件或視頻源的路徑或URL硬贯。確保路徑或URL的正確性;
6.檢查VideoStreamCollector和VideoStreamProcessor組件中的log4j.properties文件焕襟，設(shè)置stream-collector.log和stream-processor.log文件的目錄路徑。檢查應(yīng)用在這些日志文件中所生成的日志信息饭豹，如果應(yīng)用在運(yùn)行中出現(xiàn)錯(cuò)誤的話鸵赖，這些日志會(huì)有一定的用處；
7. 應(yīng)用會(huì)使用來自O(shè)penCV JAR文件的OpenCV API拄衰，但是在Maven中央倉庫中并沒有包含OpenCV JAR文件它褪。在本應(yīng)用中打包了OpenCV JAR文件，可以將其安裝到本地Maven倉庫中翘悉。在pom.xml文件中茫打，maven-install-plugin已經(jīng)進(jìn)行了配置，它會(huì)與clean階段（phase）關(guān)聯(lián)來安裝這個(gè)JAR文件。為了將OpenCV JAR安裝到本地Maven倉庫中老赤，切換至video-stream-processor文件夾并運(yùn)行如下命令：
<pre style="margin: 0px; padding: 0px; overflow-wrap: break-word; white-space: pre-wrap; font-family: "courier new", courier, monospace;">
mvn clean</pre>
8. 為了讓應(yīng)用的邏輯盡可能簡單轮洋，VideoStreamProcessor 只處理新的消息。VideoStreamProcessor應(yīng)該要先于VideoStreamCollector組件啟動(dòng)并運(yùn)行抬旺。如果要通過Maven運(yùn)行VideoStreamProcessor弊予，切換至video-stream-processor文件夾并執(zhí)行如下命令：
<pre style="margin: 0px; padding: 0px; overflow-wrap: break-word; white-space: pre-wrap; font-family: "courier new", courier, monospace;">
mvn clean package exec:java -Dexec.mainClass="com.iot.video.app.spark.processor.VideoStreamProcessor"</pre>
9.在VideoStreamProcessor啟動(dòng)之后，我們接下來就可以啟動(dòng)VideoStreamCollector組件了开财。切換至video-stream-collector文件夾并執(zhí)行如下命令：
<pre style="margin: 0px; padding: 0px; overflow-wrap: break-word; white-space: pre-wrap; font-family: "courier new", courier, monospace;">
mvn clean package exec:java -Dexec.mainClass="com.iot.video.app.kafka.collector.VideoStreamCollector" -Dexec.cleanupDaemonThreads=false</pre>
GitHub項(xiàng)目上打包了一個(gè)sample.mp4視頻文件汉柒。這個(gè)視頻文件的URL和ID已經(jīng)通過camera.url和camera.id屬性在stream-collector.properties文件中進(jìn)行了配置。在處理完視頻文件之后责鳍，圖片將會(huì)存儲(chǔ)到預(yù)先配置的目錄中（參見第4步）碾褂。圖4展現(xiàn)了應(yīng)用的示例輸出。
[圖片上傳中...(image-88112-1547015523380-0)]
<small>圖4 動(dòng)作感應(yīng)的示例輸出</small>
這個(gè)應(yīng)用能夠配置并處理多個(gè)視頻源（包括離線的和實(shí)時(shí)的）薇搁。例如斋扰，除了sample.mp4文件之外渡八，假設(shè)我們還要添加來自webcam的feed視頻啃洋，編輯stream-collector.properties文件，在camera.url屬性中添加逗號(hào)分隔的整數(shù)值（第一個(gè)webcam對應(yīng)0屎鳍，第二個(gè)webcam對應(yīng)2宏娄，以此類推），添加對應(yīng)的攝像機(jī)ID到camera.id屬性中（cam-01逮壁，cam-02等等）孵坚，同樣要使用逗號(hào)分隔。如下是一個(gè)樣例：
<pre style="margin: 0px; padding: 0px; overflow-wrap: break-word; white-space: pre-wrap; font-family: "courier new", courier, monospace;">
camera.url=../sample-video/sample.mp4,0 </pre>
<pre style="margin: 0px; padding: 0px; overflow-wrap: break-word; white-space: pre-wrap; font-family: "courier new", courier, monospace;">
camera.id=vid-01,cam-01</pre>
結(jié)論
大規(guī)模的視頻流分析需要有一個(gè)大數(shù)據(jù)技術(shù)作為支撐的健壯系統(tǒng)窥淆。像OpenCV卖宠、Kafka和Spark這樣的開源技術(shù)能夠用來構(gòu)建可容錯(cuò)的分布式系統(tǒng)，并基于此來進(jìn)行視頻流分析忧饭。我們使用OpenCV和Kafka構(gòu)建視頻流收集組件扛伍，它會(huì)從不同的源接收視頻流，并將其發(fā)送至視頻流緩沖組件词裤。Kafka作為視頻數(shù)據(jù)的緩沖組件刺洒，它為流數(shù)據(jù)提供了可持久化的存儲(chǔ)。視頻流處理組件使用OpenCV以及Spark的Structured Streaming進(jìn)行構(gòu)建吼砂。這個(gè)組件會(huì)從流數(shù)據(jù)緩沖中獲取流式數(shù)據(jù)逆航，并對數(shù)據(jù)進(jìn)行分析。處理后的文件可以放到預(yù)先配置好的HDFS或S3 bucket中渔肩。我們使用動(dòng)作感應(yīng)作為視頻流分析的用例因俐，并提供了一個(gè)示例應(yīng)用。
參考資料
- Online Security Analytics on Large Scale Video Surveillance System (Spark Summit East 2016)
- Large Scale Image Processing in Real-Time Environments with Kafka (CS & IT)
- OpenCV 文檔
- Kafka 文檔
- Apache Spark 文檔
- ZooKeeper 文檔
[原文](Video Stream Analytics Using OpenCV, Kafka and Spark Technologies)
基于Spark的實(shí)時(shí)視頻分析系統(tǒng)
