【摘要】如何對海量的DPI數(shù)據(jù)進(jìn)行實時的采集以及處理是運營商研究的熱點，傳統(tǒng)基于MapReduce的批處理模式難以滿足流式計算實時性要求，因此首先介紹了流式處理相關(guān)概念，然后分析了目前流行的流式計算技術(shù)，提出一種基于流式計算的DPI數(shù)據(jù)處理方案，并應(yīng)用在實際項目中，滿足電信運營商對數(shù)據(jù)處理實時性的要求，最后通過實踐總結(jié)了流式處理的應(yīng)用場景。
【關(guān)鍵詞】DPI；流式計算；數(shù)據(jù)處理
doi:10.3969/j.issn.1006-1010.2018.01.000 中圖分類號：TN399 文獻(xiàn)標(biāo)志碼：A 文章編號：1006-1010(2018)01-0000-00
引用格式：范家杰,田熙清. 基于流式計算的DPI數(shù)據(jù)處理方案及實踐[J]. 移動通信, 2017,42(1): 00-00.
Scheme and Practice of DPI Data Processing Based on Stream Computing
FAN Jiajie, TIAN Xiqing
(Guangdong Research Institute of China Telecom Co., Ltd., Guangzhou 510630, China)
[Abstract] How to collect and process the massive DPI data in real time is the hotspot of telecom operators. The traditional batch mode MapReduce is difficult to meet the real-time requirements based on stream calculation, so this paper firstly introduces the related concepts of stream computing, and then analyzes the current popular streaming technology, presents a stream computing based on the DPI data processing program. This scheme is applied to practical projects to meet the requirements of telecom operators for real-time data processing. Finally, the application scenarios of streaming processing are summarized through practice.
[Key words] DPI; stream computing; data processing
1 引言
隨著移動互聯(lián)網(wǎng)的不斷發(fā)展以及各類智能設(shè)備日益深入民眾日常生活中，人類社會產(chǎn)生的數(shù)據(jù)量正在以指數(shù)級快速增長，人類已經(jīng)正式邁入大數(shù)據(jù)時代[1]。如今，運營商能夠獲得的用戶數(shù)據(jù)越來越豐富，通過DPI（Deep Packet Inspector，深度分組檢測）分析技術(shù)，能夠較好地識別網(wǎng)絡(luò)上的流量類別、應(yīng)用層上的應(yīng)用種類等[2]。在這個“數(shù)據(jù)為王”的時代，如何充分利用這筆重要的戰(zhàn)略資產(chǎn)已成為重中之重。
數(shù)據(jù)規(guī)模的快速增長給大數(shù)據(jù)分析處理帶來了巨大的挑戰(zhàn)，尤其是在通信行業(yè)，數(shù)據(jù)越發(fā)呈現(xiàn)出無限性、突發(fā)性和實時性等特征[3]，傳統(tǒng)的基于MapReduce的批處理模式難以滿足數(shù)據(jù)實時性的要求，而能否在第一時間獲得數(shù)據(jù)所蘊(yùn)含的信息決定了數(shù)據(jù)的價值。因此，流式處理技術(shù)成為大數(shù)據(jù)技術(shù)研究的新熱點[4]。流式處理能夠針對數(shù)據(jù)的變化進(jìn)行實時處理，能夠在秒級獲得處理結(jié)果，特別適合一些對時效性要求很高的場景。
本文結(jié)合電信運營商的需求，對DPI數(shù)據(jù)進(jìn)行實時的采集及處理，提出一種基于流式計算的DPI數(shù)據(jù)處理方案，能夠?qū)@得DPI數(shù)據(jù)實時信息的時延降低到分鐘級，甚至秒級，實現(xiàn)對電信用戶上網(wǎng)信息的實時處理、監(jiān)測及分類匯總，為之后進(jìn)行的大數(shù)據(jù)應(yīng)用提供了良好基礎(chǔ)。
2 流式處理概述
傳統(tǒng)基于MapReduce大數(shù)據(jù)處理技術(shù)實際上是一種批處理方式，如圖1所示。批處理模式首先要完成數(shù)據(jù)的累積和存儲，然后Hadoop客戶端將數(shù)據(jù)上傳到HDFS上，最后才啟動Map/Reduce進(jìn)行數(shù)據(jù)處理，處理后再寫入到HDFS。這種方式必須要所有數(shù)據(jù)都要準(zhǔn)備好，然后統(tǒng)一進(jìn)行集中計算和價值發(fā)現(xiàn)，無法滿足實時性的要求。
圖1 批處理流程
圖1 基于MapReduce的大數(shù)據(jù)處理
2015年，Nathan Marz提出了實時大數(shù)據(jù)處理框架Lambda架構(gòu)[5]，整合了離線計算和實時計算，能夠滿足實時系統(tǒng)高容錯、低時延和可擴(kuò)展等要求，并且可集成Hadoop、Kafka、Storm、Spark及HBase等各類大數(shù)據(jù)組件。
一個典型的Lambda架構(gòu)如圖2所示，主要使用的場景是邏輯復(fù)雜且延遲低的程序。數(shù)據(jù)會分別灌入實時系統(tǒng)和批處理系統(tǒng)，然后各自輸出自己的結(jié)果，結(jié)果會在查詢端進(jìn)行合并。
圖2 Lambda架構(gòu)圖
3 流式計算架構(gòu)對比
流式計算對系統(tǒng)的容錯、時延、可擴(kuò)展及可靠性能力提出了很高的要求，當(dāng)前有許多流式計算框架（如Spark streaming[10]、Storm[11]、Kafka Stream[12]、Flink[13]和PipelineDB[14]等）已經(jīng)廣泛應(yīng)用于各行各業(yè)，并且還在不斷迭代發(fā)展，適用的場景也各不相同。
3.1 Spark streaming
Spark是由加州大學(xué)伯克利分校AMP實驗室專門為大數(shù)據(jù)處理而設(shè)計的計算框架[6]。Spark Streaming是建立在Spark上的實時計算框架，是Spark的核心組件之一，通過它內(nèi)置的API和基于內(nèi)存的高效引擎，用戶可以結(jié)合流處理、批處理和交互式查詢開發(fā)應(yīng)用。
Spark Streaming并不像其他流式處理框架每次只處理一條記錄，而是將流數(shù)據(jù)離散化處理，每次處理一批數(shù)據(jù)（DStream），使之能夠進(jìn)行秒級以下的快速批處理，執(zhí)行流程如圖3所示。Spark Streaming的Receiver并行接收數(shù)據(jù)，將數(shù)據(jù)緩存至內(nèi)存中，經(jīng)過延遲優(yōu)化后Spark引擎對短任務(wù)（幾十毫秒）進(jìn)行批處理。這樣設(shè)計的好處讓Spark Streaming能夠同時處理離線處理和流處理問題。
圖3 Spark Streaming執(zhí)行流程
Spark Streaming能在故障報錯下迅速恢復(fù)狀態(tài)，整合了批處理與流處理，內(nèi)置豐富高級算法處理庫，發(fā)展迅速，社區(qū)活躍。毫無疑問，Spark Streaming是流式處理框架的佼佼者。缺點是由于需要累積一批小文件才處理，因此時延會稍大，是準(zhǔn)實時系統(tǒng)。
3.2 Storm
Storm通常被比作“實時的Hadoop”，是Twitter開發(fā)的實時、分布式以及具備高容錯計算系統(tǒng)，可以簡單、可靠地處理大量數(shù)據(jù)流，用戶可以采用任意編程語言來開發(fā)應(yīng)用。
在Storm中，一個用于實時計算的圖狀結(jié)構(gòu)稱之為拓?fù)洌╰opology），拓?fù)涮峤坏郊�，由集群中的主控�?jié)點分發(fā)代碼，分配任務(wù)到工作節(jié)點執(zhí)行。一個拓?fù)渲邪╯pout和bolt兩種角色，其中spout發(fā)送消息，負(fù)責(zé)將數(shù)據(jù)流以tuple元組的形式發(fā)送出去；而bolt則負(fù)責(zé)轉(zhuǎn)換這些數(shù)據(jù)流，在bolt中可以完成映射map、過濾filter等操作，bolt自身也可以隨機(jī)將數(shù)據(jù)發(fā)送給其他bolt。
圖4 Storm數(shù)據(jù)流動
Storm能將數(shù)據(jù)在不同的bolt中流動、移動數(shù)據(jù)，真正實現(xiàn)流式處理，易于擴(kuò)展，靈活性強(qiáng)，高度專注于流式處理。Storm在事件處理與增量計算方面表現(xiàn)突出，能夠以實時方式根據(jù)不斷變化的參數(shù)對數(shù)據(jù)流進(jìn)行處理。
3.3 Kafka Stream
Kafka Stream是Apache Kafka開源項目的一個組成部分，是一個功能強(qiáng)大、易于使用的庫，它使得Apache Kafka擁有流處理的能力。
Kafka Stream是輕量級的流計算類庫，除了Apache Kafka之外沒有任何外部依賴，可以在任何Java程序中使用，使用Kafka作為內(nèi)部消息通訊存儲介質(zhì)，因此不需要為流處理需求額外部署一個集群。
Kafka Stream入門簡單，并且不依賴其他組件，非常容易部署，支持容錯的本地狀態(tài)，延遲低，非常適合一些輕量級流處理的場景。
3.4 Flink
Flink是一個面向分布式數(shù)據(jù)流處理和批量數(shù)據(jù)處理的開源計算平臺，同時支持批處理以及流處理，主要針對流數(shù)據(jù)，將批數(shù)據(jù)視為流數(shù)據(jù)的一個極限特例。
Flink核心是一個流式的數(shù)據(jù)流執(zhí)行引擎，它提供了數(shù)據(jù)分布、數(shù)據(jù)通信以及容錯機(jī)制等功能。流執(zhí)行引擎之上，F(xiàn)link提供了更高層次的API以便用戶使用。Flink還針對某些領(lǐng)域提供了領(lǐng)域庫，例如Flink ML、Flink的機(jī)器學(xué)習(xí)庫等。
Flink適合有極高流處理需求，并有少量批處理任務(wù)的場景。該技術(shù)可兼容原生Storm和Hadoop程序，可在YARN管理的集群上運行。目前Flink最大的局限之一是在社區(qū)活躍度方面，該項目的大規(guī)模部署尚不如其他處理框架那么常見。
3.5 PipeLineDB
PipelineDB是基于PostgreSQL的一個流式計算數(shù)據(jù)庫，效率非常高，通過SQL對數(shù)據(jù)流做操作，并把操作結(jié)果儲存起來。其基本過程是：創(chuàng)建PipelineDB Stream、編寫SQL、對Stream做操作、操作結(jié)果被保存到continuous view。
PipelineDB特點是可以只使用SQL進(jìn)行流式處理，不需要代碼，可以高效可持續(xù)自動處理流式數(shù)據(jù)，只存儲處理后的數(shù)據(jù)，因此非常適合流式數(shù)據(jù)處理，例如網(wǎng)站流量統(tǒng)計、網(wǎng)頁的瀏覽統(tǒng)計等。
3.6 架構(gòu)對比
上文提到的5種流式處理框架對比如表1所示：
表1 流式框架對比
Storm的特點是成熟，是流式處理框架實際上的標(biāo)準(zhǔn)，模型、編程難度都比較復(fù)雜，框架采用循環(huán)處理數(shù)據(jù)，對系統(tǒng)資源，尤其是CPU資源消耗很大，當(dāng)任務(wù)空閑時，需要sleep程序，減少對資源的消耗。Spark Streaming兼顧了批處理以及流式處理，并且有Spark的強(qiáng)大支持，發(fā)展?jié)摿Υ螅cKafka的接口平滑性不夠。Kafka Stream是Kafka的一個開發(fā)庫，具有入門、編程、部署運維簡單的特點，并且不需要部署額外的組件，但對于多維度的統(tǒng)計來說，需要基于不同topic來做分區(qū)，編程模型復(fù)雜。Flink跟Spark Streaming很像，不同的是Flink把所有任務(wù)當(dāng)成流來處理，在迭代計算、內(nèi)存管理方面比Spark Streaming稍強(qiáng)，缺點是社區(qū)活躍度不高，還不夠成熟；PipelineDB是一個流式計算數(shù)據(jù)庫，能執(zhí)行簡單的流式計算任務(wù)，優(yōu)勢是基本不需要開發(fā)，只要熟悉SQL操作均可以輕松使用，但對于集群計算，需要商業(yè)上的支持。
4 DPI數(shù)據(jù)處理方案
基于實際任務(wù)需求以及上文流式框架的對比，由于Kafka Stream編程難度小，不需要另外安裝軟件，與Kafka等組件無縫連接，比較穩(wěn)定，并且各種性能均比較優(yōu)秀，因此本文選擇了Kafka Stream作為流式處理的核心組件。
4.1 寬帶DPI處理
為了完成寬帶DPI數(shù)據(jù)的實時抓包、資料填補(bǔ)、清洗、轉(zhuǎn)換及并入庫等工作，應(yīng)用了上述DPI數(shù)據(jù)處理方案。具體項目方案如圖5所示：
圖5 廣州寬帶DPI處理方案
Mina進(jìn)程是一個JAVA程序，基于mina框架開發(fā)，主要接收AAA數(shù)據(jù)包，獲得用戶賬戶信息，解析計算，并持久化到redis，最后發(fā)送給抓包（Capture）程序。Capture程序由C語言編寫，使用開源pcap抓取網(wǎng)卡http包，解析，結(jié)合用戶帳號資料，把DPI寫入到Kafka中。Kafka stream完成DPI的實時清洗和轉(zhuǎn)換工作。
Flume[15]是Cloudera開源的分布式可靠、可用、高效的收集，聚合和移動不同數(shù)據(jù)源的海量數(shù)據(jù)系統(tǒng)，配置簡單，基本無需開發(fā)，資源消耗低，支持傳輸數(shù)據(jù)到HDFS，非常適合與大數(shù)據(jù)系統(tǒng)結(jié)合。本項目將流式處理完后的數(shù)據(jù)通過Flume從Kafka中寫入到HDFS，建立hive表，為上層應(yīng)用提供數(shù)據(jù)。
Kafka Stream采用自主研發(fā)的ETL框架[16]，負(fù)責(zé)數(shù)據(jù)過濾（圖片、視頻等去掉），數(shù)據(jù)處理（獲取網(wǎng)絡(luò)ID、字段解析等）。ETL框架采用JAVA語言開發(fā)，支持多種數(shù)據(jù)源，包括普通文本、壓縮格式及xml立體格式等。支持多種大數(shù)據(jù)計算框架，包括Map/Reduce、Spark streaming、Kafka Stream和Flume等；具有擴(kuò)展方便、字段校驗、支持字段的通配符及支持維表查詢等功能。在運維方面，支持變量引用以及出錯處理等功能。
4.2 4G DPI實時統(tǒng)計
以電信4G DPI信息作為數(shù)據(jù)源，通過流式處理，完成DPI的實時統(tǒng)計工作，包括多粒度（5分鐘/1小時/1天）去重用戶統(tǒng)計、多粒度去重不同號碼頭用戶統(tǒng)計、多粒度流量統(tǒng)計及多粒度去重域名統(tǒng)計等。4G DPI實時統(tǒng)計具體項目方案如圖6所示：
.png)
圖6 4G DPI實時統(tǒng)計方案圖
數(shù)據(jù)源是gzip壓縮文件，因為flume原生不支持.gz或.tar.gz文件格式，所以修改了Flume底層代碼，實現(xiàn)對壓縮文件的處理，省去了解壓時間。Flume采集文件時以用戶手機(jī)號碼作為分區(qū)的key，將同一號碼的數(shù)據(jù)分到同一分區(qū)，便于去重。通過Kafka集群管理工具，Kafka Manager[17]可以很好地監(jiān)測Kafka集群的狀態(tài)。Kafka集群生產(chǎn)者如圖7所示：
圖7 Kafka集群生產(chǎn)者
Kafka Stream消費4GDPI的數(shù)據(jù)，并行處理。在程序里設(shè)置不同的計數(shù)器，所有數(shù)據(jù)都經(jīng)過這些計數(shù)器處理，為了解決去重問題，引入了布隆過濾，雖然有一定的誤判率，但是還是能比較好的完成去重，同時保證系統(tǒng)的性能。同樣消費者也可以通過Kafka Manager進(jìn)行管理，可以直觀觀察到消費者的落后程度。
為了滿足不同的輸出要求，程序設(shè)置了三種輸出供選擇。粒度為天的數(shù)據(jù)將會寫到MySQL作為備份，針對熱點區(qū)域的監(jiān)控數(shù)據(jù)將會輸出到Redis，同時，為了方便管理以及數(shù)據(jù)呈現(xiàn)，還采用了ELK框架（ElasticSearch+Logstash+Kibana），將所有數(shù)據(jù)傳到Kibana做前端展示。Kibana界面如圖8所示：
圖8 Kibana界面
5 實踐及分析
5.1 部署實踐
上述兩個系統(tǒng)均已應(yīng)用在實際的生產(chǎn)中，均有不錯的表現(xiàn)，能夠滿足任務(wù)需求，并且已經(jīng)穩(wěn)定運行。
寬帶DPI處理項目有2臺采集機(jī)、1臺AAA服務(wù)器及5臺Kafka機(jī)器。采集機(jī)每臺每秒產(chǎn)生115 MB數(shù)據(jù)，兩臺1.8 G流量。采集機(jī)寫Kafka 33萬條/秒，Kafka Stream寫Kafka 22萬條/秒，清洗率（清洗工作把諸如圖片、視頻及js請求等與業(yè)務(wù)無關(guān)的DPI信息去掉）為33%。Kafka Stream落后處理穩(wěn)定在500萬數(shù)據(jù)，延遲處理在15 s之內(nèi)，F(xiàn)lume寫HDFS落后保持在100萬左右，5 s內(nèi)的延遲。寬帶DPI處理項目性能如圖9所示：
圖9 寬帶DPI處理項目性能
4G DPI實時統(tǒng)計項目共6臺機(jī)器，1臺為Flume采集機(jī)，其余5臺部署Kafka、Kafka Stream及ELK。采集機(jī)寫Kafka一般為10萬條/秒，峰值可達(dá)到25萬條/秒。ElasticSearch集群一共8個實例，每個實例配置2 G內(nèi)存。目前集群有13億條數(shù)據(jù)，占361 G空間。通過Logstash導(dǎo)入數(shù)據(jù)到ElasticSearch峰值可以達(dá)到8~9萬條/秒。Kafka Stream處理數(shù)據(jù)落后在10 s內(nèi)，Logstash寫ElasticSearch落后在5 s內(nèi)，如圖10所示。目前4G DPI實時統(tǒng)計項目日均處理文件超過15 000個，大小達(dá)到1.6 T，日均處理記錄數(shù)超過100億。
圖10 4G DPI實時統(tǒng)計項目性能
5.2 存在的問題
在4G DPI實時統(tǒng)計項目開發(fā)過程中，隨著項目的需求越來越多，后面增加了對域名和CGI的去重，而且同一域名或者CGI不在同一Kafka分區(qū)，導(dǎo)致結(jié)果有偏差。為了解決這一問題，程序設(shè)計了二次去重，第一次去重的結(jié)果把CGI或者域名作為key輸出到Kafka集群，再做了一次去重工作，導(dǎo)致延遲時間變大和系統(tǒng)維護(hù)變復(fù)雜。
由于寬帶DPI處理中不涉及去重，只是數(shù)據(jù)過濾和數(shù)據(jù)轉(zhuǎn)換，因此Kafka Stream是非常適合的。但在涉及分區(qū)和去重的4G DPI實時統(tǒng)計項目中，應(yīng)當(dāng)采用Storm作為流式處理框架。在Storm中，數(shù)據(jù)從一個bolt流到另外一個bolt，這樣數(shù)據(jù)可以在一個bolt中按手機(jī)號碼分區(qū)，在另外一個bolt中又可以按CGI或者域名分區(qū)，可以避免二次去重問題，降低編程模型復(fù)雜度。
在程序設(shè)計之初，應(yīng)根據(jù)應(yīng)用場景需求選擇合適的技術(shù)框架。如果項目基礎(chǔ)結(jié)構(gòu)中涉及Spark，那Spark Streaming是不錯的選擇；如果像4G DPI實時統(tǒng)計項目一樣需要數(shù)據(jù)轉(zhuǎn)移或者去重，那么Storm是首選；如果是簡單的數(shù)據(jù)清洗和轉(zhuǎn)換處理，那么Kafka Stream是不錯的選擇。對于簡單小規(guī)模的實時統(tǒng)計，PipeLineDB足以勝任。
6 結(jié)束語
大數(shù)據(jù)流式計算和批處理適用于不同的業(yè)務(wù)場景，在對時效要求高的場景下，流式計算具有明顯的優(yōu)勢。本文首先概述了流式處理以及其與批處理的區(qū)別，然后對業(yè)界流行的流式計算框架進(jìn)行了對比，根據(jù)業(yè)務(wù)需求提出了以Kafka Stream為流式處理框架的DPI數(shù)據(jù)處理方案，搭配Kafka、Flume及ELK等組件，具有入門迅速、編程難度低和部署維護(hù)簡單等特點。并且將方案應(yīng)用到了寬帶DPI處理項目以及4G DPI實時統(tǒng)計項目中，完成了任務(wù)需求，性能優(yōu)異，運行穩(wěn)定。
在對實際項目實踐中，隨著任務(wù)需求的增多，發(fā)現(xiàn)Kafka Stream在應(yīng)對多維度數(shù)據(jù)去重問題時表現(xiàn)不力，需要引入二次過濾來解決問題。因此在項目需求階段，便要在技術(shù)框架選型時充分考慮可能出現(xiàn)的問題，結(jié)合技術(shù)框架適用場景，綜合考慮。
[1] Zikopoulos P, Eaton C. Understanding Big Data: Analytics for Enterprise Class Hadoop and Streaming Data[M]. McGraw-Hill Osborne Media, 1989.
[2] 陳康,付華崢,陳翀,等. 基于DPI的用戶興趣實時分類[J]. 電信科學(xué), 2016,32(12): 109-115.
[3] 孫大為,張廣艷,鄭緯民. 大數(shù)據(jù)流式計算:關(guān)鍵技術(shù)及系統(tǒng)實例[J]. 軟件學(xué)報, 2014,25(4): 839-862.
[4] 董斌,楊迪,王錚,等. 流計算大數(shù)據(jù)技術(shù)在運營商實時信令處理中的應(yīng)用[J]. 電信科學(xué), 2015,31(10): 165-171.
[5] Marz N,Warren J. Big Data: Principles and best practices of scalable realtime data systems[M]. Manning, 2015.
[6] 李祥池. 基于ELK和Spark Streaming的日志分析系統(tǒng)設(shè)計與實現(xiàn)[J]. 電子科學(xué)技術(shù), 2015,2(6): 674-678.
[7] 李圣,黃永忠,陳海勇. 大數(shù)據(jù)流式計算系統(tǒng)研究綜述[J]. 信息工程大學(xué)學(xué)報, 2016,17(1): 88-92.
[8] 姚仁捷. Kafka在唯品會的應(yīng)用實踐[J]. 程序員, 2014(1): 110-113.
[9] 郝璇. 基于Apache Flume的分布式日志收集系統(tǒng)設(shè)計與實現(xiàn)[J]. 軟件導(dǎo)刊, 2014(7): 110-111.
[10] Spark. Spark Streaming Programming Guide[EB/OL]. [2017-09-14]. http://spark.apache.org/docs/latest/streaming-programming-guide.html.
[11] Storm. Apache Storm[EB/OL]. [2017-09-14]. http://storm.apache.org/index.html.
[12] Kafka Stream. Kafka Streams API[EB/OL]. [2017-09-14]. http://kafka.apache.org/documentation/streams/.
[13] Flink. Introduction to Apache Flink®[EB/OL]. [2017-09-14]. https://flink.apache.org/introduction.html.
[14] PipelineDB. The Streaming SQL Database[EB/OL]. [2017-09-14]. https://www.pipelinedb.com/.
[15] Apache Flume™. Apache Flume™[EB/OL]. [2017-09-14]. http://flume.apache.org/index.html.
[16] Kafka Stream. ETL[EB/OL]. [2017-09-14]. https://github.com/styg/bumblebee-ETL.
[17] Kafka Stream. Kafka Manager[EB/OL]. [2017-09-14]. https://github.com/yahoo/kafka-manager. ★
作者簡介
田熙清：碩士畢業(yè)于大連理工大學(xué)系統(tǒng)工程專業(yè)，現(xiàn)任職于中國電信股份有限公司廣州研究院，研究方向為大數(shù)據(jù)平臺及處理。
范家杰：碩士畢業(yè)于中山大學(xué)，現(xiàn)任職于中國電信股份有限公司廣州研究院，研究方向為大數(shù)據(jù)平臺及處理。 
