基于HDFS和IMPALA的碰撞比對分析

王 艷，潘晨光

（公安部第一研究所，北京 100048）

實時計算是不斷獲取、計算和分析大流量數(shù)據(jù)，迅速洞察變化原委，自動化響應(yīng)變化的數(shù)據(jù)[1]。交互式即席查詢和報表查詢面臨整合異構(gòu)數(shù)據(jù)源，統(tǒng)一元數(shù)據(jù)存儲和大規(guī)模迭代運算模型等難點。公共安全領(lǐng)域積累了大量的人員、案件、軌跡和社會行為等數(shù)據(jù)信息。實時分析和計算這些持續(xù)大流量的公共安全數(shù)據(jù)是巨大的挑戰(zhàn)。

Hadoop上的Hive追求高吞吐量，導(dǎo)致時間延遲較高。Hive可支持百億級的數(shù)據(jù)量，但很難應(yīng)對秒級響應(yīng)的需求，只適合做分鐘級別的離線分析系統(tǒng)而不支持實時分析系統(tǒng)[2-3]。Hive的缺陷導(dǎo)致其不能滿足業(yè)務(wù)高速發(fā)展所帶來的實時和高維的數(shù)據(jù)處理需求，但公共安全的情報分析需實時獲取當(dāng)前正在發(fā)生的案件和嫌疑人的狀況。

如何基于公共安全數(shù)據(jù)構(gòu)建大數(shù)據(jù)查詢系統(tǒng)實現(xiàn)關(guān)系查詢和實時跟蹤，是公共安全大數(shù)據(jù)迫切需要解決的問題。本文提出了HDFS和Impala相結(jié)合的架構(gòu)，搭建了存儲海量數(shù)據(jù)的分布式文件系統(tǒng)，實現(xiàn)了交互式數(shù)據(jù)查詢和分析，提供即席查詢的功能，便于快速獲取數(shù)據(jù)和決策支持?；贖DFS和Impala構(gòu)建大數(shù)據(jù)查詢系統(tǒng)可提供統(tǒng)一的元數(shù)據(jù)訪問和管理接口，支持SQL查詢優(yōu)化、列存儲、查詢謂詞下推、高效壓縮技術(shù)、預(yù)先計算、高效索引和并行查詢等，可按照時間、空間和業(yè)務(wù)進行分層和元數(shù)據(jù)管理，方便構(gòu)建兼容應(yīng)用。

為解決公共安全大數(shù)據(jù)的實時查詢問題，本文首次并創(chuàng)新性地將Impala計算引擎應(yīng)用于公共安全大數(shù)據(jù)的智能分析，整合非結(jié)構(gòu)化、半結(jié)構(gòu)化和結(jié)構(gòu)化的數(shù)據(jù)存儲和分析，設(shè)計了數(shù)據(jù)存儲組織結(jié)構(gòu)和數(shù)據(jù)分層策略，盡量隱藏查詢對原始文件訪問的需求，即席查詢共享存儲、統(tǒng)一計算，可擴展性強，實現(xiàn)了以人查案和以案找人的業(yè)務(wù)功能，取得了較好的實戰(zhàn)效果。

1 MapReduce上Hive的缺陷

分布式文件系統(tǒng)適合處理非結(jié)構(gòu)化數(shù)據(jù)，而已存儲在數(shù)據(jù)庫中的數(shù)據(jù)是結(jié)構(gòu)化的，結(jié)構(gòu)化數(shù)據(jù)轉(zhuǎn)換為非結(jié)構(gòu)化數(shù)據(jù)會丟失很多重要價值的信息[4]。MapReduce是基于磁盤進行數(shù)據(jù)處理，每次計算要經(jīng)歷從磁盤讀取數(shù)據(jù)、計算數(shù)據(jù)和保存數(shù)據(jù)等階段，導(dǎo)致運行過程復(fù)雜，迭代任務(wù)時效低，不適合對延時要求高的交互式分析或復(fù)雜迭代的數(shù)據(jù)分析任務(wù)。

Hive是面向行存儲的數(shù)據(jù)庫，不存儲和計算數(shù)據(jù)，底層執(zhí)行依賴MapReduce引擎，不能解決已有關(guān)系數(shù)據(jù)庫中數(shù)據(jù)的遷移和查詢操作[5]。運行機制是將結(jié)構(gòu)化數(shù)據(jù)文件映射為數(shù)據(jù)庫表，提供類SQL查詢，并將SQL語句轉(zhuǎn)換為MapReduce任務(wù)運行[6]。查詢先轉(zhuǎn)化為映射-歸約作業(yè)，再提交給集群以批量方式執(zhí)行。MapReduce調(diào)度只適合批量和周期長的任務(wù)，類似查詢結(jié)構(gòu)化數(shù)據(jù)的業(yè)務(wù)效率低，Hive的運行機制導(dǎo)致查詢速度慢[7]。Hive的缺陷原理如表1所示。

表1 Hive的缺陷原理表

2 Impala的原理與架構(gòu)

Impala是Cloudera參考Google Dremel思想實現(xiàn)交互SQL大數(shù)據(jù)查詢，支持Parquet列存儲格式，結(jié)構(gòu)嵌套記錄轉(zhuǎn)換成列存儲[8]，高效狀態(tài)機實現(xiàn)記錄正向和反向轉(zhuǎn)換，減少了查詢數(shù)據(jù)量；支持多層樹，查詢樹根節(jié)點接收查詢，底層節(jié)點獲取數(shù)據(jù)執(zhí)行查詢，使任務(wù)在數(shù)千個節(jié)點上并行執(zhí)行和聚合；采用推送方式傳輸數(shù)據(jù)，分散了網(wǎng)絡(luò)壓力，提高了任務(wù)的執(zhí)行效率。

Impala主要分為Impalad、StateStore和CLI等模塊[9]。Impalad與DataNode在相同節(jié)點上運行，接收查詢請求Coordinator。通過JNI調(diào)用Java前端解釋SQL查詢語句，生成查詢計劃樹，通過調(diào)度器把執(zhí)行計劃分發(fā)給數(shù)據(jù)對應(yīng)的Impalad運行，讀寫數(shù)據(jù)并行執(zhí)行查詢。StateStore跟蹤集群中Impalad運行狀態(tài)和位置信息，創(chuàng)建多線程處理注冊訂閱和心跳檢測。進程離線后，進入recovery模式反復(fù)注冊；進程重新加入集群后，自動恢復(fù)正常，更新緩存數(shù)據(jù)。CLI提供查詢的命令行、Hue、JDBC和ODBC使用接口。查詢的執(zhí)行過程如圖1所示。

圖1 Impala的運行架構(gòu)

1）客戶端SQL查詢通過ODBC發(fā)送到集群內(nèi)任一Impalad。查詢規(guī)劃器采用Jflex和CUP解析SQL語句，解析查詢請求為多個執(zhí)行片段發(fā)送至查詢協(xié)調(diào)器，查詢節(jié)點單獨原子執(zhí)行相關(guān)操作。

2）查詢規(guī)劃器初始化Impalad執(zhí)行任務(wù)，RDBMS存儲表的元數(shù)據(jù)信息。進程StateStored調(diào)度查詢請求，分發(fā)metadata數(shù)據(jù)，提供對外的Thrift服務(wù)，存儲集群中進程的資源。

3）查詢協(xié)調(diào)器執(zhí)行聚合函數(shù)Limit n，截取Top-n，完成局部Aggregation回傳結(jié)果至客戶端。查詢工作引擎通過流式交換輸出，協(xié)調(diào)客戶端提交查詢請求，分配任務(wù)至其他Impalad并收集執(zhí)行結(jié)果。Impalad執(zhí)行分配的任務(wù)，操作本地HDFS和HBase的數(shù)據(jù)完成查詢請求。

3 HDFS和Impala結(jié)合的數(shù)據(jù)分析

3.1 HDFS和Im pala結(jié)合的優(yōu)勢

由于Hive本身的缺陷，本文提出了采用Impala直接為存儲在HDFS中的數(shù)據(jù)提供快速、交互式SQL查詢的技術(shù)方案。HDFS和Impala結(jié)合的原理是把HDFS接入Impala后端作為存儲引擎，直接從HDFS獲取查詢所需數(shù)據(jù)，請求被解析成片段調(diào)度至相應(yīng)節(jié)點上執(zhí)行，某些源數(shù)據(jù)或中間數(shù)據(jù)存放在HDFS中[10]。Impala把多個執(zhí)行計劃分配到內(nèi)存中并行執(zhí)行，高效I/O調(diào)度和優(yōu)化的LLVM本地代碼完成初始化，中間結(jié)果在進程間進行流式回傳。HDFS和Impala架構(gòu)比MapReduce和Hive架構(gòu)的優(yōu)勢分析如表2所示。

表2 HDFS和Impala結(jié)合架構(gòu)的優(yōu)勢分析表

HDFS和Impala架構(gòu)的優(yōu)勢體現(xiàn)在：1）Impala直接在HDFS中存取數(shù)據(jù)，不必把中間過程寫入磁盤，節(jié)省了大量I/O開銷；2）減小了MapReduce的啟動作業(yè)開銷，Impala直接從對應(yīng)服務(wù)進程進行作業(yè)調(diào)度，提高了執(zhí)行效率；3）去掉MapReduce不太適合做SQL查詢的范式，Impala支持實時分析的MPP查詢引擎，降低了不必要的shuffle和sort等開銷；4）采用LLVM統(tǒng)一編譯代碼，減少了通用編譯的開銷；5）支持數(shù)據(jù)的I/O調(diào)度機制，盡量將數(shù)據(jù)分布到所在節(jié)點內(nèi)存中并行完成，省去了大量I/O網(wǎng)絡(luò)開銷。

3.2 HDFS和Im pala相結(jié)合的實現(xiàn)方法

Impala由JAVA前端與C++后端組成，接收客戶端連接進行查詢的Coordinator，通過JNI接口調(diào)用JAVA前端對查詢SQL分析生成執(zhí)行計劃樹。JAVA前端的執(zhí)行計劃樹以Thrift數(shù)據(jù)格式回傳Impala C++后端。其原子操作由計劃片段表示，查詢語句可由多個片段組成，片段0表示執(zhí)行樹的根，匯聚結(jié)果回傳查詢，執(zhí)行樹的葉子結(jié)點由Scan操作，可分布式并行執(zhí)行。

數(shù)據(jù)存儲信息通過Libhdfs與HDFS進行交互，通過HDFSGetHosts方式獲取文件數(shù)據(jù)塊所在節(jié)點位置信息，Simplescheduler由Round-robin算法實現(xiàn)，通過調(diào)度器Exec對生成執(zhí)行計劃樹分配給對應(yīng)的后端執(zhí)行器執(zhí)行。調(diào)用GetNext方法獲取計算結(jié)果，執(zhí)行insert語句將計算結(jié)果通過Libhdfs寫回HDFS。Shuffle Join有穩(wěn)定性能，適用大型復(fù)雜關(guān)聯(lián)操作。其流程框圖如圖2所示。

圖2 Impala查詢請求的流程控制

Broadcast Join將右表作小表分發(fā)在Join，Shuffle Join是分發(fā)后左表驅(qū)動右表進行Join。嵌套類型數(shù)據(jù)Parquet列存儲格式及擴展SQL查詢語義通過基于LLVM的Just-In-Time運行時代碼生成，查詢以最大CPU速度執(zhí)行，能快速擴展系統(tǒng)功能。Parquet格式實現(xiàn) Dictionary Encoding、Bit Packing、Delta Encoding、Run-Length Encoding等壓縮技術(shù)，過濾無關(guān)數(shù)據(jù)減少I/O。Run Length Encoding在列壓縮中減少3個數(shù)量級存儲，提升2～3個數(shù)量級的內(nèi)存應(yīng)用，Dictionary Encoding對磁盤空間的占用約為之前的1/20，對內(nèi)存的占用約為之前的1/5。

4 改進的CURE碰撞比對算法

碰撞比對算法的應(yīng)用是對嫌疑人多種信息進行分析處理，查找與嫌疑人或案件的相關(guān)信息、活動軌跡和網(wǎng)絡(luò)行為等。為滿足碰撞比對的需要，將數(shù)據(jù)從HDFS同步到Impala的表中。Impala上運行CURE聚類算法設(shè)計是將改進的CURE聚類算法對訓(xùn)練集進行聚類，對簇進行標識基于矩形的建模建立相關(guān)性模型，將待檢測數(shù)據(jù)與該模型進行碰撞比對。若符合該模型則是與嫌疑人相關(guān)的數(shù)據(jù)，否則判斷為與嫌疑人不相關(guān)的數(shù)據(jù)。

CURE算法是自下而上的層次聚類，用定量特征點來表示簇，合并相鄰簇直到簇的數(shù)目在特定閾值范圍內(nèi)。由于簇的個數(shù)無法提前預(yù)設(shè)，需對多個簇進行強制合并或把簇強行分割，影響聚類效果。為提高聚類的質(zhì)量，本文提出將聚合條件設(shè)定為相鄰簇間距離達到設(shè)定閾值時聚類形成，簇間相似度決定簇的個數(shù)。Impala處理大數(shù)據(jù)量時，CURE聚類算法采用隨機取樣數(shù)據(jù)技術(shù)，分區(qū)聚類后將局部聚類的中間結(jié)果進行分析得到最后結(jié)果。先局部后整體的方法應(yīng)用到分布式Impala系統(tǒng)中，CURE聚類算法可高效處理海量數(shù)據(jù)。Impala上改進CURE算法描述如下：

Dis（X1，X2）表示X1和X2間的距離，其距離度量是歐幾里得距離、曼哈頓距離或閔可夫距離等，本文采用歐幾里得距離。X1和X2是簇時，定義Dis（X1，X2）為相鄰簇中特征點間的距離，即Di（s X1，X2）=min{Di（s ri，r）j，ri∈Q(X1),rj∈Q(X2)}。

步驟1，輸入＜key，value＞，從源數(shù)據(jù)集中抽取隨機樣本S，向量di創(chuàng)建簇Ci，實現(xiàn)S={C1，C2，…，Cn}，Q（Ci），Q（Ci）={di}。

步驟2，將樣本S分割，若|S|＜2，終止。

步驟3，將S聚類，找出簇集S中特征點相鄰距離的簇Cu、Cv，Dis（Ci，Cj）=min{Dis（Ci，Cj），Ci∈S，Cj∈S,i≠j}。若Dis（Cu、Cv）＞w，終止。

步驟4，隨機取樣剔除孤立點，合并簇Cu和Cv,Cnew←Cu?Cv，tmpSet←φ ，計算 Cnew的 中 心 ：

步驟5，對局部簇聚類，合并距離近的簇，從Cnew中選擇di，若 tmpSet=φ ，Dis（di,tmpSet）=max{dist（dj,tmpSet），dj∈ Cnew}，Dis（ dj,tmpSet ）=min{Dis（ di,dk） ，dk∈ tmpSet}，將 di并入tmpSet，tmpSet←tmpSet?{di}。

步驟6，簇標簽標記數(shù)據(jù)，若|tmpSet|＜min{|Cnew|，λ}，執(zhí)行步驟5。

步驟 7，輸出＜key，value＞收縮代表點：Q（Cnew）←{dk+a*(hnew-dk∈ tmpSet),dk}，更 新 簇 集 S ← SCu-Cv+Cnew，執(zhí)行步驟2。KD數(shù)存放數(shù)據(jù)點，小頂堆存放簇，將簇按照與其最近鄰簇間距離升序排序。

Hadoop平臺下使用Hive類SQL語句實現(xiàn)不同粒度的聚合，類SQL語句會轉(zhuǎn)化為Map和Reduce任務(wù)去執(zhí)行，在某粒度上聚合實際數(shù)據(jù)時會造成的較大開銷，而Hive無法一次性實現(xiàn)多粒度融合。為提高在不同粒度的查詢響應(yīng)時間，基于Impala的改進CURE聚類算法將不同粒度上的實時數(shù)據(jù)一次性聚合后存儲到Impala中，可識別任意形狀的簇，不斷凝聚或分裂簇，對非球形簇的識別度較高。改進CURE聚類算法對孤立點敏感度低。在簇識別的過程中，若簇增長緩慢或異常的小，可作為異常點來剔除，降低了孤立點敏感度。

5 實驗與系統(tǒng)實現(xiàn)

碰撞比對系統(tǒng)通過界面拖拽可實現(xiàn)數(shù)據(jù)的任意碰撞或根據(jù)自定義規(guī)則進行碰撞，支持兩兩數(shù)據(jù)源碰撞和多數(shù)據(jù)源碰撞，方便實現(xiàn)以人找案和以案找人的功能。碰撞比對系統(tǒng)支持單點碰撞比對和分布式碰撞比對。省廳里某些數(shù)據(jù)在本地數(shù)據(jù)源里沒有碰撞出來，可分步到各地市數(shù)據(jù)源進行碰撞，將結(jié)果分別返回并且合并匯總再統(tǒng)一展示。碰撞效率高，比傳統(tǒng)的架構(gòu)要快數(shù)10倍，同時支持數(shù)據(jù)源上傳和碰撞結(jié)果下載。該應(yīng)用準確并極速地實現(xiàn)單類多源、多類多源數(shù)據(jù)間的碰撞比對，比傳統(tǒng)基于Oracle數(shù)據(jù)庫的碰撞比對性能提高上百倍，大大提高了破案效率。其系統(tǒng)的界面實現(xiàn)如圖3所示。

6 結(jié)束語

由于Hadoop和Hive處理數(shù)據(jù)存在不足，不適合對延時要求高的交互式分析、復(fù)雜迭代的數(shù)據(jù)處理和實時分析系統(tǒng)。為適應(yīng)公共安全領(lǐng)域?qū)崟r查詢的應(yīng)用需求，本文創(chuàng)新性提出將Impala框架應(yīng)用于公共安全領(lǐng)域數(shù)據(jù)的實時查詢分析中，研制了在Impala和HDFS上運行的改進CURE碰撞比對算法，為存儲在HDFS的數(shù)據(jù)提供快速、交互式的ANSI-92 SQL所有子集的SQL查詢，實現(xiàn)了異構(gòu)數(shù)據(jù)源的統(tǒng)一查詢，其并發(fā)客戶端處理的速度上超越了Hive。Impala不使用緩慢的Hive和MapReduce批處理，通過與商用并行關(guān)系數(shù)據(jù)庫中類似分布式查詢引擎，直接從HDFS中用SELECT、Join和統(tǒng)計函數(shù)查詢數(shù)據(jù)降低了延遲。該系統(tǒng)的實現(xiàn)對公安構(gòu)建大規(guī)模的數(shù)據(jù)分析查詢系統(tǒng)具有借鑒意義，可提供技術(shù)參考。

[1]MELNIK S，GUBAREV A，LONG Jingjing，et al.Dremel:innteractive analysis of Web-scale datasets[J].Proceedings of the VLDB Endowment，2010，3（1）：330-339.

[2] CDH4里的Impala安裝使用文檔[EB/OL].[2015-01-03].http://download.csdn.net/detail/lostage2/4911752.

[3] ENGLE C，LUPHER A，XIN R，et al.Shark:fast data analysis using coarse-grained distributed memory[EB/OL].[2015-02-03].http://libra.msra.cn/Publication/56916420.

[5]D'ORAZIO L，BIMONTE S.Multidimensional arrays for warehousing data on clouds[C]//Proc.the Data Management in Grid and Peer-to-Peer Systems.Berlin，Heidelberg:Spring-Verlag，2010：26-37.

[6]OLSTON C，REED B，SRIVASTAVA U，et al.Pig latin：A notso-foreign language for data processing[EB/OL].[2015-02-03].http://citeseer.ist.psu.edu/viewdoc/summary?doi=10.1.1.124.5496.

[7] DEBRABANT J，PAVLO A，TU S，et al.Anti-Caching:a new approach to database management system architecture[EB/OL].[2015-02-03].http://www.dajudeng.com/d20120810089e6ef5158fb 770bf68a5518.htm l.

[8] YOU J G，XI J Q，ZHANG P J，et al.A parallel algorithm for closed cube computation[J].Computer and Information Science，2008（8）：103-115.

[9]HAN H，LEE Y C，CHOI S，et al.Cloud-aware processing of MapReduce-based OLAP applications[EB/OL].[2015-02-03].http://www.researchgate.net/publication/262242831_Cloud-aware_proces sing_of_MapReduce-based_OLAP_applications.

[10]LICHTENWALTER R N，LUSSIER JT，CHAWLA N V.New perspectives and methods in link prediction[EB/OL].[2015-02-03].http://videolectures.net/kdd2010_lichtenwalter_npml/.