基于流式計(jì)算的DPI數(shù)據(jù)處理方案

1 引言

隨著移動(dòng)互聯(lián)網(wǎng)的不斷發(fā)展以及各類智能設(shè)備日益深入民眾日常生活中，人類社會(huì)產(chǎn)生的數(shù)據(jù)量正在以指數(shù)級快速增長，人類已經(jīng)正式邁入大數(shù)據(jù)時(shí)代[1]。如今，運(yùn)營商能夠獲得的用戶數(shù)據(jù)越來越豐富，通過DPI（Deep Packet Inspector，深度分組檢測）分析技術(shù)，能夠較好地識(shí)別網(wǎng)絡(luò)上的流量類別、應(yīng)用層上的應(yīng)用種類等[2]。在這個(gè)“數(shù)據(jù)為王”的時(shí)代，如何充分利用這筆重要的戰(zhàn)略資產(chǎn)已成為重中之重。

數(shù)據(jù)規(guī)模的快速增長給大數(shù)據(jù)分析處理帶來了巨大的挑戰(zhàn)，尤其是在通信行業(yè)，數(shù)據(jù)越發(fā)呈現(xiàn)出無限性、突發(fā)性和實(shí)時(shí)性等特征[3]，傳統(tǒng)的基于MapReduce的批處理模式難以滿足數(shù)據(jù)實(shí)時(shí)性的要求，而能否在第一時(shí)間獲得數(shù)據(jù)所蘊(yùn)含的信息決定了數(shù)據(jù)的價(jià)值。因此，流式處理技術(shù)成為大數(shù)據(jù)技術(shù)研究的新熱點(diǎn)[4]。流式處理能夠針對數(shù)據(jù)的變化進(jìn)行實(shí)時(shí)處理，能夠在秒級獲得處理結(jié)果，特別適合一些對時(shí)效性要求很高的場景。

本文結(jié)合電信運(yùn)營商的需求，對DPI數(shù)據(jù)進(jìn)行實(shí)時(shí)的采集及處理，提出一種基于流式計(jì)算的DPI數(shù)據(jù)處理方案，能夠?qū)@得DPI數(shù)據(jù)實(shí)時(shí)信息的時(shí)延降低到分鐘級，甚至秒級，實(shí)現(xiàn)對電信用戶上網(wǎng)信息的實(shí)時(shí)處理、監(jiān)測及分類匯總，為之后進(jìn)行的大數(shù)據(jù)應(yīng)用提供了良好基礎(chǔ)。

2 流式處理概述

傳統(tǒng)基于MapReduce大數(shù)據(jù)處理技術(shù)實(shí)際上是一種批處理方式，如圖1所示。批處理模式首先要完成數(shù)據(jù)的累積和存儲(chǔ)，然后Hadoop客戶端將數(shù)據(jù)上傳到HDFS上，最后才啟動(dòng)Map/Reduce進(jìn)行數(shù)據(jù)處理，處理后再寫入到HDFS。這種方式必須要所有數(shù)據(jù)都要準(zhǔn)備好，然后統(tǒng)一進(jìn)行集中計(jì)算和價(jià)值發(fā)現(xiàn)，無法滿足實(shí)時(shí)性的要求。

2015年，Nathan Marz提出了實(shí)時(shí)大數(shù)據(jù)處理框架Lambda架構(gòu)[5]，整合了離線計(jì)算和實(shí)時(shí)計(jì)算，能夠滿足實(shí)時(shí)系統(tǒng)高容錯(cuò)、低時(shí)延和可擴(kuò)展等要求，并且可集成Hadoop、Kafka、Storm、Spark及HBase等各類大數(shù)據(jù)組件。

一個(gè)典型的Lambda架構(gòu)如圖2所示，主要使用的場景是邏輯復(fù)雜且延遲低的程序。數(shù)據(jù)會(huì)分別灌入實(shí)時(shí)系統(tǒng)和批處理系統(tǒng)，然后各自輸出自己的結(jié)果，結(jié)果會(huì)在查詢端進(jìn)行合并。

圖2 Lambda架構(gòu)圖

3 流式計(jì)算架構(gòu)對比

流式計(jì)算對系統(tǒng)的容錯(cuò)、時(shí)延、可擴(kuò)展及可靠性能力提出了很高的要求，當(dāng)前有許多流式計(jì)算框架（如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í)驗(yàn)室專門為大數(shù)據(jù)處理而設(shè)計(jì)的計(jì)算框架[6]。Spark Streaming是建立在Spark上的實(shí)時(shí)計(jì)算框架，是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è)計(jì)的好處讓Spark Streaming能夠同時(shí)處理離線處理和流處理問題。

圖3 Spark Streaming執(zhí)行流程

Spark Streaming能在故障報(bào)錯(cuò)下迅速恢復(fù)狀態(tài)，整合了批處理與流處理，內(nèi)置豐富高級算法處理庫，發(fā)展迅速，社區(qū)活躍。毫無疑問，Spark Streaming是流式處理框架的佼佼者。缺點(diǎn)是由于需要累積一批小文件才處理，因此時(shí)延會(huì)稍大，是準(zhǔn)實(shí)時(shí)系統(tǒng)。

3.2 Storm

Storm通常被比作“實(shí)時(shí)的Hadoop”，是Twitter開發(fā)的實(shí)時(shí)、分布式以及具備高容錯(cuò)計(jì)算系統(tǒng)，可以簡單、可靠地處理大量數(shù)據(jù)流，用戶可以采用任意編程語言來開發(fā)應(yīng)用。

在Storm中，一個(gè)用于實(shí)時(shí)計(jì)算的圖狀結(jié)構(gòu)稱之為拓?fù)洌╰opology），拓?fù)涮峤坏郊?，由集群中的主控?jié)點(diǎn)分發(fā)代碼，分配任務(wù)到工作節(jié)點(diǎn)執(zhí)行。一個(gè)拓?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ù)流動(dòng)

Storm能將數(shù)據(jù)在不同的bolt中流動(dòng)、移動(dòng)數(shù)據(jù)，真正實(shí)現(xiàn)流式處理，易于擴(kuò)展，靈活性強(qiáng)，高度專注于流式處理。Storm在事件處理與增量計(jì)算方面表現(xiàn)突出，能夠以實(shí)時(shí)方式根據(jù)不斷變化的參數(shù)對數(shù)據(jù)流進(jìn)行處理。

3.3 Kafka Stream

Kafka Stream是Apache Kafka開源項(xiàng)目的一個(gè)組成部分，是一個(gè)功能強(qiáng)大、易于使用的庫，它使得Apache Kafka擁有流處理的能力。

Kafka Stream是輕量級的流計(jì)算類庫，除了Apache Kafka之外沒有任何外部依賴，可以在任何Java程序中使用，使用Kafka作為內(nèi)部消息通訊存儲(chǔ)介質(zhì)，因此不需要為流處理需求額外部署一個(gè)集群。

Kafka Stream入門簡單，并且不依賴其他組件，非常容易部署，支持容錯(cuò)的本地狀態(tài)，延遲低，非常適合一些輕量級流處理的場景。

3.4 Flink

Flink是一個(gè)面向分布式數(shù)據(jù)流處理和批量數(shù)據(jù)處理的開源計(jì)算平臺(tái)，同時(shí)支持批處理以及流處理，主要針對流數(shù)據(jù)，將批數(shù)據(jù)視為流數(shù)據(jù)的一個(gè)極限特例。

Flink核心是一個(gè)流式的數(shù)據(jù)流執(zhí)行引擎，它提供了數(shù)據(jù)分布、數(shù)據(jù)通信以及容錯(cuò)機(jī)制等功能。流執(zhí)行引擎之上，F(xiàn)link提供了更高層次的API以便用戶使用。Flink還針對某些領(lǐng)域提供了領(lǐng)域庫，例如Flink ML、Flink的機(jī)器學(xué)習(xí)庫等。

Flink適合有極高流處理需求，并有少量批處理任務(wù)的場景。該技術(shù)可兼容原生Storm和Hadoop程序，可在YARN管理的集群上運(yùn)行。目前Flink最大的局限之一是在社區(qū)活躍度方面，該項(xiàng)目的大規(guī)模部署尚不如其他處理框架那么常見。

3.5 PipeLineDB

PipelineDB是基于PostgreSQL的一個(gè)流式計(jì)算數(shù)據(jù)庫，效率非常高，通過SQL對數(shù)據(jù)流做操作，并把操作結(jié)果儲(chǔ)存起來。其基本過程是：創(chuàng)建PipelineDB Stream、編寫SQL、對Stream做操作、操作結(jié)果被保存到continuous view。

PipelineDB特點(diǎn)是可以只使用SQL進(jìn)行流式處理，不需要代碼，可以高效可持續(xù)自動(dòng)處理流式數(shù)據(jù)，只存儲(chǔ)處理后的數(shù)據(jù)，因此非常適合流式數(shù)據(jù)處理，例如網(wǎng)站流量統(tǒng)計(jì)、網(wǎng)頁的瀏覽統(tǒng)計(jì)等。

3.6 架構(gòu)對比

上文提到的5種流式處理框架對比如表1所示：

表1 流式框架對比

Storm的特點(diǎn)是成熟，是流式處理框架實(shí)際上的標(biāo)準(zhǔn)，模型、編程難度都比較復(fù)雜，框架采用循環(huán)處理數(shù)據(jù)，對系統(tǒng)資源，尤其是CPU資源消耗很大，當(dāng)任務(wù)空閑時(shí)，需要sleep程序，減少對資源的消耗。Spark Streaming兼顧了批處理以及流式處理，并且有Spark的強(qiáng)大支持，發(fā)展?jié)摿Υ?，但與Kafka的接口平滑性不夠。Kafka Stream是Kafka的一個(gè)開發(fā)庫，具有入門、編程、部署運(yùn)維簡單的特點(diǎn)，并且不需要部署額外的組件，但對于多維度的統(tǒng)計(jì)來說，需要基于不同topic來做分區(qū)，編程模型復(fù)雜。Flink跟Spark Streaming很像，不同的是Flink把所有任務(wù)當(dāng)成流來處理，在迭代計(jì)算、內(nèi)存管理方面比Spark Streaming稍強(qiáng)，缺點(diǎn)是社區(qū)活躍度不高，還不夠成熟；PipelineDB是一個(gè)流式計(jì)算數(shù)據(jù)庫，能執(zhí)行簡單的流式計(jì)算任務(wù)，優(yōu)勢是基本不需要開發(fā)，只要熟悉SQL操作均可以輕松使用，但對于集群計(jì)算，需要商業(yè)上的支持。

4 DPI數(shù)據(jù)處理方案

基于實(shí)際任務(wù)需求以及上文流式框架的對比，由于Kafka Stream編程難度小，不需要另外安裝軟件，與Kafka等組件無縫連接，比較穩(wěn)定，并且各種性能均比較優(yōu)秀，因此本文選擇了Kafka Stream作為流式處理的核心組件。

4.1寬帶DPI處理

為了完成寬帶DPI數(shù)據(jù)的實(shí)時(shí)抓包、資料填補(bǔ)、清洗、轉(zhuǎn)換及并入庫等工作，應(yīng)用了上述DPI數(shù)據(jù)處理方案。具體項(xiàng)目方案如圖5所示：

圖5 廣州寬帶DPI處理方案

Mina進(jìn)程是一個(gè)JAVA程序，基于mina框架開發(fā)，主要接收AAA數(shù)據(jù)包，獲得用戶賬戶信息，解析計(jì)算，并持久化到redis，最后發(fā)送給抓包（Capture）程序。Capture程序由C語言編寫，使用開源pcap抓取網(wǎng)卡http包，解析，結(jié)合用戶帳號資料，把DPI寫入到Kafka中。Kafka stream完成DPI的實(shí)時(shí)清洗和轉(zhuǎn)換工作。

Flume[15]是Cloudera開源的分布式可靠、可用、高效的收集，聚合和移動(dòng)不同數(shù)據(jù)源的海量數(shù)據(jù)系統(tǒng)，配置簡單，基本無需開發(fā)，資源消耗低，支持傳輸數(shù)據(jù)到HDFS，非常適合與大數(shù)據(jù)系統(tǒng)結(jié)合。本項(xiàng)目將流式處理完后的數(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ù)計(jì)算框架，包括Map/Reduce、Spark streaming、Kafka Stream和Flume等；具有擴(kuò)展方便、字段校驗(yàn)、支持字段的通配符及支持維表查詢等功能。在運(yùn)維方面，支持變量引用以及出錯(cuò)處理等功能。

4.24GDPI實(shí)時(shí)統(tǒng)計(jì)

以電信4G DPI信息作為數(shù)據(jù)源，通過流式處理，完成DPI的實(shí)時(shí)統(tǒng)計(jì)工作，包括多粒度（5分鐘/1小時(shí)/1天）去重用戶統(tǒng)計(jì)、多粒度去重不同號碼頭用戶統(tǒng)計(jì)、多粒度流量統(tǒng)計(jì)及多粒度去重域名統(tǒng)計(jì)等。4G DPI實(shí)時(shí)統(tǒng)計(jì)具體項(xiàng)目方案如圖6所示：

圖6 4G DPI實(shí)時(shí)統(tǒng)計(jì)方案圖

數(shù)據(jù)源是gzip壓縮文件，因?yàn)閒lume原生不支持.gz或.tar.gz文件格式，所以修改了Flume底層代碼，實(shí)現(xiàn)對壓縮文件的處理，省去了解壓時(shí)間。Flume采集文件時(shí)以用戶手機(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消費(fèi)4GDPI的數(shù)據(jù)，并行處理。在程序里設(shè)置不同的計(jì)數(shù)器，所有數(shù)據(jù)都經(jīng)過這些計(jì)數(shù)器處理，為了解決去重問題，引入了布隆過濾，雖然有一定的誤判率，但是還是能比較好的完成去重，同時(shí)保證系統(tǒng)的性能。同樣消費(fèi)者也可以通過Kafka Manager進(jìn)行管理，可以直觀觀察到消費(fèi)者的落后程度。

為了滿足不同的輸出要求，程序設(shè)置了三種輸出供選擇。粒度為天的數(shù)據(jù)將會(huì)寫到MySQL作為備份，針對熱點(diǎn)區(qū)域的監(jiān)控?cái)?shù)據(jù)將會(huì)輸出到Redis，同時(shí)，為了方便管理以及數(shù)據(jù)呈現(xiàn)，還采用了ELK框架（ElasticSearch+Logstash+Kibana），將所有數(shù)據(jù)傳到Kibana做前端展示。Kibana界面如圖8所示：

圖8 Kibana界面

5 實(shí)踐及分析

5.1 部署實(shí)踐

上述兩個(gè)系統(tǒng)均已應(yīng)用在實(shí)際的生產(chǎn)中，均有不錯(cuò)的表現(xiàn)，能夠滿足任務(wù)需求，并且已經(jīng)穩(wěn)定運(yùn)行。

寬帶DPI處理項(xiàng)目有2臺(tái)采集機(jī)、1臺(tái)AAA服務(wù)器及5臺(tái)Kafka機(jī)器。采集機(jī)每臺(tái)每秒產(chǎn)生115 MB數(shù)據(jù)，兩臺(tái)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處理項(xiàng)目性能如圖9所示：

圖9 寬帶DPI處理項(xiàng)目性能

4G DPI實(shí)時(shí)統(tǒng)計(jì)項(xiàng)目共6臺(tái)機(jī)器，1臺(tái)為Flume采集機(jī)，其余5臺(tái)部署Kafka、Kafka Stream及ELK。采集機(jī)寫Kafka一般為10萬條/秒，峰值可達(dá)到25萬條/秒。ElasticSearch集群一共8個(gè)實(shí)例，每個(gè)實(shí)例配置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實(shí)時(shí)統(tǒng)計(jì)項(xiàng)目日均處理文件超過15 000個(gè)，大小達(dá)到1.6 T，日均處理記錄數(shù)超過100億。

圖10 4G DPI實(shí)時(shí)統(tǒng)計(jì)項(xiàng)目性能

5.2 存在的問題

在4G DPI實(shí)時(shí)統(tǒng)計(jì)項(xiàng)目開發(fā)過程中，隨著項(xiàng)目的需求越來越多，后面增加了對域名和CGI的去重，而且同一域名或者CGI不在同一Kafka分區(qū)，導(dǎo)致結(jié)果有偏差。為了解決這一問題，程序設(shè)計(jì)了二次去重，第一次去重的結(jié)果把CGI或者域名作為key輸出到Kafka集群，再做了一次去重工作，導(dǎo)致延遲時(shí)間變大和系統(tǒng)維護(hù)變復(fù)雜。

由于寬帶DPI處理中不涉及去重，只是數(shù)據(jù)過濾和數(shù)據(jù)轉(zhuǎn)換，因此Kafka Stream是非常適合的。但在涉及分區(qū)和去重的4G DPI實(shí)時(shí)統(tǒng)計(jì)項(xiàng)目中，應(yīng)當(dāng)采用Storm作為流式處理框架。在Storm中，數(shù)據(jù)從一個(gè)bolt流到另外一個(gè)bolt，這樣數(shù)據(jù)可以在一個(gè)bolt中按手機(jī)號碼分區(qū)，在另外一個(gè)bolt中又可以按CGI或者域名分區(qū)，可以避免二次去重問題，降低編程模型復(fù)雜度。

在程序設(shè)計(jì)之初，應(yīng)根據(jù)應(yīng)用場景需求選擇合適的技術(shù)框架。如果項(xiàng)目基礎(chǔ)結(jié)構(gòu)中涉及Spark，那Spark Streaming是不錯(cuò)的選擇；如果像4G DPI實(shí)時(shí)統(tǒng)計(jì)項(xiàng)目一樣需要數(shù)據(jù)轉(zhuǎn)移或者去重，那么Storm是首選；如果是簡單的數(shù)據(jù)清洗和轉(zhuǎn)換處理，那么Kafka Stream是不錯(cuò)的選擇。對于簡單小規(guī)模的實(shí)時(shí)統(tǒng)計(jì)，PipeLineDB足以勝任。

6 結(jié)束語

大數(shù)據(jù)流式計(jì)算和批處理適用于不同的業(yè)務(wù)場景，在對時(shí)效要求高的場景下，流式計(jì)算具有明顯的優(yōu)勢。本文首先概述了流式處理以及其與批處理的區(qū)別，然后對業(yè)界流行的流式計(jì)算框架進(jìn)行了對比，根據(jù)業(yè)務(wù)需求提出了以Kafka Stream為流式處理框架的DPI數(shù)據(jù)處理方案，搭配Kafka、Flume及ELK等組件，具有入門迅速、編程難度低和部署維護(hù)簡單等特點(diǎn)。并且將方案應(yīng)用到了寬帶DPI處理項(xiàng)目以及4G DPI實(shí)時(shí)統(tǒng)計(jì)項(xiàng)目中，完成了任務(wù)需求，性能優(yōu)異，運(yùn)行穩(wěn)定。

在對實(shí)際項(xiàng)目實(shí)踐中，隨著任務(wù)需求的增多，發(fā)現(xiàn)Kafka Stream在應(yīng)對多維度數(shù)據(jù)去重問題時(shí)表現(xiàn)不力，需要引入二次過濾來解決問題。因此在項(xiàng)目需求階段，便要在技術(shù)框架選型時(shí)充分考慮可能出現(xiàn)的問題，結(jié)合技術(shù)框架適用場景，綜合考慮。