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

邊靜，戈振興

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

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

1 安全分析平臺總體設(shè)計

1.1 平臺需求分析

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

1.2 技術(shù)介紹

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

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

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

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

2 安全分析平臺搭建與應用

2.1 日志收集

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

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

2.2 日志分析處理

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

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

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

2.3 安全分析平臺應用

2.3.1 日志查詢和可視化展示

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

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

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

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

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

3 總結(jié)與展望

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