基于Kafka 和Spark 的網絡安全分析平臺的構建與應用

邊靜，戈振興

（集寧師范學院，內蒙古 集寧 012000）

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

1 安全分析平臺總體設計

1.1 平臺需求分析

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

1.2 技術介紹

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

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

1.3 平臺總體結構

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

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

2.1 日志收集

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

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

2.2 日志分析處理

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

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

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

2.3 安全分析平臺應用

2.3.1 日志查詢和可視化展示

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

2.3.2 故障告警和設備監控

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

2.3.3 攻擊發現和非法挖礦

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

3 總結與展望

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