基于Spark Streaming的實時數據分析系統及其應用

韓德志，陳旭光，雷雨馨，戴永濤，張 肖

(1.上海海事大學 信息工程學院，上海 201306； 2.鄭州大學 信息工程學院，鄭州 450001)

基于Spark Streaming的實時數據分析系統及其應用

韓德志1*，陳旭光1，雷雨馨2，戴永濤1，張 肖1

(1.上海海事大學 信息工程學院，上海 201306； 2.鄭州大學 信息工程學院，鄭州 450001)

(*通信作者電子郵箱dezhihan88@sina.com)

為了實現對實時網絡數據流的快速分析，設計一種分布式實時數據流分析系統(DRDAS)，能有效解決并發訪問數據流的收集、存儲和實時分析問題，為大數據環境的網絡安全檢測提供了一種有效的數據分析平臺；根據Spark Streaming運行的原理設計一種動態采樣的K-Means并行算法，與DRDAS結合能實時有效地檢測大數據環境下的各種分布式拒絕服務(DDoS)攻擊。實驗結果顯示：DRDAS具有好的可擴展性、容錯性和實時處理能力，與動態采樣的K-Means并行算法結合能實時地檢測各種DDoS攻擊，縮短了攻擊的檢測時間。

Spark Streaming框架; 分布式流處理；網絡數據分析；分布式拒絕服務攻擊

0 引言

隨著互聯網的高速發展，普通用戶帶寬普遍升級，尤其是在一些大城市，家庭用戶的帶寬都已經達到20 MB/s，甚至更高。此外，隨著3G網絡的普及，以及4G網絡的逐步推廣，移動互聯網也進入了一個蓬勃發展的時期。個人網絡帶寬的快速增長和不斷增加的網絡用戶，對企業網絡安全帶來了巨大的挑戰。因為，這些高帶寬的網絡用戶一旦被黑客控制并參與到分布式拒絕服務(Distributed Denial of Service, DDoS)攻擊中時，其影響將無法估量。

全球著名內容分發網絡(Content Delivery Network, CDN)服務商Incapsula，在2014年發布的DDoS攻擊趨勢報告[1]中指出,2014年DDoS攻擊行為增加了240%，且流量已經超過了100 GB。此外，該公司的最近一份分析報告[2]指出，目前或有數十萬甚至上百萬臺的小型或家庭辦公環境 (Small Office Home Office, SOHO)專用路由器已成為僵尸路由器，被黑客用來執行大規模DDoS攻擊。該公司2014年的一項關于DDoS造成損失的調查發現：49%的DDoS攻擊會持續6～24 h，每小時的平均經濟損失為40 000美元[3]。

此外，近兩年來，網絡安全事件更是頻繁發生：2014年12月20日— 21日，一家將服務器部署在阿里云上的游戲公司遭遇了DDoS攻擊，453.8 GB/s的峰值流量使之成為全球最大的DDoS攻擊受害者；而知名的代碼托管網站GitHub，于2015年3月26日開始遭遇其網站最大規模的DDoS攻擊，導致部分地區服務中斷，且攻擊持續了80多小時；2015年5月11日，網易遭受了最新型的DDoS攻擊——鏈路洪泛攻擊(Link Flooding Attack, LFA)[3]，以至服務中斷9 h，損失超過1 500萬元。

對超大流量的DDoS攻擊，研究如何高效、及時地檢測并報警有重要的理論意義和重大的經濟價值。本文將新型的分布式流式計算框架——Spark Streaming[4]，應用于實時的大流量網絡數據實時分析，為提高大數據環境的DDoS檢測速度提供保證。

1 相關工作

1.1 大數據流特點

在大數據時代，流式數據已經在很多環境中加以應用，并取得很好的效果。本文在文獻[5]的基礎上，總結出了大數據流具有以下6個特點：

1)并發性。體現在兩個方面：服務并發和數據并發。在當前的大部分系統中，流處理都是在一些高并發的應用中，所提供的都是高并發服務，而并發服務必然會并發產生大量數據，這就要求系統具有很好的并發收集、處理和分析的能力。

2)實時性。數據實時產生，也需要實時計算，保證數據的時效性，讓數據更快、更好地體現其價值，因為很多場景，比如實時推薦系統、實時監控系統，數據是具有實效性的，若不能在短時間內加以處理和分析，就會失去數據的價值。

3)易失性。原始數據流一般被實時處理和分析后，就會被丟棄，只有少量日志被持久保存，或者只保存處理后的結果，想要再次分析原始日志可能性不大。

4)突發性。因為數據是實時產生的，只受當前服務的影響，跟前一時刻的數據量大小無關，也就是說前后兩個時間的數據量可能會相差很大。

5)無序性。在大數據環境下，不同數據源之間的產生和傳輸并不能保證按照其產生的相對順序，因此在分析時需要更加完善的處理邏輯，而不是只根據其到達的先后順序。

6)無限性。只要服務存在，數據就一直會持續、動態地增加，所以數據的大小理論上是無限的，不能用具體的數字來量化。當然，實際使用中不可能用磁盤無限地存儲，應定期地對歷史日志進行清理或壓縮備份。

1.2 大數據流相關研究

由于單臺計算機的存儲、計算能力的限制，已經不能滿足大數據時代海量數據的處理和分析的要求。同樣地，在網絡安全檢測中，單臺服務器已經不能及時處理和分析短時間內產生的大量網絡數據，這給安全檢測系統帶來了巨大的挑戰。目前，已經有一些學者開始研究用分布式計算框架處理海量的網絡數據。

文獻[6]使用Hadoop對海量數據流的日志進行分析，該方法能提供良好的容錯性，且能大幅縮短計算的時間，但是需要先將所有的日志收集起來，再導入到Hadoop集群中，這樣浪費了大量的時間，不能對大數據流進行實時分析。

文獻[7]中將Hadoop應用到海量網絡數據流攻擊的日志分析中，雖然能處理和分析大量的網絡監控數據的日志信息，但是由于其采用離線分析方式，該方法只適用于研究網絡攻擊的一些行為和特征，不能真正用于網絡檢測系統中。

文獻[8]在文獻[7]的方法上作了一些改進，將原來的分析文本日志改成了直接分析抓取到的二進制數據，并實現了P3(Parallel Packet Processor)系統，在一定程度上減輕了數據抓取服務器的負擔，但還是采用離線分析處理，缺乏實用性。

文獻[9]用Hadoop進行DDoS攻擊檢測，實現HTTP GET flooding攻擊檢測算法，由于使用的是Hadoop系統，仍然不能解決響應時間慢的問題。

文獻[6-9]雖將分布式計算方法用于網絡檢測和數據流分析中，但是都不能很好地解決時效性的問題。針對以上情況，本文提出了一種基于Kafka[10]和Spark Streaming[4]的實時流處理和分析方法，對海量的網絡數據進行實時分析，適合于大流量的實時網絡分析和異常檢測。

海量的數據是大數據出現的前提，而數據收集則是大數據的基石。日志數據收集在流數據收集中占有重要比重，許多公司的業務平臺每天都會分散地產生大量日志數據，收集并匯總這些業務日志數據，供離線和在線的分析系統使用。日志收集系統所需考慮的基本特征包括：高可靠性、高可用性和高可擴展性?！胺稚⑹占?，集中處理”是當前日志處理系統的一個主流思想。日志收集也是流式日志處理系統的前提和基礎，日志只有被實時收集、匯總后，才能進行后續的相關處理操作。下面針對當前流行的開源日志數據流收集系統，進行介紹和對比。

2 數據流實時數據分析系統總體設計

2.1 數據流實時數據分析系統整體架構

實時網絡數據分析系統(Distributed Real-time Data Analysis System, DRDAS)的整體架構如圖1所示，其組成部分包括數據收集系統、數據分析系統和數據存儲系統。在圖1中，Kafka Broker是數據收集系統,負責大數據流的實時搜集和傳遞；Spark Streaming是數據分析系統，負責對Kafka Broker傳遞的數據流進行實時分析，并將分析結果傳遞應用系統進行各種處理，同時將正常的數據傳遞存儲系統；Database表示結構化的數據存儲系統，Hbase表示非結構化的數據存儲系統。

圖1 系統整體架構Fig. 1 Overall architecture of DRDAS

2.2 數據流實時數據收集系統

大數據流具有如本文1.1節所描述的6大特性，在大數據流并發產生的情況下，一方面要保證數據的實時性，另一方面盡量減少數數據丟失，這些都給數據的收集帶來了巨大挑戰。

在一些廣告、搜索、推薦、安全系統中，為了滿足海量日志或事件的實時傳輸，及時分析、處理，并實時反饋的需求，很多人都在為解決這一難題而研究。LinkedIn公司開發了一套專用的分布式消息訂閱和發布系統——Kafka[10]。早在2012年LinkedIn公司就已經將Kafka應用在實際生產系統，并每天收集和處理上百億的消息，峰值時能處理每秒高達 172 000條消息[8]。LinkedIn將Kafka于2011年開源，并成為Apache的開源項目之一。

由于Kafka是一套分布式系統，其吞吐量可以隨著集群的擴展而線性增加。

圖2為Kafka的整體架構。 圖中給出了Kafka的三大構成部分：生產者(Producer)，即日志的來源；代理(Broker)，消息的中間管理者；消費者(Consumer)，消息的使用者。Producer負責將消息收集并推送(Push)到Broker，而Broker則負責接收Producer發送來的消息，并將消息本地持久化，Consumer則是消息的真正使用者，從Broker拉取(Pull)消息并進行處理。

整個Kafka架構的核心就是Broker，Broker的吞吐量直接關系到整個系統的可用性。而為了加強其處理能力，設計者從兩方面作了考慮：1)巧妙的消息管理策略；2)線性化的擴展方式。

圖2 Kafka架構Fig. 2 Kafka architecture

圖3展示了Kafka的消息管理策略，通過Topic名稱將消息分開管理，一個Topic由多個Partition組成，而每個Partition又是由多個Segment文件構成，每個Segment存儲指定數量的消息，存滿后會存放到一個新的Segment文件中。消息在同一個Partition中是有序的，而且能保證消息讀取時是有序(按照Partition中的存儲順序)。

圖3 Kafka消息管理策略Fig. 3 Kafka message management strategy

圖4為Kafka對同一個Topic的消息讀寫管理：消息流入時，根據該消息所在Topic的分區數和復制數，將不同的消息分別存儲到不同的Partition上，同時，也將同一條消息復制多份，保存在不同的主機上，達到數據冗余備份和提高讀取速度的目的；Consumer讀取消息時，對于屬于同一個Group的Consumer，在同一個Partition上，不能并發讀取，但是在不同的Partition上可以并發操作，也就是說同一個Topic的Partition數量越多，可允許的并發數也就越多。

圖4 Kafka對Partitions的讀寫操作Fig. 4 Kafka’ read and write operations in the Partitions

2.3 數據流實時數據分析系統

Spark[11]于2009年誕生于加州伯克利大學的APMLab實驗室，2010年正式開源，并于2013年成為Apache基金項目，在2014年成為Apache基金的頂級項目。為了解決Hadoop[12]計算時將中間結果存入磁盤導致計算速度緩慢的問題，Spark應運而生。

Spark的生態系統如圖5所示，該生態系統包括了批處理、流處理、機器學習、圖計算、數據分析等，相比Hadoop生態圈，顯得更加全面，是一個更適合大數據應用場景的分布式計算框架。

彈性分布式數據集(Resilient Distribute Datasets，RDD)[13]是Spark的核心，也是Spark實現故障恢復、數據依賴的關鍵。RDD模型以簡單的邏輯——Lineage很好地解決了數據之間的依賴，且保證很好的容錯性，并能將中間結果存入內存，盡量減少數據的磁盤讀寫，使得計算速度得到了很大的提升，尤其是在迭代式計算中，計算速度提升了一個數量級。

圖5 Spark生態圈Fig. 5 Spark ecosystem

不同于Hadoop中的MapReduce，Spark將相應的MapReduce很好地封裝在RDD中。在RDD上可執行兩類操作：轉換(Transformation)和動作(Action)。圖6展示了數據從輸入到得到最終結果的操作過程。數據在RDD中并不是以原始形態存在，而是以數據所在的具體位置的形式包含在RDD中，并在RDD中經過不同的轉換得到新的RDD，直到執行動作時才執行真正的計算，得到最終想要的結果。

圖6 RDD上的相關操作Fig. 6 Related operations in RDD

Spark Streaming[4]是Spark生態系統中用于實時計算的一個框架，其核心也是基于RDD的，它可以無縫地和Spark銜接，將歷史數據和實時數據完美地融合。Spark Streaming與Yahoo 推出的S4[14]以及Twitter推出的Storm[15]并列為三大主要數據流處理系統。大數據流處理系統的典型實例見文獻[16]，Spark Streaming、S4和Storm等分布式流處理系統的對比見文獻[17]。

圖7展示了Spark Streaming流處理的實質：將一小段時間內的數據合并，再作微型的批處理，而不是分別對每一條數據進行實時處理。這也是它與其他流處理系統的最大區別，因此它并不是真正意義上的實時處理，而是延時較低的微批處理，正好適合本文中的應用場景。此外，Spark Streaming還能通過滑動窗口將最近一段時間的數據進行合并，或直接與Spark結合將歷史數據和實時數據一起分析，為實時數據和歷史分析提供了一個統一的處理平臺。

圖7 Spark Streaming的處理過程Fig. 7 Processing procedures of Spark Streaming

2.4 日志數據流收集系統對比

日志數據流收集系統具備3個基本組件，分別是Agent(接收原始數據，并將數據發給Collector)、Collector(接收多個Agent發送過來的數據，匯總后將數據發往Store)和Store(中央存儲系統，將匯總后的數據進行持久化存儲)。表1綜合對比了Scribe、Flume[18]、Chukwa、LogStash四種日志數據流收集系統。

表1 4種日志收集系統對比Tab. 1 Comparison of four log collection systems

3 數據流實時數據分析系統性能分析

3.1 數據流分析系統整體數據流圖

圖8為數據流分析系統整體數據流圖，整個系統包括：數據收集、數據分析、數據存儲、模型(或算法)訓練、入侵檢測。系統的數據流向為：1) 數據的來源為不同的服務器，通過各種抓包軟件，如：TcpDump、NetFlow、Sniff等，對特定的網卡或端口進行數據包抓取，并通過Flume將不同服務器上的網絡數據匯總，將數據抓、分析和檢測分離，減輕應用服務器的負擔。Flume最初是由Cloudera的工程師設計用于合并日志數據的系統，后將其開源出來，并逐漸發展成為一款開源、高可靠、高擴展、易管理、支持客戶擴展的分布式數據采集系統，主要是用于日志數據的收集和聚合。2)數據匯聚之后，將所抓取的網絡數據作為Kafka Producer的消息源，并傳送到Kafka Broker，讓Broker對所有網絡數據進行有序的管理。3)Spark Streaming則實時從Kafka Broker中拉取數據，再將數據分散到不同的Spark Executor進行分析和統計。4) Spark將抓取的網絡數據處理后，一方面可以將結果傳給其他的應用，作進一步分析；另一方面可以將結果持久化，存儲在數據庫中，供后續分析使用。5)對于得到的實時數據，可以使之與之前得到的歷史數據進行合并，進而進行模型(或算法)訓練或者直接通過模型進行DDoS檢測，并得到檢測結果。

圖8 數據流分析系統整體數據流圖Fig. 8 Data flow of data analysis system

3.2 實時性分析

要做到實時數據分析，就要求系統具有實時的數據收集、處理功能。本系統中，使用了Flume和Kafka來保證數據能實時地被收集并傳到Spark Streaming進行實時分析。

Flume能監控某一指定的目錄下的文件或整合自定義的數據收集器，能將新增的數據實時地傳輸到數據中心服務器，并作為Kafka消息隊列的消息來源。

Kafka則能將接收到的實時數據轉存為有序的消息隊列，以有序的方式分別存儲在不同服務器的Partition上，讀取消息時，通過并發方式從不同Partition讀取數據并使用ZeroCopy技術保證消息快速有序的讀取，后面會有相關的實驗進行測試和說明。

Spark Streaming能將接收的數據匯總成一個個小的數據集，并復制多份存儲，然后進行微批量的實時處理，其這一特性很好地避免了并發數據處理中頻繁的任務分配和調度問題，能達到次秒級延時的實時處理。

3.3 可擴展性分析

為了適應數據量的不斷增長，需要系統具有很好的可擴展性。本系統的擴展性體現在兩個方面：

1)數據收集可擴展。隨著數據總量增長，已有資源不一定能滿足數據收集的要求，這就需要數據收集系統具有很好的擴展性。而本系統中的Flume和Kafka都能通過擴展集群，讓處理能力得到近乎線性的增長，很好地滿足可擴展性要求。

2)數據分析可擴展。Spark Streaming是一個分布式的實時處理系統，其數據處理能力也是隨著集群數量的擴展而遞增，實驗部分會有相關的數據證明。目前，已經有多個超過上千臺Spark服務器組成的集群應用在生產環境當中，證明了其在數據分析方面的可擴展性。

3.4 容錯性分析

為了保證系統可靠和穩定的運行，容錯性也是絕大多數系統需要考慮的一個關鍵性因素。在本系統中的容錯性體現在：

1)數據收集的可靠性。Flume能將數據緩存在本地磁盤，并記錄已傳輸數據的偏移量，若由于網絡或其他原因導致的數據傳輸失敗，會在故障恢復時，繼續之前的數據傳輸；而Kafka則是通過復制因子來控制數據保存的份數，通過將數據復制到不同的機器上，解決了數據丟失或系統單點故障問題。

2)數據處理的可靠性。Spark Streaming有一種很好的容錯機制—— 提前日志寫(Write Ahead Logs, WAL)機制，即：先將數據寫入日志，再計算。通過將實時收集的數據寫入到日志中，并復制多份，能解決計算過程中單點故障造成的數據丟失，便于故障恢復和轉移。

4 數據分析系統相關的實驗分析

實驗環境說明：本實驗使用vSphere虛擬機搭建Linux集群，每個Linux虛擬機的配置為2×2核CPU，內存4 GB，內網網速1 000 Mb/s，Kafka版本0.8.21，Flume版本1.6.0，Spark版本1.4.1。

為了測試Kafka接收和讀取消息速度，本實驗使用了Kafka自帶的壓測工具，并通過其腳本kafka-consumer-perf-test.sh和kafka-producer-perf-test.sh進行測試，配置項都使用默認值。圖9和圖10是在不同數量Kafka Server集群下，Kafka消息接收和讀取速度的測試。其中圖9是Kafka接收和讀取消息每秒的字節數測試結果，圖10是Kafka每秒鐘接收和讀取的消息數量測試結果。雖然，測試結果的好壞由很多因素決定：磁盤讀寫速度、CPU、內存、網絡、配置參數等，但是從圖9和圖10可以明顯看出，Kafka消息處理的能力隨著集群數量的增加呈近似線性增長關系，證實了Kafka具有線性擴展能力。

除了數據接收和讀取能力測試外，本文還使用spark-perf[19]和visualvm[20]等對Spark Streaming數據處理能力進行壓力測試和監控。圖11是對應的測試數據結果，從圖中可以看出Spark Streaming處理數據的能力也跟集群數量成正比關系。當然，數據處理速度的具體數值會受到處理復雜度的影響，但是總體趨勢還是類似的，即通過擴展Spark集群數量，可以提高集群數據的處理速度和數據量。

圖9 Kafka服務器數量與消息吞吐量的關系Fig. 9 Relationship between Kafka server number and message throughput

圖10 Kafka服務器數量與消息吞吐率的關系Fig. 10 Relationship between Kafka server number and message throughput rate

圖9～11對由Kafka、Flume和Spark Streaming組成的數據分析系統，在數據吞吐量和數據處理速度方面進行測試，測試結果可知：整個數據分析系統的數據吞吐量隨Kafka集群數成線性增長，數據實時分析處理能力隨Spark Streaming集群成線性增長。

圖11 Spark Streaming處理速度測試Fig. 11 Processing speed test for Spark Steaming

表2對3種不同的數據分析場景和特點進行了對比：單機環境下不能處理大量的數據，沒有容錯性，不能進行擴展，其數據處理能力是有限制的；Hadoop雖然可以線性擴展，能處理大量的數據，但其設計初衷就是為了離線處理海量數據。因此，Hadoop不適合于實時處理場合；Spark Streaming在各方面有明顯的優勢。

表2 單機、Hadoop、Spark Streaming對比Tab. 2 Comparison of single-machine, Hadoop, Spark Streaming

5 實時數據分析系統的應用

K-Means[21]是一種典型的局域原型的目標函數聚類算法，屬于無監督學習范疇，已成功應用于網絡環境的DDoS檢測系統中。為了將經典的K-Means算法應用于大數據環境下的實時數據分析系統中，本文根據Spark運行的原理設計動態采樣的改進K-Means并行化算法[22]，并將該算法成功應用于基于Spark Streaming實時數據分析系統中。

5.1 動態采樣K-Means算法

在普通的聚類算法[22]中，隨機均勻選擇k個點作為聚類中心，k的取值以及初始中心的選擇將直接影響最終聚類的效果。如果k選取不當，K-Means算法在計算過程中易收斂于局部最優解，從而無法得到正確結果。在DDoS檢測系統中，K-Means聚類需要處理大量混雜攻擊流的數據，這對初始中心的選擇造成了極大的困難。為了解決這一問題，采用一種動態采樣的K-means聚類改進算法，以滿足DDoS檢測系統的需求，算法流程如圖12。

圖12 動態采樣改進的K-Means算法流程Fig. 12 Flow of improved K-Means algorithm by dynamic sampling

改進K-Means算法的主要思想是預先只選擇一個點作為聚類中心，構建一個規模函數V(X)，該函數表示數據點到其所屬的聚類中心的距離平方和，通過不斷迭代最小化規模函數值使聚類結果收斂。改進K-Means算法的主要原理：首先從數據集合X中選擇一個點作為始聚類中心并加入動態采樣集合C，根據規模函數V(X)計算得到初始規模N，然后進行N次的循環；在每次循環內再選取m個點，每次循環計算采樣概率P(X)，采樣概率的意義是：聚類中心是相對分散的點，離本聚類中心越遠，其成為另外一個聚類中心的可能性越高，即是盡可能地選擇遠離當前聚類中心的點作為采樣數據。每經過一次迭代，重新計算規模函數V(X)的值，更新下一次采樣的概率。然后將本次采樣的簇中心點集合C′與原采樣集合C求并集作為新的采樣集合C。當N次循環結束后，產生一個新的采樣集合C，該集合中一共有N*m個數據。此時得到的數據集C的規模遠小于初始數據集X，且數據是經過篩選的相對集中。最后再對C運行普通的K-Means算法，因為C是經過預處理得到的集合，因此整個聚類算法將會異常迅速。本算法在時間復雜度上也有改進，采用了N次迭代替代定義收斂閾值的方法，降低了運算過程中執行迭代的次數，對于通過機器學習方法檢測大數據環境下的DDoS攻擊具有重要意義。

具體的算法定義如下：

定義1 規模函數V(X)定義如式(1)所示：

(1)

其中D2(X,C)表示X中的點到所處聚類中心的距離平方和。

定義2 動態采樣概率函數P(X)定義如式(2)所示:

(2)

定義3 初始限定規模函數值V，初始采樣個數m(m

具體算法實現如下：

輸入 數據集,K；

輸出K個聚類中心。

步驟1 從集合X中隨機取1個點加入集合C；

步驟2 根據式(1)計算C的初始限定規模函數值，記為V；

步驟3 循環lgV=N次開始，按式(2)計算動態采樣概率記為P，從集合X中按概率P取出m個點加入集合C′，求C∪C′，記為C，循環結束；

步驟4 采用普通K-Means算法求出集合C的聚類中心。

5.2 基于Spark動態采樣改進K-Means算法并行化

普通的單機DDoS檢測算法無法直接在Spark平臺上運行，根據Spark運行的原理設計動態采樣的改進K-Means算法的并行化。具體的流程如下：

1)算法開始，Master節點上程序從數據輸入源獲取初始數據集。這里的輸入源是預先定義的接口，可以通過多種途徑獲取數據，如InputStream、Hadoop分布式文件系統 (Hadoop Distributed File System, HDFS)、本地文件等，這一設計方便對本算法的測試。獲取數據后，系統將其轉換為RDD1，并調用cache方法將RDD1加載至內存，該RDD作為待處理的數據。

2)開始進行數據的分割，為并行化作準備。系統以塊為單位(64 MB)將RDD1劃分成多個分塊。接下來Master節點調用map方法，把眾多的數據塊分配到多個Worker節點上。Worker節點接收數據塊，并執行Master的map指令，對數據塊進行處理。經過這一步驟，原數據集中的String文本被轉換為DenseVector向量對象，這些對象是程序能夠直接使用的數據，并且分布在各個Worker節點上等待計算。map方法結束后，RDD1生成了新的RDD2。

3)隨機選取初始的聚類中心。程序調用takeSample方法，從RDD2中選取一個作為聚類中心向量，并創建RDD3對象。

4)開始進入循環過程，程序根據5.1節中具體算法實現中的步驟3進行迭代計算。在每次循環體中根據定義1和定義2重新計算當前采樣概率函數P，接著調用takeSample方法根據概率P選取新的RDD向量作為中心點。經過一次循環，一共采樣了1+m個向量，生成RDD4。緊接著系統調用union方法，把RDD3和RDD4合并成RDD5。

5)經過lgV次后，循環結束，此時RDD5中的向量個數≤1+m*lgV。這一數據量遠遠小于初始的數據量。

6)系統將RDD5作為結果輸出。

上述整個采樣階段RDD轉換流程如圖13所示。圖中圓角矩形框代表RDD；RDD內直角矩形框表示該RDD中的數據分片，這些分片散布在不同的Worker節點上；箭頭指向的方向表示RDD轉換的過程。

圖13 采樣階段RDD轉換流程Fig. 13 RDD conversion process in sampling phase

5.3 基于實時數據分析系統的DDoS攻擊檢測

為了測試動態采樣的改進K-Means并行化算法在基于Spark的數據分析系統上的檢測速度和準確率，設計了如下實驗：采用KDD99[23]數據集的訓練全集(500萬條數據)作為實驗樣本，分別從中抽取5組作為實驗數據。這五組的數據量分別為：1萬、50萬、100萬、200萬、500萬。

實驗設計了3個實驗: 第1個實驗，采用單機算法串行處理數據樣本；第2個實驗采用在Spark集群上實現的普通K-Means算法，并行處理數據樣本；第3個實驗采用在Spark的數據分析系統實現的動態采樣的改進K-Means算法。

分別對3個實驗的時間消耗、每輪迭代平均時間、正確率進行統計分析,最終的結果如圖14～16所示。

圖14 三種算法在5組實驗的耗時對比Fig. 14 Comparison of time-consumption of three algorithms in 5 experiments

圖15 三種算法在5組實驗的每輪迭代平均時間對比Fig. 15 Comparison of average time for each iteration of three algorithms in 5 experiments

圖16 兩種算法的正確率對比Fig. 16 Comparison of accuracy rates of two algorithms

通過圖14和圖15對比發現，在數據量較小的情況下，三種算法的耗時差距不大，單機運行的算法正確率也相對較高，Spark的數據分析系統并行計算的優勢不明顯；當數據量大于100萬時，單機運行的時間急劇增長，準確率迅速下降，此時Spark的數據分析系統并行計算的優勢相當顯著。對比在Spark的數據分析系統實現的K-Means和動態采樣的改進的K-Means算法，改進的算法在正確率上保持相對穩定，而算法的效率則比普通方法更高。從圖16可以進一步看清動態采樣改進的K-Means算法和普通K-Means算法準確率的差別，本實驗能較好地體現基于Spark的實時數據分析系統和動態采樣的改進K-Means算法相結合的優勢。

6 結語

本文在已有研究基礎上，進一步地將網絡數據分析與分布式計算相結合，克服已有數據分析方案的時延大、缺乏時效性、不能滿足大數據實時處理和分析的缺陷。通過將網絡數據抓取、收集并導入Kafka，產生實時的消息隊列，再使用Spark Streaming進行網絡數據實時分析：一方面可以實時地對網絡數據進行分析，為大數據環境提供快速高效的異常檢測；另一方面，滿足了大數據環境下數據處理速度和數據處理量的要求。

通過驗證分布式實時計算框架Spark Streaming在大規模網絡環境中對海量網絡數據實時分析的可行性；另一方面，確定了大規模實時網絡檢測的整體架構，為進一步研究大規模網絡數據實時分析和檢測理論和實現技術奠定基礎；最后，為了驗證實時數據分析系統在大數據環境下的DDoS攻擊檢測能力，本文根據Spark運行的原理設計了動態采樣改進K-Means并行化算法，并將該算法應用于基于Spark Streaming實時數據分析系統中，實驗結果顯示基于Spark Streaming的實時分析系統用于大數據環境下的DDoS檢測是可行的。下一步，將結合已有的隱半馬爾可夫模型異常檢測算法，進一步設計能實時有效檢測大數據環境的各種DDoS攻擊行為的算法。

References)

[1] Incapsula. Report: 2014 DDoS Trends-Botnet activity is up by 240%[EB/OL].[2014-06-20].https://www.incapsula.com/blog/ddos-threat-landscape-report-2014.html.

[2] Incapsula. Lax security opens the door for mass-scale abuse of SOHO routers[EB/OL].[2015-09-20].https://www.incapsula.com/blog/ddos-botnet-soho-router.html.

[3] Incapsula. DDoS impact survey reveals the actual cost of DDoS attacks[EB/OL]. [2015-08-20].https://www.incapsula.com/blog/ddos-impact-cost-of-ddos-attack.html.

[4] ZAHARIA M, DAS T, LI H, et al. Discretized streams: an efficient and fault-tolerant model for stream processing on large clusters[C]// Proceedings of the 4th USENIX Conference on Hot Topics in Cloud Ccomputing. Berkeley, CA: USENIX Association, 2012:10.

[5] 孫大為,張廣艷,鄭緯民.大數據流式計算：關鍵技術及系統實例[J].軟件學報,2014,25(4):839-862.(SUN D W, ZHANG G Y, ZHENG W M. Big data stream computing: technologies and instances[J]. Journal of Software, 2014,25(4):839-862.)

[6] LEE Y, KANG W, SON H. An Internet traffic analysis method with MapReduce[C]// Proceedings of the 2010 IEEE/IFIP Network Operations & Management Symposium Workshops. Piscataway, NJ: IEEE, 2010:357-361.

[7] KHATTAK R, BANO S, HUSSAIN S, et al. DOFUR: DDoS forensics using MapReduce[C]// Proceedings of the 2011 Frontiers of Information Technology. Washington, DC: IEEE Computer Society, 2011:117-120.

[8] GOODHOPE K, KOSHY J, KREPS J. Building LinkedIn’s real-time activity data pipeline[C]// Proceedings of the 2012 IEEE Computer Society Technical Committee on Data Engineering. Washington, DC: IEEE Computer Society, 2012: 33-45.

[9] KREPS J, CORP L, NARKHEDE N, et al. Kafka: a distributed messaging system for log processing[C]// Proceedings of the 2011 ACM SIGMOD Workshop on Networking Meets Databases. New York: ACM, 2011:231-240.

[10] Apache Software Foundation. Apache Kafka[EB/OL]. [2015-09-16]. http://kafka.apache.org/.

[11] Apache Software Foundation. Apache Spark: Lightning-fast cluster computing[EB/OL]. [2015-09-16]. http://spark.apache.org/.

[12] Apache Software Foundation. Welcome to Apache Hadoop[EB/OL]. [2015-09-16]. http://hadoop.apache.org/.

[13] ZAHARIA M, CHOWDHURY M, DAS T, et al. Resilient distributed datasets: a fault-tolerant abstraction for in-memory cluster computing[C]// Proceedings of the 9th USENIX Conference on Networked Systems Design and Implementation. Piscataway, NJ: IEEE, 2011: 141-146.

[14] Apache Software Foundation. S4: distributed stream compute platform[EB/OL].[2015-09-16]. http://incubator.apache.org/s4/.

[15] Apache Software Foundation. Storm: distributed and fault-tolerance real-time computation[EB/OL]. [2015-09-16]. http://storm.apache.org/.

[16] Apache Software Foundation. Apache Kafka[EB/OL]. [2015-09-16]. http://kafka.apache.org/.

[17] 崔星燦,禹曉輝,劉洋,等. 分布式流處理技術綜述[J]. 計算機研究與發展, 2015, 52(2): 318-332.(CUI X C, YU X H, LIU Y, et al. Distributed stream processing: a survey[J]. Journal of Computer Research and Development, 2015, 52(2): 318-332.)

[18] Apache Software Foundation. Apache Flume[EB/OL].[2016-01-20]. http://flume.apache.org/.

[19] Databricks. Spark-perf[EB/OL]. [2015-09-16]. https://github.com/databricks/ spark-perf.

[20] 虞立軍, 王建光, 倪力. 使用VisualVM 進行性能分析及調優[EB/OL]. [2013-02-18]. http://www.ibm.com/developerworks/ cn/java/j-lo-visualvm/.(YU L J, WANG J G, NI L. performance analysis and tuning by using VisualVM[EB/OL]. [2013-02-18]. http://www.ibm.com/developerworks/ cn/java/j-lo-visualvm/.)

[21] 周麗娟, 王慧, 王文伯,等. 面向海量數據的并行K-Means算法[J]. 華中科技大學學報(自然科學版), 2012,40(S1):150-152.(ZHOU L J, WANG H, WANG W B, et al. ParallelK-Means algorithm for mass data[J]. Journal of Huazhong University of Science and Technology(Natural Science Edition), 2012,40(S1):150-152.)

[22] 劉罕. 基于Spark框架的DDoS攻擊檢測系統研究[D].上海: 上海海事大學, 2016:5.(LIU H. Research of the DDoS Attack Detection System base on the Spark Framework[M]. Shanghai: Shanghai Mritime University, 2016:5.)

[23] HETTICH S, BAY S D. KDD cup 1999 data [EB/OL]. [1999-10-20]. http://kdd.ics.uci.edu/databases/kddcup99/kddcup99.html.

This work is partially supported by the National Natural Science Foundation of China (61672338，61373028).

HAN Dezhi, born in 1966, Ph.D., professor. His research interests include cloud computing, cloud storage and security, big data application.

CHEN Xuguang, born in 1993, M.S. candidate. His research interests include cloud computing, big data real-time analysis.

LEI Yuxin, born in 1996. Her research interests include data mining, network security.

DAI Yongtao, born in 1991, M.S. candidate. His research interests include cloud computing, distributed computing, data mining, network security.

ZHANG Xiao, born in 1994, M.S. candidate. Her research interests include cloud computing, big data real-time analysis.

Real-time data analysis system based on Spark Streaming and its application

HAN Dezhi1*, CHEN Xuguang1, LEI Yuxin2, DAI Yongtao1, ZHANG Xiao1

(1.CollegeofInformationEngineering,ShanghaiMaritimeUniversity,Shanghai201306,China;2.SchoolofInformationEngineering,ZhengzhouUniversity,ZhengzhouHenan450001,China)

In order to realize the rapid analysis of massive real-time data, a Distributed Real-time Data Analysis System (DRDAS) was designed, which resolved the collection, storage and real-time analysis for mass concurrent data. And according to the operation principle of Spark Streaming, a dynamic samplingK-means parallel algorithm was proposed, which could quickly and efficiently detect all kinds of DDoS (Distributed Denial of Service) attacks. The experimental results show that the DRDAS has good scalability, fault tolerance and real-time processing ability, and along with newK-means parallel algorithm, the DRDAS can real-time detect various DDoS attacks, and shorten the detecting time of attacks.

Spark Streaming framework; distributed stream processing; network data analysis; Distributed Denial of Service (DDoS) attack

2016-07-15;

2016-11-26。 基金項目:國家自然科學基金資助項目(61373028,61672338)。

韓德志(1966—)，男，河南信陽人，教授，博士，CCF會員，主要研究方向：云計算、云存儲及其安全、大數據應用； 陳旭光(1993—)，男，河南信陽人，碩士研究生，主要研究方向：云計算、大數據實時分析； 雷雨馨(1996—)，女，河南鄭州人，主要研究方向：數據挖掘、網絡安全； 戴永濤(1991—)，男，湖南邵陽人，碩士研究生，主要研究方向：云計算、分布式計算、數據挖掘、網絡安全； 張肖(1994—)，女，安徽蚌埠人，碩士研究生，主要研究方向：云計算、大數據實時分析。

1001-9081(2017)05-1263-07

10.11772/j.issn.1001-9081.2017.05.1263

TP316.2

A