基于Spark Streaming的實時日志處理平臺設(shè)計與實現(xiàn)*

薛瑞，朱曉民

（1 北京郵電大學網(wǎng)絡(luò)與交換技術(shù)國家重點實驗室，北京 100876；2 東信北郵信息技術(shù)有限公司，北京 100191）

1 引言

目前大多數(shù)日志分析工具都只能用于單機的數(shù)據(jù)分析處理，而面對海量化日志處理以及對分析結(jié)果低延時反饋的實際需求，傳統(tǒng)的日志處理方式已經(jīng)遠遠無法滿足人們的需要。

Hadoop是Apache軟件基金會（Apache Software Foundation）組織下的一個開源項目，提供分布式計算環(huán)境下的可靠、可擴展軟件，可以方便地在普通硬件上架設(shè)自己的大規(guī)模機群系統(tǒng)。由于Hadoop優(yōu)勢突出，基于Hadoop的應(yīng)用已經(jīng)被大量開發(fā)，尤其是在互聯(lián)網(wǎng)領(lǐng)域，它可以承擔PB級別的數(shù)據(jù)處理，其原生的Mapreduce編程模型也已經(jīng)成為主流的分布式編程模型。但是，對于在線/近線系統(tǒng)來說，處理在線系統(tǒng)產(chǎn)生的日志，需要低延時高可靠的處理海量數(shù)據(jù)，Mapreduce編程模型作為離線計算的代表，無法滿足日志實時監(jiān)控、統(tǒng)計分析的需求。

本文介紹了一種基于Spark Streaming的分布式大規(guī)模日志實時處理平臺的設(shè)計方案，其特色主要體現(xiàn)在海量日志存儲及處理和實時統(tǒng)計分析兩個方面。文章首先闡述了平臺的構(gòu)建方案，之后介紹了系統(tǒng)的各個模塊的功能，對企業(yè)海量實時日志處理平臺建設(shè)具有一定的指導意義。

2 設(shè)計目標

提供方便快捷的海量日志數(shù)據(jù)實時處理與統(tǒng)計服務(wù)，具體的設(shè)計目標如下：（1）提供海量日志數(shù)據(jù)的安全存儲、快速查詢服務(wù)；（2）提供實時或準實時的日志處理服務(wù)；（3）根據(jù)不同用戶的需求，允許用戶添加指定的日志源，并支持基本的數(shù)據(jù)統(tǒng)計與展示；（4）為了更好的用戶體驗，通過Web端實現(xiàn)對日志數(shù)據(jù)的管理和處理統(tǒng)計。

日志信息對運維人員和數(shù)據(jù)分析人員非常重要。通過日志信息，可以獲取軟件訪問數(shù)據(jù)、軟件系統(tǒng)內(nèi)部的核心鏈路和系統(tǒng)異常等。本文提供了一個日志處理平臺設(shè)計方案，既適用于線上服務(wù)的監(jiān)控，也適用于有效信息的挖掘，具有一定的實際意義。

3 系統(tǒng)設(shè)計

3.1 系統(tǒng)設(shè)計圖

系統(tǒng)設(shè)計如圖1所示，圖中，箭頭的方向代表數(shù)據(jù)的流向：Kafka作為消息隊列服務(wù)器，不斷地接收Web服務(wù)器Push的日志數(shù)據(jù)；同時，數(shù)據(jù)處理層和數(shù)據(jù)資源層不斷從日志數(shù)據(jù)獲取層拉取日志信息；數(shù)據(jù)處理后的結(jié)果存入HBase數(shù)據(jù)庫，業(yè)務(wù)邏輯層通過操作HBase數(shù)據(jù)庫獲取所需信息，而顯示層與邏輯層之間通過API調(diào)用進行通信。

3.1.1 日志數(shù)據(jù)獲取層

日志數(shù)據(jù)獲取層的核心是消息隊列系統(tǒng)。消息隊列系統(tǒng)是消息隊列系統(tǒng)通過Kafka實現(xiàn)。Kafka是一個分布式高吞吐量的消息隊列系統(tǒng)，Kafka的集群有多個服務(wù)器組成，每個類型的消息被定義為Topic，同一Topic內(nèi)部的消息按照一定的Key和算法被分區(qū)存儲在不同的服務(wù)器上，消息生產(chǎn)者和消費者可以在多個服務(wù)器上生產(chǎn)/消費。

本系統(tǒng)可以包含多種日志數(shù)據(jù)源，如線上系統(tǒng)產(chǎn)生的應(yīng)用日志、操作系統(tǒng)、安全系統(tǒng)日志等。本系統(tǒng)允許用戶通過Web端設(shè)置日志數(shù)據(jù)路徑的方式定義日志數(shù)據(jù)源。日志數(shù)據(jù)作為消息發(fā)布訂閱系統(tǒng)的生產(chǎn)者，發(fā)布消息時，Kafka客戶端先構(gòu)造一條消息，將消息加入到消息集合中（Kafka支持批量發(fā)布，可以往消息集合中添加多條消息，一次性發(fā)布），發(fā)送消息時，客戶端需指定消息所屬的消息Topic。

圖1 系統(tǒng)設(shè)計圖

消費者可以通過如下CreateStream方法從Kafka獲取消息流：

defcreateStream（ssc: StreamingContext, zookeeperQurum: String,

groupId: String,topic: Map[String, Int]) : ReceiverInputDStream[（String, String)]

本系統(tǒng)中，數(shù)據(jù)處理層的計算模塊作為消費者不斷從Kafka服務(wù)器拉取消息；同時，數(shù)據(jù)經(jīng)過一定的過濾策略存儲到HBase中。

3.1.2 數(shù)據(jù)資源層

考慮到日志數(shù)據(jù)的快速增長以及快速查詢的需求，采用HBase[1]作為存儲。HBase是Hadoop的數(shù)據(jù)庫，位于結(jié)構(gòu)化存儲層，Hadoop HDFS為HBase提供了高可靠性的底層存儲支持，能夠支持海量數(shù)據(jù)的分布式實時存儲；同時，HBase會對存儲的數(shù)據(jù)自動切分，并支持高并發(fā)讀寫操作，使海量數(shù)據(jù)存儲具有更強的擴展性；并且，HBase是面向列的KeyValue數(shù)據(jù)庫，可以存儲異構(gòu)數(shù)據(jù)并且方便范圍查詢。這些特性剛好解決了日志信息存儲模塊存在的問題。

在本系統(tǒng)中，同一日志源信息將會存儲在一張HBase表中。表的Rowkey設(shè)計為（Long.MaxValue（)-requestTime) +requestPath，通過（Long.MaxValue（)-requestTime）可以將最新的日志放在最前面，便于查看，查詢某個時間段的記錄時，只要設(shè)置scan.setStartRow （beginTime)和scan.setStopRow（endTime)，大大提高了scan的效率[2]。另外，這樣設(shè)計可以保證批量獲取的日志信息集中在同一個Region下，在遍歷結(jié)果時，可以獲得更好的性能。同時，HBase數(shù)據(jù)庫也保存數(shù)據(jù)處理的結(jié)果信息，以供查詢統(tǒng)計。

3.1.3 數(shù)據(jù)處理層

數(shù)據(jù)處理層是本系統(tǒng)的核心。數(shù)據(jù)處理層需要從日志獲取層拉取信息，并實時處理。本系統(tǒng)的數(shù)據(jù)處理功能基于Spark Streaming實現(xiàn)[3,4]。Spark Streaming是建立在Spark上的實時計算框架，它將流式計算分解成一系列短小的批處理作業(yè)，交由Spark引擎處理，每一段數(shù)據(jù)都轉(zhuǎn)換成Spark中的RDD，然后將Spark Streaming中對DStream的操作變?yōu)獒槍park中對RDD的Transformation操作[5]，將RDD經(jīng)過操作變成中間結(jié)果保存在內(nèi)存中。整個流式計算根據(jù)業(yè)務(wù)的需求可以對中間的結(jié)果進行疊加，或者直接存儲到HBase。

Spark Streaming的基本原理是將輸入數(shù)據(jù)流以時間片（秒級）為單位進行拆分，然后以類似批處理的方式處理每個時間片數(shù)據(jù)，其基本原理如圖2所示。

圖2 Spark Streaming基本原理圖

首先，Spark Streaming把實時輸入的日志數(shù)據(jù)流以時間片Δt （如1s）為單位切分成塊。Spark Streaming會把每塊數(shù)據(jù)作為一個RDD，并使用RDD操作處理每一小塊數(shù)據(jù)。每個塊都會生成一個Spark Job進行處理，最終結(jié)果匯總后存入HBase。

3.1.4 業(yè)務(wù)邏輯層

業(yè)務(wù)邏輯層處于數(shù)據(jù)資源層與顯示層中間，起到了數(shù)據(jù)交換中承上啟下的作用。本系統(tǒng)中，消息源配置、日志查找、資源配置、報表顯示等業(yè)務(wù)邏輯的實現(xiàn)均在業(yè)務(wù)邏輯層完成。同時，業(yè)務(wù)邏輯層還實現(xiàn)了系統(tǒng)的用戶管理和用戶權(quán)限體系邏輯：不同用戶對不同消息源的訪問權(quán)限、進行配置是否有視圖權(quán)限、編輯權(quán)限等。

3.1.5 顯示層

顯示層位于最外層，為用戶提供Web交互式操作的界面，用于顯示數(shù)據(jù)和接收用戶輸入的數(shù)據(jù)。本系統(tǒng)中，用戶可以通過Web界面添加日志源、查詢篩選日志信息、查詢?nèi)罩緢蟊淼龋芾韱T也可以通過Web端對不同用戶的操作權(quán)限進行設(shè)定。

3.2 平臺核心功能模塊分析

3.2.1 日志查詢模塊

日志查詢模塊可以根據(jù)關(guān)鍵字和日志產(chǎn)生的時間進行查詢，底層通過HBase的SQL實現(xiàn)。根據(jù)關(guān)鍵字查詢是指基于特定的語法查詢，依據(jù)條件篩選符合條件的字段類型或者字段內(nèi)容的日志信息，適合查詢是否存在業(yè)務(wù)異常、程序執(zhí)行的關(guān)鍵邏輯、IO操作等；根據(jù)時間查詢包括相對時間查詢（相對于當前時間點之前5 min、15 min、30 min、1 h、2 h、3 h、1天、1個 月、1年和所有記錄）和絕對時間查詢（定義篩選的開始時間和結(jié)束時間，格式為：yyyy-MM-ddHH:mm:ss，查詢該時間段內(nèi)的所有日志記錄）。同時，也可以支持將兩種查詢方式結(jié)合的聯(lián)合查詢。

3.2.2 日志統(tǒng)計分析模塊

數(shù)據(jù)處理層提供了一些基本日志數(shù)據(jù)統(tǒng)計的接口。目前， 基本的統(tǒng)計功能包括：總數(shù)、最大值、最小值、均值、百分位和TopN。比如，提供了計算WebServer日志中某一狀態(tài)碼數(shù)量的getStatusCode方法：

其中Option[AccessLogRecord]是分析器的返回值[6]。調(diào)用log.filter（line≥getStatusCode（p.parseRecord （line)) == "404").count將返回狀態(tài)碼是404的行數(shù)。

3.2.3 日志報表視圖（Dashboard）模塊

Dashboard是將各個顯示的報表以組件的形式加入到顯示儀表盤中。創(chuàng)建Dashboard輸入標題、描述信息即創(chuàng)建完成。一個Dashboard可以添加多個組件，DashBorad創(chuàng)建列表中查看當前Dashboard關(guān)聯(lián)組件的總和、創(chuàng)建人、創(chuàng)建時間。實現(xiàn)編輯、刪除等操作。

視圖模塊可以將查詢的日志信息結(jié)果默認以報表柱狀圖的方式查看。目前支持以年、月、周、天、小時、分鐘6種時間維度查看報表。報表可以顯示當前時間維度日志信息的總數(shù)，同時可根據(jù)該日志流中選擇字段（支持所有字段、自選字段、默認字段）的方式查看相應(yīng)的統(tǒng)計值。報表支持多種查看類型，包括餅圖、條形圖折線圖、散點圖；支持年、月、周、天、小時、分鐘等視圖方式。

4 結(jié)束語

目前，海量日志數(shù)據(jù)的實時處理成為越來越普遍的需求，但是目前的解決方案均無法同時完美解決海量、實時這兩個需求。本文介紹海量日志實時處理方案，基于Hadoop yarn平臺，具有低成本、高效率、易于擴展的優(yōu)點，通過Spark Streaming流數(shù)據(jù)處理技術(shù)實時處理日志信息，同時，使用HBase作為數(shù)據(jù)存儲數(shù)據(jù)庫，很好地滿足了海量數(shù)據(jù)存儲、快速查詢的需求。為用戶提供Web端的可視化界面，更方便用戶進行統(tǒng)計查詢等操作。然而，本系統(tǒng)中提供的日志計算處理的方式比較單一，只支持部分基本信息的統(tǒng)計分析，比如總數(shù)、最小值、最大值、均值、百分位、TopN等，暫不支持復雜的日志信息統(tǒng)計，未來的工作中會對此進行改進。

[1]卓海藝.基于HBase的海量數(shù)據(jù)實時查詢系統(tǒng)設(shè)計與實現(xiàn)[J].北京郵電大學, 2013（2):4.

[2]張榆, 馬友忠, 孟小峰.一種基于HBase的高效空間關(guān)鍵字查詢策略[J].小型微型計算機系統(tǒng), 2012,33（10):2141-2146.

[3]Liu X, Iftikhar N, Xie X.Survey of real-time processing systems for big data[C]//Proceedings of the 18th International Database Engineering & Applications Symposium.ACM, 2014.

[4]Hunter T, Das T, Zaharia M, et al.Large-scale online expectation maximization with spark streaming[J].2012.

[5]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.USENIX Association, 2012.

[6]Deng L.Building a Big Data Analytics Service Framework for Mobile Advertising and Marketing[J].