基于Spark的油田應用日志行為分析系統①

陳雷鳴,張偉光,李翛然,李寧寧

(中國石油大學(華東)計算機與通信工程學院,青島 266580)

隨著油田信息化、智能化建設的不斷加快,各類IT系統的在企業中廣泛應用.某油田現有超過一千套業務系統分別由各級單位獨立運維管理.在這些業務系統給企業提供便捷服務的同時,如何對這些業務進行監控分析和安全評估上卻面臨難題.由于油田現有的應用系統數量龐大、類型繁雜、開發技術多樣、部署分散,如何以最小的切入方式完成對這些應用的運行狀況和安全狀況的評估成為企業關注的重要問題.由于這些應用系統和網站每天都會產生大量的日志數據,這些日志中包括用戶的訪問信息和應用安全狀況等信息.通過分析應用日志數據可以評估應用的使用情況、應用運行的安全狀況,進而為各企業信息決策提供重要依據.隨著各類應用系統的規模迅速擴大導致應用所產生的日志數據呈爆炸式增長,若繼續采用傳統的數據儲存和處理方式將無法及時評估出各類業務運行情況和安全狀況.

針對這一難題,主流的海量日志處理方案是借助于大數據計算框架提供的分布式處理技術.大數據技術的發展大致可分為以下階段:第一階段是基于Hadoop提供的MapReduce計算框架做分析.由于MapReduce的編程機制需要嚴格按照Map和Reduce兩個階段,因此缺少了程序設計的靈活性[1,2].然后是Pig[3]數據分析程序以及Hive[4]數據倉庫等工具的出現.這類工具簡化了MapReduce的編程過程,然而在任務執行時依然需要先轉換為MapReduce作業任務然后再交給Hadoop執行[5].由于Hadoop在處理大批量數據時,需要把中間結果緩存到磁盤上,這一過程受限于磁盤IO速率,因此嚴重影響分析效率.針對這一問題,基于內存計算的批處理框架Spark應運而生,由于Spark將數據直接保存在內存中進行多次迭代操作[6],從而不再從磁盤中重復的讀寫數據源,因此具有更快的處理速度.本文基于Spark內存計算框架來代替MapReduce計算框架來提高計算速率,并基于Spark提供的各類功能模塊設計數據分析算法來完成應用日志數據的預處理和行為分析.

1 Spark數據分析平臺

Apache Spark是由加州大學伯克利分校AMP實驗室開發的分布式并行計算框架.Spark支持復雜的機器學習、圖計算和實時流處理等功能模塊[7].如圖1所示為Spark生態圈,從下至上依次為:數據持久層、資源調度層、Spark核心計算層、Spark主要功能組件.其中數據持久層:包括分布式文件系統HDFS和分布式數據庫HBase、Cassandra.資源調度層:為Spark提供統一的資源調度和管理,目前主流的資源調度組件為Yarn和Mesos.

Spark 核心層:包含Spark的基本功能,定義了RDD的API和基本操作,Spark其它的功能模塊都是構建在RDD和Spark Core之上的.最后一層為Spark主要的功能組件包括:用于對流數據實時處理的Spark Streaming、用于機器學習的算法庫MLlib、用于圖操作的算法工具集合GraphX和用于在內存數據集上提供查詢功能的Spark SQL.本文基于Spark 提供的基本算子函數操先對日志數據進行預處理并利用 Spark SQL模塊對預處理后的數據進行指標分析.

圖1 Spark生態圈

2 系統架構設計

根據油田應用系統部署分散的特點,設計應用行為分析平臺架構如圖2所示.應用行為分析平臺主要由數據收集層、數據存儲層、數據處理層、數據可視化層、可視化管理調度平臺5部分組成.

圖2 應用日志行為分析平臺架構圖

數據收集層:用于將分散在各主機應用日志數據集中收集,該模塊基于Flume日志收集框架設計完成.

數據儲存層:日志文件儲存在基于HDFS的文件儲存系統上,基于HBase儲存經預處理分析后的結構化數據,使用MariaDB作為業務數據庫,用于儲存分析的最終結果和系統業務數據.

數據處理層:基于Quartz任務調度框架[8]來完成各類任務的定時執行;基于Yarn來完成集群環境下計算資源的分配和Spark任務調度.基于Spark計算框架設計數據的預處理和數據分析算法.

數據可視化層:用于分析結果的圖表的可視化展示.其中數據的圖形化展示基于Echarts可視化插件來完成,圖表的數據通過報表程序模塊來完成.

3 系統設計與實現

應用日志行為分析平臺需要完成以下功能:系統的可視化管理、各類計算框架的集成管理、分析算法的調度管理,因此需要設計以下三個主要的模塊:基于Web管理平臺、調度引擎和算法數據庫.

Web管理平臺:向用戶提供交互功能和分析結果的可視化展示,該模塊基于SSM框架完成,用于分析任務的管理、分析錯誤告警信息的管理、算法庫管理、各類應用信息的管理以及與平臺業務相關功能.

調度引擎:該模塊基于Apache Felix[9]的OSGI框架開發完成,主要完成不同數據源的數據信息拉取儲存、數據處理分析模塊的調度、分析任務定時執行,該模塊主要利用各類大數據框架提供的API封裝成相應的功能模塊集成開發完成.

算法信息庫:用于儲存與日志行為分析的算法,算法主要基于Java編程語言開發,每個算法為單獨jar包,由調度引擎選擇并提交到Spark集群執行.行為分析系統各類組件的調度流程為:

第一步:通過在應用服務器上安裝日志收集代理(Flume Agent),將分散在各應用服務器的日志文件定時匯集到日志儲存服務器,然后經Flume框架上傳到HDFS文件儲存系統中規劃的文件夾.

第二步:由調度引擎執行定時任務、調度各類框架.并由調度引擎選取各類算法提交給Spark集群.

第三步:Spark集群從HDFS拉取數據,首先對日志數據預處理,并將結果反饋給調度引擎.若處理過程無異常,則將分析結果儲存到HBase數據庫.

第四步:由調度引擎依次進行各類行為分析算法的調度,并將分析結果儲存到相關數據庫中.

3.1 日志數據的預處理

油田應用日志的特點為:業務量較小的應用每天生成一個日志文件,大業務量的應用日志可能會被切分成多個日志文件(在分析處理時若應用每天產生多個日志文件則邏輯上當作一個文件處理).分析系統需要處理前一天所有應用產生的日志文件.因此調度引擎模塊會在每天0點開始執行總的分析任務.每分析一個日志文件就執行一次預處理算法任務.在日志的預處理分析順序上,調度引擎會根據傳輸到HDFS的日志文件順序,按照先來先服務的原則生成任務執行列表,然后依次對各日志進行預處理分析.

圖3 系統模塊調度流程圖

由于油田在部署各類應用系統時使用的服務軟件種類繁多,主流的服務軟件包括:iis、tomcat、apache、nginx等.不同類型的服務軟件產生的日志類型不一樣;同類型的服務軟件可能有多個版本.因此需要設計多種分析規則來處理不同類型的日志.設計的原則為:面向同類型日志分別設計相應的處理規則.其中應用日志預處理分析流程如圖4所示.

圖4 應用日志預處理分析流程

結合各類應用特點和部署環境等因素,數據預處理過程可分為以下階段:數據清洗、用戶識別、會話識別、數據格式化[10].數據清洗階段主要完成對殘缺信息的過濾(字段缺失、信息缺失)、冗余信息的過濾(主要過濾掉與請求無關的靜態數據文件如JS文件、CSS文件、圖片數據等)、核心字段的提取.然后根據客戶端IP地址將訪問信息按照時間先后順序分組排序.在用戶識別階段,采用的是IP和客戶端組合方式來識別.分析規則為:不同的IP為不同用戶,同一IP、不同客戶端為不同用戶.在會話識別階段,采用的是基于固定閥值會話識別算法(固定閾值為 30 min)[11].為了便于下一階段進行應用行為分析,需要對多種日志類型預處理后的結果各字段進行數據格式統一,最后將處理后的數據儲存到HBase數據庫中.

預處理算法的設計主要基于Spark Core模塊提供的操作RDD的算子實現.RDD是Spark計算框架提供的分布式數據架構及彈性數據集,它會在集群環境中的多個節點進行數據分區,但是在邏輯上可看成一個分布式數組[12].預處理算法的設計原理:主要利用Spark提供的各類算子設計相應函數,從而實現對各類RDD的操作;Spark最終會將者一系列對RDD的操作翻譯成有向無環圖(DAG)的形式進行調度和分布式任務分發[13];最終整個執行過程會形成預處理分析算法.根據分析流程設計分析預處理算法:首先日志文件數據會由spark讀取加載到內存,并將源數據轉變成分布式數據集;然后按照各階段目標,設計并實現相應算法模塊或者基于各類算子設計相應的函數實現對已有的RDD進行轉變操作.應用日志預處理算法主要的分析步驟如算法1.

算法1.數據預處理算法1)根據日志類型選取處理方法[A|B|C…].2)利用textfile()函數將日志文件加載到內存,并轉換為可操作的RDD數據集.3)調用字段檢查函數對數據字段完整性檢查,對字段完整的數據利用map()算子實現數據類型轉換.4)使用filter()算子對url字段數據過濾,去除與訪問請求無關的數據以及自動加載的靜態資源數據等.5)利用map()算子提取與分析目標相關的核心字段.6)調用設備解析算法模塊對agent字段進行解析,解析出客戶端的設備、操作系統、瀏覽類型版本等信息.7)使用sortByKey()算子按照IP、時間將訪問記錄排序.8)調用用戶識別函數對數據處理.9)基于固定時間間隔會話識別算法,劃分用戶會話{cuserID|(pid,time,url1,url2…)}.11)調用map()算子對數據格式進行歸一轉換.12)調用數據儲存模塊將數據儲存到HBase數據庫.

3.2 應用系統的安全分析與行為分析

油田日志的分析包括應用系統的安全性分析和行為分析.在安全分析方面,由于被攻擊的應用日志記錄中會包含了兩類請求:正常訪問請求和惡意攻擊請求,本文主要通過匹配記錄中的惡意訪問信息的特征來判斷應用系統是否被攻擊.在安全檢測方法上采用基于特征方式的檢查方法,該方法的實現主要借助于預先設計攻擊特征庫和基于RDD算子設計的函數模塊.其中攻擊特征庫是依據各類攻擊特征設計正則表達式,從而匹配出可能存在的攻擊類型[14].基于RDD算子設計的函數模塊主要在集群環境下通過Spark并發處理機制來提高日志安全分析檢索速率.

圖5 日志安全分析流程

基于RDD算子的安全分析算法主要步驟如算法2.

算法2.應用日志安全檢查算法1)利用map()算子提取相關字段進行單條數據分析.2)調用攻擊特征庫,通過正則表達式完成攻擊行為匹配,并確定疑似攻擊類型.3)利用sortByKey()算子重現攻擊者訪問行為軌跡.4)利用union()算子進行多條記錄關聯分析.5)提取post字段,利用filter()算子判斷可疑文件.6)記錄漏洞特征,推斷大致入侵流程并發出告警信息.

在油田應用行為分析方面,主要是在基于HBase的鍵值存儲模型上運行各類分析算法.由于油田在規劃內部網絡時會預留一部分IP地址作為應用服務地址,因此可以根據IP+端口(appID)來作為應用的唯一標識,根據分析指標需求設計HBase表存儲結構包括:一個唯一標識的行健(RowKey)和兩個信息列簇.其中行健由:應用ID、訪問時間、用戶IP三者的組合來標識;兩個列簇分別為用于描述用戶設備信息和請求訪問結果信息,其中每個列簇又包括多個列.應用行為分析存儲結構詳細描述如表1.

表1 HBase鍵值表結構

在應用日志的行為分析算法方面,主要基于Spark計算框架中的Spark SQL模塊設計完成,Spark SQL向用戶提供了在大數據集上的類SQL查詢功能,同時還支持將原有持久化儲存數據遷移到Spark環境下進行分析[15].Spark SQL的分析的核心模塊是DataFrame.DataFrame是一個以命名列方式組織的分布式數據集.它類似于關系型數據庫中的一張表.DataFrame可以由結構化數據、現存在的RDD或者從外部的關系數據庫導入并轉換而來[16].其中DataFrame包括:用于描述列字段的集合Schema和行數據集DataSet

根據油田管理評估要求需要統計的應用行為指標包括:應用每小時的訪問量、應用運行安全狀況、各模塊的使用量、應用模塊異常信息、使用次數用戶排名等27個行為指標.由于HBase根據rowkey來檢索數據并且支持以字符串匹配方式的掃描方法.因此將時間和IP作為查詢條件,可以在各類應用間進行用戶訪問行為的關聯分析,進而描繪出用戶每天在各類應用的停留時間和訪問軌跡并推斷出用戶訪問喜好.

本文在實現應用行為分析算法時,將這些應用統計指標封裝在一個算法內,因此執行一次算法就可統計出所有應用指標.在Spark執行行為分析時需要確定數據源和具體的分析算法:其中算法選取由調度引擎來完成并提交給Spark集群來;數據源來自于上一階段的數據預處理算法處理后儲存在HBase中的結構化數據,需要調度引擎將要分析數據的起始行鍵提交給Spark集群.Spark集群根據HBase起始行鍵拉取數據并執行指定算法程序,完成處理后返回處理結果.由于每個分析算法需要完成多個分析指標的統計,因此需要根據分析指標制作多個DataFrame數據集.因為在數據量過大時,制作DataFrame數據集非常耗時.因此制作數據集時要盡可能滿足多個查詢需求,以減少重復制作數據集的處理時耗.算法流程如下圖6所示,其中每一個分支流程對應一個分析指標.

每個行為指標具體的分析流程如下:首先選取相應的字段并對字段數據進行格式轉換、數據規約,該過程主要借助于Spark提供的算子函數完成;然后將RDD數據集轉換成DataSet

圖6 分析算法流程圖

圖7 行為指標分析流程圖

4 系統開發環境及實驗分析

4.1 系統平臺部署

系統平臺的主要由三個部分組成:數據收集層、數據分析層、Web業務層.依據實際生產場景,系統開發環境部署規劃如下,數據收集層由1臺日志儲存服務器組成,用于部署Flume日志收集框架.數據分析平臺是由1臺主機點和3臺計算節點組成計算集群,各節點分別搭建Hadoop服務集群、Spark服務集群、HBase儲存集群,并在主節點搭建調度引擎程序.業務層由一臺Web服務器組成,用于部署業務管理平臺和業務數據庫.系統具體部署規劃如圖8所示.

4.2 實驗結果分析

實驗分析聚焦在數據分析層上,主要統計各類算法的分析耗時,本文的實驗環境是由4臺節點組成的集群環境,日志文件儲存在HDFS上,基于Spark框架設計分析算法完成數據的分析,基于HBase儲存分析數據,并將Spark任務直接提交到Yarn上,由Yarn完成資源分配和Spark任務調度.其中各節點的環境信息和部署組件信息如表2所示.

圖8 系統部署圖

表2 Spark 集群運行環境

實驗分別在單節點環境和四節點組成的集群環境下測試了2個典型算法的耗時,測試的算法為:日志文本數據的預處理算法和應用行為指標分析算法A(該算法主要用于統計IIS類型應用日志的行為指標,包括統計每小時IP量、總UV量、每小時PV、總PV量、各模塊的訪問量、TOPN用戶等27個行為指標).日志預處理算法選取了某油田企業內部具有代表性的應用日志數據,日志數據格式為IIS W3C格式.選取并整理的單個日志數據大小依次為106 MB、511 MB、1.1 GB、5.1 GB、9.8 GB、20 GB.實驗對比結果如圖9所示.

圖9 預處理算法時長對比圖

由實驗結果可以看出,當數據量較小時,單節點的處理時長較短;當數據容量大于5 GB時,集群環境下的處理時長遠小于單節點的處理時長.

算法A的實驗數據為儲存在HBase中的結構化數據,分別選取的數據集分別為:956 887條、1975 511條、5911 511條、29 479 329條、63 906 591條數據.這里統計數據算法A的耗時為從數據加載到內存到預處理數據分析完成的時間(不包括將數據寫回數據庫中的時間),結果如表3.該算法的時間消耗主要在于:制作DataFrame數據集的耗時和運行查詢SQL的耗時,算法A完成27個指標的統計,需要制作9個DataFrame數據集,運行了35次SQL查詢.

表3 算法A處理時長對比

從實驗結果可以看出,當數據集增長到一定程度,采用集群環境的處理耗時遠低于單機處理耗時.

從兩個分析算法的耗時統計可以得出:當數據量大小在單節點處理能力范圍內,單節點處理時長要小于集群環境下處理時長;若數據量過大,采用集群環境的處理耗時要小.這是由于集群環境下涉及到數據的分片,任務間的通信,代碼序列化分發,如果數據儲存不在本地,還會涉及到數據的移動問題,此外處理時長還受主機磁盤IO傳輸速率、網絡帶寬的傳輸速率的影響,這些多方面的因素都會影響處理時長.因此集群環境在處理大批量數據時才會發揮優勢.

5 結論

面對油田應用部署分散、種類繁多、數量龐大的復雜場景.本文借助于各類主流的大數據處理框架實現對海量數據收集和儲存;在數據處理分析方面,本文基于Spark計算框架設計了應用日志行為分析系統,并設計了應用的安全狀況分析和行為指標分析的算法;此外為了方便運維人員使用該系統,又基于Web設計了可視化的管理平臺實現了各類框架的集成與管理.該系統解決了油田進行海量應用數據分析的滯后性難題;為油田迅速評估各類應用系統的運行狀況和安全狀況提供了決策依據;并為油田快捷高效的管理各類業務系統帶來了一系列巨大優勢.