基于Kafka 和Spark 的網(wǎng)絡(luò)安全分析平臺(tái)的構(gòu)建與應(yīng)用

邊靜，戈振興

（集寧師范學(xué)院，內(nèi)蒙古 集寧 012000）

隨著計(jì)算機(jī)技術(shù)和網(wǎng)絡(luò)技術(shù)的高速發(fā)展，網(wǎng)絡(luò)應(yīng)用逐漸深入各行各業(yè)，網(wǎng)絡(luò)安全問(wèn)題也隨之越來(lái)越多，引起了相關(guān)各方面的高度重視。網(wǎng)絡(luò)運(yùn)維人員可通過(guò)對(duì)網(wǎng)絡(luò)日志分析來(lái)監(jiān)控網(wǎng)絡(luò)運(yùn)行狀態(tài)及用戶訪問(wèn)情況，及時(shí)發(fā)現(xiàn)各類故障及問(wèn)題，提升運(yùn)維管理水平[1]。此外，當(dāng)前的網(wǎng)絡(luò)攻擊行為越來(lái)越難以發(fā)現(xiàn)，APT 攻擊可以跳過(guò)網(wǎng)安設(shè)備，攻擊者在得手后會(huì)銷毀這些攻擊記錄，因此，如何快速實(shí)時(shí)地存儲(chǔ)備份這些重要的日志信息和迅速地反饋這類信息成為目前的當(dāng)務(wù)之急，通過(guò)建立可以將各類系統(tǒng)的日志信息進(jìn)行統(tǒng)一集中的存儲(chǔ)，網(wǎng)絡(luò)安全分析平臺(tái)可以解決這些問(wèn)題[2]。

1 安全分析平臺(tái)總體設(shè)計(jì)

1.1 平臺(tái)需求分析

平臺(tái)需處理的日志信息主要有網(wǎng)絡(luò)設(shè)備日志、應(yīng)用服務(wù)器系統(tǒng)日志、數(shù)據(jù)中心服務(wù)器日志、機(jī)房動(dòng)力系統(tǒng)日志等。各種日志的內(nèi)容不同、格式多樣、結(jié)構(gòu)復(fù)雜，日志信息中有價(jià)值的信息密度較低，非專業(yè)的人士很難讀懂；由于信息化的高速發(fā)展、應(yīng)用業(yè)務(wù)的數(shù)量增長(zhǎng)、用戶使用的增加、網(wǎng)絡(luò)流量越來(lái)越大，日志信息數(shù)據(jù)增長(zhǎng)很快，大量的日志難以收集、整理、存儲(chǔ)；需要具備TB 級(jí)的海量日志分析存儲(chǔ)平臺(tái)來(lái)滿足這一需求。本安全分析平臺(tái)采用了Hadoop、Kafka、Spark 實(shí)時(shí)計(jì)算技術(shù)等，構(gòu)建了一個(gè)分布式存儲(chǔ)、實(shí)時(shí)計(jì)算的處理分析平臺(tái)。利用分布式技術(shù)的高可用、高可擴(kuò)展、高吞吐等特性，實(shí)現(xiàn)了日志信息的存儲(chǔ)及實(shí)時(shí)分析處理。

1.2 技術(shù)介紹

目前在大數(shù)據(jù)存儲(chǔ)領(lǐng)域的新技術(shù)中Hadoop 是最為廣泛使用的技術(shù)之一，它是一個(gè)集分布式存儲(chǔ)、管理和計(jì)算于一體的生態(tài)系統(tǒng)。Hadoop 生態(tài)系統(tǒng)的核心是分布式文件系統(tǒng)（Hadoop Distributed File System，簡(jiǎn)稱HDFS）和MapReduce 框架，HDFS 實(shí)現(xiàn)了一個(gè)集群式的存儲(chǔ)系統(tǒng)，MapReduce 框架可以在HDFS 上實(shí)現(xiàn)大數(shù)據(jù)的分析計(jì)算[3]。Hadoop 工作原理：在集群的多個(gè)節(jié)點(diǎn)上按照一定的冗余進(jìn)行數(shù)據(jù)的存儲(chǔ)，通過(guò)塊保存到多個(gè)節(jié)點(diǎn)上，為海量數(shù)據(jù)提供高可靠性、高容錯(cuò)性、高可擴(kuò)展性、高吞吐的存儲(chǔ)方案[4]。

Spark 是一個(gè)易于使用的、強(qiáng)大的分布式處理大數(shù)據(jù)框架[5]，主要應(yīng)用在復(fù)雜的大數(shù)據(jù)處理場(chǎng)景中。目前已經(jīng)在很多實(shí)際的生產(chǎn)環(huán)境中廣泛使用，它可以穩(wěn)定地操作PB 級(jí)的數(shù)據(jù)。它可以完成日志數(shù)據(jù)的抽取、清洗、轉(zhuǎn)化、分析等功能。在Spark 中，數(shù)據(jù)被高度抽象且存儲(chǔ)在彈性分布式數(shù)據(jù)集RDD 中。Spark 的處理速度比MapReduce 快10～100 倍，因此它是比Hadoop 更加高效的大數(shù)據(jù)計(jì)算框架。每次RDD 數(shù)據(jù)集的操作結(jié)果都可以保存在內(nèi)存中，這樣下一個(gè)操作可以直接從內(nèi)存中讀取數(shù)據(jù)，省去了MapReduce 中大量的磁盤(pán)I/O操作，明顯提升了處理的速度。Spark 本身并沒(méi)有提供分布式文件系統(tǒng)，因此Spark 分析處理后將數(shù)據(jù)存儲(chǔ)于HDFS 中，將基于內(nèi)存計(jì)算的Spark 與Hadoop 相互結(jié)合可提高各類處理的運(yùn)行效率和處理規(guī)模。

1.3 平臺(tái)總體結(jié)構(gòu)

各種需要處理的日志信息數(shù)據(jù)首先會(huì)通過(guò)網(wǎng)絡(luò)按照一定的規(guī)則被發(fā)送到各自的服務(wù)器中，然后各個(gè)服務(wù)器根據(jù)Flume NG 特有的接收方式將日志數(shù)據(jù)發(fā)送到特定的端口，根據(jù)實(shí)時(shí)處理和批處理的兩種分析類型，F(xiàn)lume NG 集群將日志數(shù)據(jù)進(jìn)行分別分發(fā)：需要實(shí)時(shí)處理的數(shù)據(jù)會(huì)通過(guò)Kafka Sink 發(fā)送到Kafka[6]集群中等待實(shí)時(shí)分析處理程序處理，Spark 的Spark Streaming 模塊以消費(fèi)者的消費(fèi)數(shù)據(jù)的方式從Kafka 中提出數(shù)據(jù)進(jìn)行相應(yīng)的實(shí)時(shí)分析處理，分析處理后的結(jié)果會(huì)被傳輸?shù)紿Base 分布式開(kāi)源數(shù)據(jù)庫(kù)中，系統(tǒng)提供一個(gè)統(tǒng)一的可視化查詢界面來(lái)對(duì)分析處理后的結(jié)果進(jìn)行展示；另一部分需要批處理的日志數(shù)據(jù)通過(guò)HDFS Sink 發(fā)送到Hadoop集群的HDFS 中，這些在HDFS 中的數(shù)據(jù)既可以作為日志的備份也可以通過(guò)Spark 的批處理方式進(jìn)行進(jìn)一步的數(shù)據(jù)挖掘，從中提取更有價(jià)值的信息。Zookeeper 負(fù)責(zé)統(tǒng)一實(shí)現(xiàn)平臺(tái)中的Hadoop、Kafka、Spark、HBase 等集群的分布式協(xié)調(diào)服務(wù)。

2 安全分析平臺(tái)搭建與應(yīng)用

2.1 日志收集

平臺(tái)的日志收集模塊主要由分布式日志收集系統(tǒng)Flume NG 實(shí)現(xiàn)，它是一個(gè)輕量級(jí)的數(shù)據(jù)采集工具，具有高可用、高容錯(cuò)、易擴(kuò)展、可恢復(fù)、高性能等優(yōu)勢(shì)，經(jīng)過(guò)簡(jiǎn)單配置就可實(shí)現(xiàn)與Kafka、Hadoop 等主流框架無(wú)縫對(duì)接，提供高效的、準(zhǔn)確的數(shù)據(jù)采集服務(wù)。Flume NG 中，按照其內(nèi)部設(shè)計(jì)結(jié)構(gòu)分為了數(shù)據(jù)采集層（Source）、數(shù)據(jù)緩沖層（Channel）和數(shù)據(jù)處理層（Sink）三個(gè)層次。客戶端按照事先訂制的規(guī)則將日志數(shù)據(jù)封裝成一系列的Event，Event 就是Flume NG 中的最小數(shù)據(jù)傳輸單位。Source 負(fù)責(zé)接收上一階段生成的Event，然后將它寫(xiě)入Channel 中，F(xiàn)lume 提供了多種Source 的實(shí)現(xiàn)，例如：Taildir Source、Exec Source、Kafka Source 等，用戶也可以自定義自己的Source；Channel 類似于Flume 內(nèi)部的一個(gè)消息緩沖隊(duì)列，它的任務(wù)是暫時(shí)將Source 端發(fā)送來(lái)的數(shù)據(jù)緩沖在Flume 中等待Sink 端拉取，常用的Channel 類型有：Memory Channel、File Channel 等；Sink 是負(fù)責(zé)拉取Channel 中的數(shù)據(jù)并發(fā)送到目標(biāo)系統(tǒng)或下一個(gè)節(jié)點(diǎn)的Source 中，它的類型有HDFS Sink、Kafka Sink 等[7]。在Flume 中Source、Channel、Sink 都包含在一個(gè)稱為Agent 的進(jìn)程中，通過(guò)一系列的Agent組件將分布在不同節(jié)點(diǎn)的日志收集到指定的位置進(jìn)而實(shí)現(xiàn)數(shù)據(jù)采集的任務(wù)。系統(tǒng)中將實(shí)時(shí)數(shù)據(jù)流通過(guò)Kafka Sink 端發(fā)送到Kafka 集群中供Spark Streaming 處理，將批處理數(shù)據(jù)通過(guò)HDFS Sink 端發(fā)送到Hadoop 的HDFS中做后續(xù)的分析處理。這里的Channel 選擇了Memory Channel，這樣可以提高傳輸?shù)乃俣?，但需要配置更多的?nèi)存。

Kafka 集群接收需要進(jìn)行實(shí)時(shí)分析計(jì)算的日志數(shù)據(jù)，在其集群內(nèi)部的每個(gè)節(jié)點(diǎn)都會(huì)創(chuàng)建相應(yīng)的broker實(shí)例緩存數(shù)據(jù)，broker 內(nèi)按照topic 對(duì)數(shù)據(jù)進(jìn)行分類存儲(chǔ)，在broker 兩端，各有多個(gè)producer 和consumer。這里Flume 充當(dāng)producer 的作用，完成向broker 發(fā)送日志數(shù)據(jù)的任務(wù)，Spark Streaming 的實(shí)時(shí)計(jì)算程序充當(dāng)consumer 的作用，不斷地從broker 中拉取各個(gè)topic 中緩存的日志數(shù)據(jù)。

2.2 日志分析處理

日志處理部分是本平臺(tái)的核心業(yè)務(wù)處理部分，該部分主要包含了兩種類型的業(yè)務(wù)：對(duì)需要實(shí)時(shí)結(jié)果的業(yè)務(wù)進(jìn)行實(shí)時(shí)計(jì)算；對(duì)時(shí)效性要求不高的業(yè)務(wù)進(jìn)行離線分析處理。

實(shí)時(shí)業(yè)務(wù)主要由Spark 集群中的Spark Streaming[8]模塊來(lái)實(shí)現(xiàn)，Spark Streaming 使用的是一種微批處理的工作模式，它將從Kafka 中流入的數(shù)據(jù)按照秒級(jí)的時(shí)間單位劃分成很短的數(shù)據(jù)單元，然后對(duì)這些數(shù)據(jù)進(jìn)行微批處理，每塊數(shù)據(jù)會(huì)被作為一個(gè)RDD。DStream 是Spark Streaming 提供的高級(jí)別抽象數(shù)據(jù)流，它實(shí)際上就是在時(shí)間維度上連續(xù)的RDD 序列，因此對(duì)DStream 的操作實(shí)際上也會(huì)被轉(zhuǎn)換成對(duì)RDD 的操作，由此來(lái)完成對(duì)每一個(gè)小塊數(shù)據(jù)的處理，Spark Streaming 的流計(jì)算就是通過(guò)這種快速小批量的批處理的方式來(lái)實(shí)現(xiàn)接近于實(shí)時(shí)計(jì)算的，所以從理論上來(lái)說(shuō)它還是存在一定的延遲的，根據(jù)實(shí)驗(yàn)測(cè)試數(shù)據(jù)顯示Spark Streaming 能實(shí)現(xiàn)秒級(jí)的延遲。Spark Streaming 通過(guò)從Kafka 中拉取所需日志數(shù)據(jù)來(lái)進(jìn)行分析處理并最終輸出分析結(jié)果。其具體處理流程如下：由SparkContex 對(duì)象初始化Master 節(jié)點(diǎn)和各個(gè)Worker 節(jié)點(diǎn)，啟動(dòng)StreamingContext 實(shí)例，通過(guò)createDirectStream（）方法創(chuàng)建一個(gè)inputstream 用來(lái)直接從Kafka 中拉取日志數(shù)據(jù)，接著通過(guò)正則表達(dá)式的方法對(duì)數(shù)據(jù)進(jìn)行過(guò)濾，將不符合規(guī)則的數(shù)據(jù)過(guò)濾掉。然后對(duì)過(guò)濾后的數(shù)據(jù)再根據(jù)內(nèi)容進(jìn)行匹配分類，這樣就可以初步得到符合規(guī)則的日志信息。這些信息會(huì)被存儲(chǔ)到HBase 中，供前端查詢或作可視化展示。例如可以對(duì)訪問(wèn)地址做頻次統(tǒng)計(jì)分析，配置好統(tǒng)計(jì)的時(shí)間范圍和其他參數(shù)后，可以在展示界面中看到一段時(shí)間范圍內(nèi)某類事件出現(xiàn)頻次的統(tǒng)計(jì)排行榜[9]。

離線分析業(yè)務(wù)是對(duì)HDFS 中收集的原始?xì)v史日志數(shù)據(jù)進(jìn)行分析處理，通過(guò)進(jìn)行更深入的數(shù)據(jù)挖掘處理可以發(fā)現(xiàn)日志數(shù)據(jù)中隱含的一些有價(jià)值的信息。離線日志分析是以Spark 的批處理工作模式實(shí)現(xiàn)的，它與Spark Streaming 實(shí)時(shí)日志處理稍有不同的，它主要被設(shè)計(jì)用來(lái)完成一些大數(shù)據(jù)量的業(yè)務(wù)，其執(zhí)行的處理流程也和實(shí)時(shí)業(yè)務(wù)模塊類似，也是對(duì)日志進(jìn)行一定的清洗、過(guò)濾、分類等操作，然后根據(jù)事先設(shè)計(jì)好的分析算法進(jìn)入相應(yīng)的處理流程，處理結(jié)果持久化到HDFS或MySQL 中。

2.3 安全分析平臺(tái)應(yīng)用

2.3.1 日志查詢和可視化展示

經(jīng)過(guò)日志分析處理模塊得到的結(jié)果會(huì)存儲(chǔ)在HBase或MySQL 中，可以以時(shí)間區(qū)間為條件對(duì)結(jié)果數(shù)據(jù)進(jìn)行數(shù)據(jù)分組查詢，也可以根據(jù)不同的條件組合查詢某個(gè)IP 相關(guān)的所有的連接信息，還可以查詢發(fā)生次數(shù)最多的源IP、源端口、目的IP 和端口等內(nèi)容。此外，在可視化展現(xiàn)模塊中，通過(guò)相應(yīng)的配置，可以將分析結(jié)果以柱狀圖、折線圖和餅狀圖等形式進(jìn)行展現(xiàn)，可視化的展現(xiàn)形式可以更加直觀地展現(xiàn)日志數(shù)據(jù)的量化值、趨勢(shì)特征和百分比等統(tǒng)計(jì)分析結(jié)果。

2.3.2 故障告警和設(shè)備監(jiān)控

通過(guò)對(duì)記錄了系統(tǒng)中硬件、軟件和系統(tǒng)運(yùn)行期間產(chǎn)生的各種信息的分析處理可以快速發(fā)現(xiàn)硬件故障、系統(tǒng)宕機(jī)、應(yīng)用系統(tǒng)崩潰、數(shù)據(jù)庫(kù)存儲(chǔ)及備份故障等問(wèn)題，及時(shí)排查錯(cuò)誤產(chǎn)生的原因，幫助運(yùn)維人員快速定位故障位置并恢復(fù)業(yè)務(wù)。在網(wǎng)絡(luò)攻擊活動(dòng)中攻擊者常常在攻擊成功后會(huì)清除攻擊行為記錄等違法行為痕跡，甚至刪除系統(tǒng)日志；日志分析平臺(tái)的收集模塊也可以起到對(duì)日志的快速備份的作用，這樣即便攻擊者刪除了設(shè)備中的攻擊日志記錄，在分析平臺(tái)中還存儲(chǔ)了備份，這些信息同樣也可以作為攻擊者違法行為的證據(jù)，為網(wǎng)絡(luò)安全事件的追蹤和溯源提供重要的線索和依據(jù)。

2.3.3 攻擊發(fā)現(xiàn)和非法挖礦

根據(jù)相關(guān)的研究數(shù)據(jù)表明，網(wǎng)絡(luò)攻擊和非法入侵行為逐年增加，分析平臺(tái)可以通過(guò)日志信息分析識(shí)別出大多數(shù)的網(wǎng)絡(luò)攻擊行為和內(nèi)網(wǎng)主機(jī)非法外聯(lián)行為。網(wǎng)絡(luò)攻擊行為通常具有短時(shí)間內(nèi)出現(xiàn)大量的訪問(wèn)的特征，通過(guò)對(duì)日志信息中訪問(wèn)量的統(tǒng)計(jì)結(jié)果可以及時(shí)發(fā)現(xiàn)此類行為。針對(duì)網(wǎng)絡(luò)爬蟲(chóng)、惡意猜解VPN 用戶名密碼等攻擊行為特征，進(jìn)行相應(yīng)的分析處理，也可以及時(shí)發(fā)現(xiàn)這些攻擊行為，這里不再贅述。近幾年隨著挖礦的盛行，服務(wù)器成為其理想的攻擊目標(biāo)之一，由于數(shù)據(jù)中心的服務(wù)器性能較高且7X24 小時(shí)在線，所以經(jīng)常會(huì)被入侵并植入木馬、病毒從而被用來(lái)挖礦。根據(jù)公開(kāi)的網(wǎng)絡(luò)安全威脅信息，對(duì)日志信息中涉及的礦池IP、域名的數(shù)據(jù)進(jìn)行分析可以在早期迅速發(fā)現(xiàn)此類行為，將風(fēng)險(xiǎn)和損失降到最低。

3 總結(jié)與展望

在大數(shù)據(jù)技術(shù)迅猛發(fā)展的今天，面對(duì)越來(lái)越復(fù)雜的網(wǎng)絡(luò)安全環(huán)境，對(duì)運(yùn)維工作提出了更高的要求，日志信息分析處理在網(wǎng)絡(luò)安全管理中的作用也越來(lái)越大，基于Kafka、Spark 等技術(shù)搭建的網(wǎng)絡(luò)安全分析平臺(tái)，可以實(shí)現(xiàn)對(duì)海量日志的存儲(chǔ)和分析，這些技術(shù)也具備了很好的擴(kuò)展性和伸縮性，可以滿足不同規(guī)模的企事業(yè)單位的業(yè)務(wù)需求，由于所使用的技術(shù)均為開(kāi)源技術(shù)，也為經(jīng)費(fèi)不足的單位提供了很好的經(jīng)濟(jì)性。同時(shí)系統(tǒng)具有很好的穩(wěn)定性和可靠性，極大地提高了網(wǎng)絡(luò)管理人員的工作效率，對(duì)進(jìn)一步提升網(wǎng)絡(luò)運(yùn)維的自動(dòng)化程度具有重要的作用。目前該平臺(tái)僅能完成日志信息的存儲(chǔ)、分析、查詢和可視化展示的功能，下一步將計(jì)劃引入機(jī)器學(xué)習(xí)的方法[10]，開(kāi)展日志信息的關(guān)聯(lián)分析、特征分析等處理方法，進(jìn)一步挖掘日志信息中有價(jià)值的數(shù)據(jù)，提高網(wǎng)絡(luò)安全管理水平。