基于SQL-on-Hadoop查詢引擎的日志挖掘及其應用

何明，常盟盟，劉郭洋，顧程祥，彭繼克

(1.北京工業大學 信息學部,北京 100124； 2. 海通證券股份有限公司 信息技術管理部，上海 200001)

基于SQL-on-Hadoop查詢引擎的日志挖掘及其應用

何明1，常盟盟1，劉郭洋2，顧程祥2，彭繼克2

(1.北京工業大學 信息學部,北京 100124； 2. 海通證券股份有限公司 信息技術管理部，上海 200001)

隨著計算機和網絡技術的迅猛發展以及數據獲取手段的不斷豐富，海量數據的實時處理需求日益增多，傳統的日志分析技術在處理海量數據時存在計算瓶頸。大數據時代下，隨著開放式處理平臺的發展，能夠處理大規模且多樣化數據的大數據處理系統應運而生。為了讓原有的業務能夠充分利用Hadoop的優勢，本文首先研究了基于大數據技術的網絡日志分析方法，構建了網絡日志分析平臺以實現萬億級日志采集、解析、存儲和高效、靈活的查詢與計算。對比分析了Hive、Impala和Spark SQL這3種具有代表性的SQL-on-Hadoop查詢系統實例，并展示了這類系統的性能特點。采用TPC-H測試基準對它們的決策支持能力進行測試及評估，通過對實驗數據的分析和解釋得到了若干有益的結論。實現了海量日志數據計算與分析在證券領域的幾種典型應用，為進一步的研究工作奠定了基礎。

大數據；日志分析；數據挖掘；Hadoop；查詢引擎；數據采集；索引存儲；證券行業

隨著互聯網的飛速發展和逐層推進，企業內部的規模和業務量也不斷增加，致使數據量猛增。企業網絡中的計算機設備和網絡組件持久地記錄著海量的網絡日志。日志文件是系統軟硬件信息和用戶行為信息記錄的載體，通過日志分析能夠實時獲取設備、網絡運行狀態和用戶行為交易等信息，有利于保證系統的穩定運行和來往業務的安全性。目前，較為成熟的日志集中管理系統解決了各類設備、服務器和應用日志的采集與格式統一問題，日志分析也從最初簡單的正則匹配向結構化查詢、報表和預測演進[1]。越來越多的行業領域面臨海量(volume)、高速(velocity)和多樣(variety)等多V挑戰，大數據時代已真正到來[2-4]。

互聯網中海量的信息為證券領域日志分析提供了豐富的數據支撐，如何利用大數據分析技術進行實時準確的日志分析成為重要的科學問題。在大型證券公司的內部網絡中，隨著網絡帶寬的迅速擴容日志量急劇增長且日志源眾多，包括網上交易日志、移動證券日志和網站日志等主要系統的日志。以海通證券為例，目前在全國設有幾十個節點，幾百臺服務器，峰值在線用戶約幾十萬，每個節點各部署了1臺負載均衡設備。網上交易應用服務器全天24小時將客戶請求數據與應答數據實時或小批量定時寫入磁盤日志文件，每臺交易應用服務器的日志文件大小為100 MB～3 GB，總計在100 GB左右。同時，每臺網上交易應用服務器還會生成一份發送給柜臺程序的網關日志數據。此外，各節點負載均衡設備的日志采用SNMP協議進行采集，采集每個站點的網絡流量、用戶連接數據。每日合計有3億多條日志，總量共計約300 GB。僅上述3類日志存儲一年就將產生約108 TB數據，若接入更多設備、操作系統、業務平臺日志，數據規模則更大。傳統的日志處理方法在面對海量大數據時，其存儲方式和計算能力都受到了限制，因此分布式存儲和并行計算成為了新的發展趨勢。如何采集、傳輸、存儲、分析及應用大規模的日志數據，已成為證券行業在大數據時代下面臨的重大挑戰。

Hadoop[5]分布式處理平臺為大數據存儲和分析提供了有效的解決方案。在大數據應用方面，雖然學術界和工業界對大數據的關注各有側重，但有一個共同的認識：大數據只有和具體的行業深入結合才能落到實處，才能產生真正的價值。通過前期的積累和算法的升級，大數據應用將對證券行業產生革命性影響。

本文的主要貢獻如下：

1)研究基于SQL-on-Hadoop查詢系統的性能特點，對比分析了Hive、Impala和Spark SQL這3種具有代表性的SQL-on-Hadoop查詢系統實例，構建了海量日志采集與實時計算分析平臺；

2)采用TPC-H測試基準對它們的決策支持能力進行測試及評估，通過對實驗數據的分析和解釋得到了若干有益的結論；

3)實現了大規模網絡日志數據分析與計算在證券領域的幾種典型應用。

1 相關工作

大數據技術在互聯網領域海量網絡日志分析和處理過程中得到了廣泛的應用，日志分析系統主要包括日志同步、數據存儲、分布式計算和數據倉庫等相關技術。開源的日志分析系統如Facebook的Scribe[6]，Apache的Chukwa[7]，LinkedIn的Kafka[8]，Cloudera的Flume[9]等。Facebook公司龐大的用戶群體產生了大量的信息與社交數據，現有8億多用戶的信息需要處理，產生了大規模的數據和日志；同時，離線的大規模數據分析計算已無法滿足實時數據分析的用戶需求， Scribe結合了Google的分布式文件系統GFS[10](google file system，GFS)。操作流程是收集異構數據源上的日志，集中存儲到分布式文件系統，從而在此基礎上進行統計分析。Amazon基于S3和EC2，開發了Amazon EMR來提供大數據處理服務，可以將數據分布在可重新調整大小的EC2集群中進行處理，包括日志分析、索引、數據倉庫和機器學習等。阿里巴巴集團使用目前國內最大的Hadoop集群“云梯”進行各部門產品的線上數據備份、系統日志以及爬蟲數據分析，并建設開放平臺為個人和企業提供各種增值服務。騰訊微信等應用產品擁有上億級別的用戶，產生了海量的個人用戶日志數據，這些數據中蘊藏著巨大的商業價值，并提出“大數據營銷”的概念。人人網基于Hadoop的Hive[11]、HBase[12]和Streaming[13]組件，構建了SNS推薦平臺進行分析計算、內容推薦等工作。百度的高性能計算系統規劃中的架構將有超過1萬個節點，每天的數據生成量在10 PB以上，主要用于日志的存儲分析以及統計挖掘等功能。Wei等設計了Analysis Farm摒棄了傳統的關系型數據庫(relational database management system，RDBMS)，利用NoSQL(not only SQL)數據庫MongoDB構建了可橫向擴展的日志分析平臺，以支撐NetFlow日志存儲和查詢[14]。Rabkin等設計了基于Hadoop的日志收集和分析系統Chukwa，日志處理程序在MapReduce框架上開發[15]。文獻[16-17]從原位分析的角度出發，分別實現了針對大規模日志分析的MapReduce(In-situ MapReduce)和Continuous處理機制， 但MapReduce模型計算代價很大，并不能很好地支持迭代運算。

然而HDFS[18]和MapReduce[19]大數據處理架構主要是針對靜態數據的批處理，在運算過程中產生的大量I/O操作無法保證處理過程的實時性。針對上述問題，本文將研究基于SQL-on-Hadoop查詢引擎構建網絡日志分析平臺，通過使用廣泛的標準SQL語言來實現快速、靈活的查詢性能。通過利用TB級日志數據對存儲、查詢性能進行測試、優化和比較，構建具有穩定性、高性能、可擴展性、易用性和安全性的網絡日志統一采集查詢和監控平臺，以滿足對TB或PB級容量和萬億日志管理的應用需求，為面向證券行業的日志大數據分析及其應用提供技術支撐。

2 基于Hadoop的結構化數據處理

網絡日志源的種類具有多樣性的特點，包括結構化、半結構化和非結構化的數據。不同類型的日志存儲方式有所不同。日志管理系統的采集器對不同格式的日志進行標準化處理，從而以結構化的形式進行日志存儲和分析。本文所采用的源數據主要分為文本數據、數據庫數據和實時/準實時數據等。

2.1 HDFS數據采集

網絡日志的生成是分布式的，與傳統的日志管理系統一樣，日志采集是本文平臺的基礎。本文平臺采集的日志直接存儲在Hadoop文件系統(HDFS)中，由于平臺構建于Hadoop之上，能夠處理海量分布式存儲的日志數據，同時易于水平擴展，本文的日志數據基本流程按功能可劃分為5層，如圖1所示。

1)原始數據層：業務上完成日志格式梳理，系統運行日志支持實時訪問和采集接口。

2)數據采集層：主要負責通用的日志數據解析、高效采集和安全可控。

3)數據處理層：主要包括對日志數據的批量式處理和實時處理。

4)數據服務層：主要提供標準的數據訪問接口ODBC、JDBC、HIVE等。

5)數據展示層：實現實時監控類和報表類數據的展示。

圖1 日志數據處理基本流程Fig.1 Basic log data processing framework

根據應用需求，本文日志的采集方式分為以下3種。

1)文件導入：對已分布在個服務器磁盤的日志文件，經網絡文件系統掛載，直接將日志文件導入HDFS。該方式允許日志文件批量可靠導入，可在網絡利用率低谷時段進行傳送。

2)流數據導入：基于Apache Flume[20]構建，實現多個日志源數據實時匯聚，接收網上交易應用服務器和網絡設備發送的日志。

3)RDBMS導入：為實現與現有日志系統兼容，基于Apache Sqoop[21]，實現與Oracle、MySQL和PostgreSQL等RDBMS對接，支持直接導入存儲在上述數據庫中的數據記錄。Sqoop同時可以將SQL-on-Hadoop處理結果輸出到RDBMS，供現有的日志分析系統進行報表及可視化處理。

2.2 SQL-on-Hadoop查詢引擎

SQL是結構化數據的查詢語言，SQL-on-Hadoop是構建在Hadoop之上的SQL查詢系統，利用Hadoop能夠進行海量數據(TB級別以上)的處理。目前已有的SQL-on-Hadoop系統大致可以分為兩大類：第一類將SQL查詢轉換為Map-Reduce job；第二類系統基于MPP(massively parallel processing)的設計方式，僅僅使用Hadoop作為存儲引擎，上層自行實現分布式查詢的邏輯。第一類系統的代表是Facebook的Hive。Hive是原始的SQL-on-Hadoop解決方案。它是一個開源的Java項目，能夠將SQL轉換成一系列可以在標準的Hadoop TaskTrackers上運行的MapReduce任務。如圖2中的Hive架構部分所示，Hive通過一個metastore(本身就是一個數據庫)存儲表模式、分區和位置以期提供像MySQL一樣的功能。它支持大部分MySQL語法，同時使用相似的 database/table/view約定組織數據集。Hive內部機制是基于MapReduce，從而導致了計算過程中消耗大量的I/O，降低了運行效率。Impala[22]是由Cloudera構建的一個針對Hadoop的開源的MPP(massively parallel processing)“交互式”SQL查詢引擎。Impala同樣提供了一種SQL查詢方法，如圖2中的Impala架構部分所示，與Hive不同的是，Impala并沒有使用MapReduce執行查詢，而是使用了自己的執行守護進程操作本地磁盤文件。由于沒有MapReduce開銷以及磁盤I/O、查詢語句編譯等一系列優化，Impala通常要比Hive具有更快的數據訪問性能[23]。Impala共享Hive的metastore，可直接與Hive管理的數據互操作。Spark[24]使用輕量級的線程作為執行器，減少了執行作業的開銷，同時提高了調度的響應速度，如圖2中的Spark部分所示。Spark SQL是在Spark之上搭建的SQL查詢引擎，支持在Spark中使用Sql、HiveSql、Scala中的關系型查詢表達式。

圖2 Hadoop、Hive、Impala與Spark執行結構圖Fig.2 Structure for implementation of Hadoop, Hive, Impala and Spark

2.3 結構化數據存儲與壓縮

目前，很多研究者提出了在Hadoop中優化結構化數據存儲的方法。He等[25]提出的RCFile格式旨在提高數據導入和處理效率。它首先將數據水平分割為多個行組(row-group)，然后對每個組內的數據垂直分割成列存儲。列存儲將數據表同一列的數據連續存放，當查詢只涉及部分列時，可大幅減少所需讀取的數據量。ORC(optimized RCFile)是對RCFile的改進，解決其在數據類型和性能上的多個局限性，改善查詢和空間利用效率。Parquet是Hadoop生態圈中一種新型列式存儲格式，靈感來自于2010年Google發表的Dremel論文[26]，它可以兼容Hadoop生態圈中大多數生態框架(Hadoop、Spark等)，被多種查詢引擎支持(Hive、Impala、Spark SQL、Drill等)，并且它與語言和平臺無關的。表1比較了本文2.2節描述的3種查詢引擎從HDFS上讀取多種格式的數據格式的支持。Text是原始的文本數據，通常為CSV或其他特定字符分隔。Hive的格式支持更為全面，由于Impala和Hive共享metastore，因此本文平臺實際應用中通常由Hive導入數據而后臺使用Spark SQL查詢。

表1Hive、Impala和SparkSQL數據格式支持比較

Table1DataformatcomparisonofHive,ImpalaandSparkSQL

數據格式HiveImpalaSparkSQL查詢插入查詢插入查詢插入Text√√√√√√RCFile√√√———ORC√√————Parquet√√√√√√

數據壓縮是另一種性能優化方法。壓縮一方面節省存儲空間，另一方面在相同磁盤I/O速度可讀寫更多記錄。Hive、Impala和Spark SQL均支持直接查詢壓縮的數據文件，常用壓縮算法有Gzip/Zlib和側重于解壓縮速度的Snappy。ORC格式本身已內嵌輕量級的壓縮機制。

2.4 結構化數據處理算法

RDD數據集包含對父RDD的一組依賴，這種依賴描述了RDD之間的傳承關系。RDD將操作分為兩類：Transformation與Action。Transformation操作不執行運算，只有當Action操作時才觸發運算。在RDD的實現機制中，基于迭代器的接口實現原理使得數據的訪問更加高效，同時避免了大量中間結果對內存的消耗。Spark SQL包含了結構化數據和數據之上進行運算的更多信息，Spark SQL使用這些信息進行優化，使得結構化數據的操作更加高效和方便，基于Spark SQL的數據操作流程如下。

算法1 SparkSQLonRdd(lt;inputgt;,lt;contextgt;)

輸入Kafka輸入數據流input，Spark上下文context;

輸出分布式集合dataframe。

1)DStream line:Kafka-gt;DStream(input)；

2)獲取Kafka流數據輸入；

3)SqlContext sc = new SqlContext(context)；

4)DStreamlt;Rowgt; rdd=line.map；

5)new Function；

6)public Row call(T) {}；

7)創建Row對象；

8)Listlt;StructFieldgt; sf= new；Listlt;StructFieldgt;()；

9)Struct Fields.add(CreateDataType(lt;Columngt;))；

10)重復步驟9)創建邏輯表結構；

11)Struct Type st: DataTypes.CreateStructType(sf)；

12)DataFrame df :

13)sc-gt;DataFrame(rdd, st)；

14)df.RegisterTable(lt;Table Namegt;)；

15)DataFrame dataframe=sc.sql(lt;Sql Querygt;)；

16)Return dataframe。

算法2 RddProcessing(lt;inputgt;)

輸入Kafka輸入數據流input；

輸出數據集對象record。

1)數據采集與預處理

①SparkConf conf = new SparkConf()；

②創建上下文對象；

③StreamingContext(conf, Interval)；

④Maplt;E,Tgt; Offsets=kafka.getOffset()；

⑤獲取kafka讀取偏移量；

⑥DStream stream；

⑦KafkaUtils.createDStream(input)；

⑧Return stream。

2)RDD數據處理

①stream.foreachRDD；

②new VoidFunctionlt;RDDgt;gt;()；

③call(RDDlt;MessageAndMetadatagt; rdd)；

④HasOffsetRanges offrange = rdd.rdd()；

⑤合并請求應答，并解析存儲數據；

⑥rdd.mapPartitionsToPair；

⑦ new FlumeKafkaFunction()；

⑧foreachPartition(ProceFunction())；

⑨kafka.setOffset(offrange)；

⑩保存kafka讀取偏移量。

3)ProceFunction數據后處理

①Iteratorlt;Tuple2lt;T, KafkaDatagt;gt; iter；

②while (iter.hasNext())；

③KafkaData data = iter.next()._2()；

④json = data.getData()；

⑤Record record =Object(json, class)；

⑥record.setCollect_time；

⑦data.getExtData(TIME))；

⑧Utils.save(item_topic, record)；

⑨Return record。

其中，RDD根據數據記錄的key對結構進行分區。分片數據采用迭代器Iterator流式訪問，hasNext方法是由RDD lineage上各個Transformation攜帶的閉包函數復合而成,使得對象被序列化，通過網絡傳輸到其他節點上進行裝載運算。Iterator每訪問一個元素，就對該元素應用相應的復合函數，得到的結果再流式地存儲。

3 平臺架構與集群環境部署

3.1 平臺架構與處理框架

本文基于Hadoop，構建證券交易應用服務器和網絡設備海量日志采集、解析、存儲與實時計算分析平臺，平臺的核心架構如下。

1)數據采集層：負責實時采集來自通達信、恒生、核新的網上交易應用服務器全天24小時的客戶請求應答數據以及網絡設備日志數據，為大數據分析平臺提供數據源。

2)數據匯集層：將各個數據采集節點的日志數據源源不斷地匯集到各自的集群。

3)數據緩沖層：根據不同的Topic對海量日志數據進行緩沖，有助于控制和優化數據流經過系統的速度。

4)數據分發與解析處理層：負責數據的解析、勾對、計算和分發。

5)數據存儲與計算層：用于存儲、管理日志數據，支持多維檢索、統計分析和查詢處理。

6)應用層：負責面向終端用戶提供日志分析與管理的泛在接入，提供實時運維監控、實時預警、明細毫秒級查詢以及實時報表輸出等應用。

可以看到，在這個大數據分析體系結構中，系統支持TB級、PB級或者更大規模數據的分析和處理；系統可以處理結構化數據、非結構化和半結構化數據，有良好的擴展性。基于上述平臺結構，本文設計了能夠有效地利用大數據技術解決海量系統訪問日志多條件實時快速查詢的處理框架，如圖3所示。

圖3 處理框架Fig.3 Processing framework

該處理框架能夠保證平臺系統如下的幾個特性。

1)實時性：實時采集Agent包，從產生時刻起到實時采集，再到傳輸到數據中心，整個時間間隔控制在1 s內實時勾對、解析等計算，并保存到數據中心的集群，這個過程的時間間隔控制在3～5 s。

2)準確性和完整性：傳輸通道實現不重傳、不漏傳、斷點續傳，保證數據完整性。

3)安全性：非對稱加密算法對傳輸的日志數據進行加密，使用SSL/TLS協議，保障網絡傳輸通道的安全性。

4)穩定性和可靠性：基于成熟的、經過實踐驗證穩定可靠的Hadoop技術組件服務器節點非常容易實現橫向擴展，分布式環境保障集群中的任意一臺服務器出現宕機時不影響系統的穩定可靠運行。

3.2 環境部署

基于Hadoop的網絡日志分析平臺在海通證券網絡信息中心的搭建部署，如圖4所示。共42臺服務器，其中11臺是Flume匯聚節點(256 GB內存，2×600 GB，RAID1陣列)，5臺Kafka節點(256 GB內存，2×600 GB，RAID1陣列)，3臺Couchbase節點(512 GB內存，2×600 GB，RAID1陣列)，5臺Zookeeper節點(256 GB內存，2×600 GB，RAID1陣列)，2臺作為Namenode(256 GB內存，2×600 GB，RAID1陣列)，14臺是Datanode節點(256 GB內存，2×600 GB，RAID1陣列，2×600 GB，RAID1陣列 +6×2 TB，RAID0陣列)，2臺Tomcat(256 GB內存，2×600 GB，RAID1陣列)。

圖4 集群拓撲圖Fig.4 Cluster topology

所有節點通過10 GB以太網互聯。Hadoop部署采用Cloudera的發行版，版本為CDH5.5.0，HDFS總容量近60 TB。接入日志分析平臺的數據來自網上交易應用服務器日志數據和網絡設備日志數據。網上交易日志每天產生的記錄數約1.2億條，體積約100 GB；網絡設備日志數據日志每天的記錄數約650萬條，體積約6 GB。

4 實驗與性能評估

4.1 實驗環境與數據集

我們采用的實驗環境為7臺物理測試機構建的集群，選取2臺機器作為主節點，其余作為計算節點進行SQL-on-Hadoop實驗，測試集群拓撲如圖5所示。

圖5 測試環境拓撲圖Fig.5 Test environment topology

實驗采用針對OLAP應用的TPC-H測試基準來評估執行引擎的性能。TPC-H面向商務采購應用，其數據庫模式遵循第三范式。性能評測基準定義了22個復雜SELECT語句和2個更新數據語句，遵循SQL-92標準。數據庫的規模由自帶的擴展因子(scale factor，SF)決定，有10個級別，從1 GB到100 TB不等供用戶選擇。TPC-H 基準以每小時內執行的查詢數作為度量標準，在工業和科研領域當中應用廣泛。

文獻[23]討論了ORCFile和Parquet兩種列式存儲格式的性能差別，通過空間使用和查詢性能比較，認為Parquet針對文本文件壓縮率較高，從而節省了HDFS的存儲空間，同時減小了磁盤I/O的開銷，但是解壓縮會占用部分計算資源，對性能有一定影響。因此，本文采用Parquet 緊湊的列存儲格式，并選用了壓縮比和解壓速度較為均衡的Snappy，相對原始文本日志節省了近70%的空間。

本文實驗使用TPC-H 作為測試數據集，在SF=300的數據規模上進行測試，其描述和相關壓縮處理如表2所示。

表2 實驗數據集

4.2 性能評估

本文選擇SQL-on-Hadoop作為基礎查詢引擎，對3個引擎的處理時效性進行分析。從表3中各個引擎的總運行時間可以看出，Impala比Hive快了1.5倍，Spark SQL 比Hive快了2.7倍。

表3 查詢執行時間比較

實驗結果表明，Impala在Q1、Q6、Q12、Q15上的性能優于Hive，查詢語句結構如下：

SELECT

{

field1, field12,

SUM(field3) as alias1,

SUM(field4) as alias2,

……,

AVG(field5) as alias3,

……，

COUNT(*) as alias4

}

FROM TableExpression

WHERE

[field6 lt;= date′yyyy-mm-dd′-interval ′[DELTA]′ day (3)]

GROUP BY [field6, field7]

ORDER BY [field6, field7]

根據該語句的執行計劃，可以判斷查詢時對整個表進行了遍歷。對于Spark SQL而言，其在大多數查詢上的表現優于Hive和Impala。由于Spark的接口豐富和SQL優勢，在執行查詢時的速度較快。

4.3 Q22資源消耗情況

Q22的查詢語句如下：

SELECT

cntrycode, COUNT(*) as numcust, sum(c_acctbal) as totacctbal

FROM (

SELECT substring(c_phone from 1 for 2) as cntrycode,

c_acctbal FROM customer

WHERE substring(c_phone from 1 for 2) in

(′[I1]′,′[I2]′,′[I3]′,′[I4]′,′[I5]′,′[16]′,′[I7]′)

and c_acctbal gt; (

SELECT AVG (c_acctbal)

FROM customer WHERE c_acctbal gt; 0.00

and substring (c_phone from 1 for 2) in

(′[1]′,′[12]′,′[13]′,′[14]′,′[15]′,′[16]′,′[I7]′))

and not exists (SELECT * FROM orders where o_custkey=c_custkey)

) as custsale

Group BY cntrycode ORDER BY cntrycode;

如圖6所示，Q22中作業由3個子查詢組成。子查詢S1對customer表進行掃描并將結果保存到臨時表Temp1中；子查詢S2對Temp1進行聚集操作AGG1后將結果保存到臨時表Temp2中；子查詢S3在與表Orders執行聚集操作AGG2后依次與Temp1和Temp2進行關聯操作求笛卡爾乘積AGG3然后排序。

圖6 Implementation of Q22Fig.6 Q22的執行過程

實驗分析對比了不同的查詢方式在運行Q22時集群資源使用情況(如圖7～11所示)，包括CPU、內存、網絡、磁盤I/O。注意到，在查詢Q22執行過程中，Impala對集群資源的占用是最少的，其次是Hive，Spark SQL占用資源最多。由于Spark SQL是基于內存計算的框架，所以在內存占用方面和磁盤讀取上更為明顯。

圖7 集群平均CPU使用率Fig.7 Average cluster CPU usage

由于Hive和Spark SQL均在JVM之上運行，對CPU 和內存的使用依賴于JVM。如圖7所示，Impala的CPU占用時間要明顯少于Hive和Spark SQL，這是由于Impala在執行查詢過程中，在每個計算節點上運行只占用一個CPU線程。而Hive和Spark SQL在CPU使用上的優化完全依賴于JVM。如圖8所示，Impala和Hive內存使用率明顯小于Spark SQL，同時使用線程來執行耗費資源較多的Executor Backend進程。

圖8 集群內存平均使用量Fig.8 Average cluster memory usage

在磁盤性能方面，Impala和Hive的磁盤讀取速率優于Spark SQL。從圖9可以看出，Hive和Impala數據訪問量在S1時相對一致，高于SparkSQL。在S2中，Impala數據訪問量較小，Hive次之，SparkSQL最高。在圖10中，Impala在S1和S2執行結束后將結果寫入HDFS時，對磁盤的寫入速率迅速增加。

圖9 集群磁盤讀取總速率Fig.9 Cluster disk read speed

圖10 集群磁盤寫入總速率Fig.10 Cluster disk write speed

在圖11中，Impala在最后一個階段的網絡流量迅速增長，主要由于執行過程中的內部表連接產生的結果通過網絡傳輸給其他節點導致。

圖11 集群網絡流量Fig.11 Cluster network traffic

綜上所述，Spark SQL執行效率相對較快，它的查詢速度比Hive要快2.7倍。然而，當查詢總大小超過內存大小時，Impala則無法查詢。Hive處理的結果準確率較高，處理速度較慢。因此，Hive比較適用于批處理應用；Impala適合交互式查詢，系統的穩定性還有待提高；Spark SQL能夠降低Hive的延遲，比較適合多并發和流數據處理場景。

通過以上的實驗分析和比較，從文件格式角度來講，本文選擇能夠更好地適配Spark SQL的 Parquet列式存儲格式，以便快速地從HDFS中掃描找到相應的數據；從壓縮角度來看，本文采用Snappy壓縮方式，以減少數據輸入量和加快查詢速度；從Spark SQL自身特性分析，Spark SQL基于內存計算執行速度快，可以操作Hadoop上多樣化格式的數據并進行高效的結構化分析與處理，提供可用性更好的API進行數據分析，更加靈活且易擴展。鑒于此，綜合考慮以上幾個方面并結合應用驅動的大數據分析與計算實際需求，本文選擇Spark SQL作為SQL-on-Hadoop查詢系統，以適應快速證券大數據分析與計算場景下的高并發實時查詢應用需求。

5 實際應用

在本文實現基于SQL-on-Hadoop網上交易日志實時分析與計算平臺上，目前已存儲約60 TB的網上交易日志，并開發和移植了實時監控、統計分析、明細查詢等實際應用。

5.1 實時運維監控

對分布在全國各個節點和服務器的狀態實時監控并對各種狀態進行及時判斷和處理，能夠對整個系統的使用狀況有宏觀的把控。實時運維監控主要包括技術指標監控、業務指標監控和客戶分布。

1)如圖12所示，技術指標監控主要針對實時的請求延遲、成功率、系統冗余(帶寬流量/在線數)等指標進行監控。數據從千萬級別的當日日志數據中實時提取，從采集到存儲達到秒級實現。延遲情況主要包括登錄、委托、查詢(資金查詢)和轉賬這4類業務單位時間段內的平均耗時和峰值情況。系統冗余用于指示系統的資源使用情況，包括系統容量情況和系統帶寬使用情況，能實時展示系統當前冗余，有助于系統管理員及時掌握當前系統的使用情況。成功率主要包括登錄、委托、轉賬這幾類業務單位時間段內的處理成功率情況。

圖12 技術指標監控Fig.12 Technical index monitor

2)如圖13所示，業務指標監控主要針對登錄情況、轉賬情況、委托情況和實時在線人數等指標進行監控。可以從千萬級別的當日數據中實時觀察當前系統的客戶登陸數、系統發生的交易數量和轉賬金額等情況，整個過程實現秒級響應。

圖13 業務指標監控Fig.13 Business index monitor

3)如圖14所示，客戶分布主要針對實時在線客戶與委托分布，站點和來源省份的分布。在線分布從千萬級別的數據中指示一段時間客戶的登陸來源和委托來源分布。系統通過登陸和委托源IP關聯全球IP分布區域得出客戶的分布情況， IP分布來自10億級別的IP分布數據源。

圖14 客戶分布Fig.14 Customer distribution

5.2 客戶交易行為監控

日志分析更有價值的應用在于發現客戶的異常行為。如圖15、16所示，通過大數據平臺，可以實時掌握不同區域和營業部活躍用戶的分布，為業務部門做績效考核、精準營銷等提供數據支撐。

圖15 活躍用戶分布Fig.15 Active user distribution

圖16 行為軌跡分析Fig.16 Behavioral traces analysis

5.3 業務統計查詢與分析

1)統計報表：如圖17所示，統計報表是實際日志處理中的一項重要需求和應用，對計算的性能和實時性要求更高。根據具體業務功能需求，按照指定周期完成系統資源、登錄、委托和業務監控等統計任務。統計報表主要是提供給系統管理員分析系統的資源使用情況和系統健康狀態，以便做好相應的措施和規劃，同時為管理者決策提供數據支撐和參考依據。

圖17 業務報表Fig.17 Business report

2)明細查詢：如圖18所示，對網上交易日志的明細查詢，實現千億級數據秒級查詢響應。主要用于實時明細查詢，根據時間、系統、站點等多維條件查詢從TB級別的日志數據中快速準確地找到所需數據，查詢時效均能達到秒級響應。極大地方便了運維管理人員的工作，在節約大量的時間的同時提高了問題排查效率。

圖18 明細查詢Fig.18 Query details

6 結束語

本文研究了SQL-on-Hadoop技術在網絡日志分析中的應用。我們選取了其中最有代表性的3種SQL查詢引擎——Hive、Impala和Spark SQL，并使用TPC-H的測試基準對它們的決策支持能力進行測試及評估。構建面向證券行業的網絡日志分析平臺，實現萬億級日志存儲和高效、靈活的查詢系統，為海量日志集中分析與管理系統應用提供支持。目前SQL-on-Hadoop系統還存在若干問題有待解決，在有限的資源使用情況下和特定數據分布場景下提高查詢處理效率等問題都有待進一步的研究。

[1]OLINER A, GANAPATHI A, XU W. Advances and challenges in log analysis [J]. Communications of the ACM, 2012, 55(2): 55-61.

[2]李國杰，程學旗. 大數據研究：未來科技及經濟社會發展的重大戰略領域——大數據的研究現狀與科學思考[J]. 中國科學院院刊，2012， 27(6)： 647-657.

LI Guojie, CHENG Xueqi. Research status and scientific thinking of big data[J]. Bulletin of Chinese academy of sciences, 2012, 27(6): 647-657.

[3]王元卓，靳小龍，程學旗. 網絡大數據：現狀與展望[J]. 計算機學報, 2013, 36(6)： 1125-1138.

WANG Yuanzhuo, JIN Xiaolong, CHENG Xueqi. Network big data: present and future[J]. Chinese journal of computer, 2013, 36(6): 1125-1138.

[4]孟小峰，慈祥. 大數據管理：概念、技術與挑戰[J]. 計算機研究與發展， 2013， 50(1)： 146-149.

MENG Xiaofeng, CI Xiang. Big data management: Concepts, techniques and challenges [J]. Journal of computer research and development, 2013, 50(1): 146-149.

[5]JOSHI S B. Apache hadoop performance-tuning methodologies and best practices[C]//Proceedings of the 3rd ACM/SPEC International Conference on Performance Engineering. New York, USA, 2012: 241-242.

[6]LAMB W. The storyteller, the scribe, and a missing man: hidden influences from printed sources in the gaelic tales of duncan and neil macdonald [J]. Oral tradition, 2012, 27(1): 109-160.

[7]Apache.org. Apache Chukwa [EB/OL]. [2017-06-07].http://chukwa.apache.org/

[8]GOODHOPE K, KOSHY J, KREPS J, et al. Building LinkedIn’s real-time activity data pipeline [J]. Data engineering, 2012, 35(2): 33-45.

[9]APACHE ORG. Apache Flume [EB/OL]. [2017-06-07]. https://flume.apache.org.

[10]GHEMAWAAT S, GOBIOFF H, LEUNG S T. The Google file system[C]//Proc of the 19th ACM Symp on Operating Systems Principles. New York, USA, 2003: 29-43.

[11]THUSOO A, SARMA J S, JAIN N, et al. Hive—a petabyte scale data warehouse using Hadoop [C]// Proc of 2010 IEEE 26th International Conference. Piscataway, NJ, 2010: 996-1005.

[12]APACHE ORG. Apache HBase [EB/OL]. [2017-06-07]. https://Hbase.apache.org.

[13]APACHE ORG. Hadoop Streaming [EB/OL]. [2017-06-07].http://hadoop.apache.org/docs/r1.2.1/streaming.html.

[14]WEI J, ZHAO Y, JIANG K, et al. Analysis farm: A cloud-based scalable aggregation and query platform for network log analysis[C]//International Conference on Cloud and Service Computing. Hong Kong, China, 2011: 354-359.

[15]RABKIN A, KATZ R H. Chukwa: a system for reliable large-scale log collection[C]// International Conference on Large Installation System Administration. New York ,USA, 2010: 163-177.

[16]LOGOTHETIS D, TREZZO C, WEBB K, et al. In-situ mapreduce for log processing [C]//Usenix Conference on Hot Topics in Cloud Computing. Berkeley, USA, 2012: 26-26.

[17]TREZZO C J. Continuous mapreduce: an architecture for large-scale in-situ data processing[J]. Dissertations and theses-gradworks, 2010, 126(7): 14.

[18]Apache.org. HDFS Architecture Guide [EB/OL]. [2017-06-07]. http://hadoop.apache.org/docs/r2.7.2/hadoop-project-dist/hadoop-hdfs/HdfsUserGuide.html.

[19]DEAN J, GHEMAWAT S. Mapreduce: simplified data processing on large culsters[C]//Proc of the 6th Symp on Operating System Design and Implementation. San Francisco, USA, 2004: 137-150.

[20]HAN U G, AHN J. Dynamic load balancing method for apache flume log processing[C]//Information Science and Technology. Shenzhen, China, 2014: 83-86.

[21]Apache.org. Apache sqoop [EB/OL]. [2017-06-07]. http://sqoop.apache.org/.

[22]BITTORF M, BOBROVYTSKY T, ERICKSON CCACJ, et al. Impala: a modern, open-source SQL engine for Hadoop[C]//Proceedings of the 7th Biennial Conference on Innovative Data Systems Research. CA, USA, 2015: 4-7.

[23]FLORATOU A, MINHAS U F, OZCAN F. SQL-on-Hadoop: full circle back to shared-nothing database architectures [J]. Proc of the VLDB endowment, 2014, 7(12): 1199-1208.

[24]ZAHARIA M, CHOWDHURY M, FRANKLIN M J, et al. Spark: cluster computing with working sets [J]. Book of extremes, 2010, 15(1): 1765-1773.

[25]HE Y, LEE R, HUAI Y, et al. RCFile: a fast and space-efficient data placement structure in MapReduce-based warehouse systems. [C]// Proc of 27th IEEE Int Conf on Data Engineering. CA: IEEE Computer Society, 2011:1199-1208.

[26]MELNIK S, GUBAREV A, LONG J J, et al. Dremel: interactive analysis of web-scale datasets[J]. Communications of the Acm, 2011, 3(12):114-123.

何明，男，1975年生，博士，主要研究方向為大數據、推薦系統、機器學習。

常盟盟，男，1987年生，碩士研究生，主要研究方向為數據挖掘、機器學習。

劉郭洋，男，1986年生，碩士研究生，主要研究方向為大數據、數據挖掘。

Logminingandapplicationbasedonsql-on-hadoopqueryengine

HE Ming1， CHANG Mengmeng1， LIU Guoyang2， GU Chengxiang2， PENG Jike2

(1.Faculty of Information Technology, Beijing University of Technology, Beijing 100124, China; 2.Information Technology Management Department， Haitong Securities Co., Ltd., Shanghai 200001, China)

With the rapid development of computing and networking technologies, and the increase in the number of data acquisition methods, the demand for real-time processing of massive amounts of log data is increasing every day, and there is a calculation bottleneck when traditional log analysis technology is used to process massive amounts of data. With the development of open processing platforms in the era of big data, a number of big data processing systems have emerged for dealing with large-scale and diverse data. To effectively apply the advantages of Hadoop to the original businesses, in this study, we first investigated network log analysis methods based on big data technology and constructed a network log analysis platform for the acquisition, analysis, storage, high-efficiency and flexible queries, and the calculation of trillions of log entries. In addition, we compared and analyzed three representative SQL-on-Hadoop query systems including Hive, Impala, and Spark SQL, and identified the performance characteristics of this type of system. We used the TPC-H testing reference to test and assess their decision-making support abilities. We drew some useful conclusions from the analysis of the experimental data. We also suggest a few typical applications for this analysis and processing system for massive log data in the securities fields, which provides a solid foundation for further research.

big data; log analysis; data mining; Hadoop; query engine; data collection; indexed storage; securities business

10.11992/tis.201706016

http://kns.cnki.net/kcms/detail/23.1538.TP.20171021.1350.014.html

TP391

A

1673-4785(2017)05-0717-12

中文引用格式：何明，常盟盟，劉郭洋，等.基于SQL-on-Hadoop查詢引擎的日志挖掘及其應用J.智能系統學報， 2017, 12(5): 717-728.

英文引用格式：HEMing，CHANGMengmeng，LIUGuoyang，etal.Logminingandapplicationbasedonsql-on-hadoopqueryengineJ.CAAItransactionsonintelligentsystems, 2017, 12(5): 717-728.

2017-06-07. < class="emphasis_bold">網絡出版日期

日期：2017-10-21.

國家自然科學基金項目(91646201, 91546111, 60803086); 國家科技支撐計劃子課題(2013BAH21B02-01); 北京市自然科學基金項目(4153058, 4113076); 北京市教委重點項目(KZ20160005009); 北京市教委面上項目(KM201710005023).

何明. E-mail：heming@bjut.edu.cn.