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

王 艷，潘晨光

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

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

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

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

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

1 MapReduce上Hive的缺陷

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

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

表1 Hive的缺陷原理表

2 Impala的原理與架構

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

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

圖1 Impala的運行架構

1）客戶端SQL查詢通過ODBC發送到集群內任一Impalad。查詢規劃器采用Jflex和CUP解析SQL語句，解析查詢請求為多個執行片段發送至查詢協調器，查詢節點單獨原子執行相關操作。

2）查詢規劃器初始化Impalad執行任務，RDBMS存儲表的元數據信息。進程StateStored調度查詢請求，分發metadata數據，提供對外的Thrift服務，存儲集群中進程的資源。

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

3 HDFS和Impala結合的數據分析

3.1 HDFS和Im pala結合的優勢

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

表2 HDFS和Impala結合架構的優勢分析表

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

3.2 HDFS和Im pala相結合的實現方法

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

數據存儲信息通過Libhdfs與HDFS進行交互，通過HDFSGetHosts方式獲取文件數據塊所在節點位置信息，Simplescheduler由Round-robin算法實現，通過調度器Exec對生成執行計劃樹分配給對應的后端執行器執行。調用GetNext方法獲取計算結果，執行insert語句將計算結果通過Libhdfs寫回HDFS。Shuffle Join有穩定性能，適用大型復雜關聯操作。其流程框圖如圖2所示。

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

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

4 改進的CURE碰撞比對算法

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

CURE算法是自下而上的層次聚類，用定量特征點來表示簇，合并相鄰簇直到簇的數目在特定閾值范圍內。由于簇的個數無法提前預設，需對多個簇進行強制合并或把簇強行分割，影響聚類效果。為提高聚類的質量，本文提出將聚合條件設定為相鄰簇間距離達到設定閾值時聚類形成，簇間相似度決定簇的個數。Impala處理大數據量時，CURE聚類算法采用隨機取樣數據技術，分區聚類后將局部聚類的中間結果進行分析得到最后結果。先局部后整體的方法應用到分布式Impala系統中，CURE聚類算法可高效處理海量數據。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＞，從源數據集中抽取隨機樣本S，向量di創建簇Ci，實現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，簇標簽標記數據，若|tmpSet|＜min{|Cnew|，λ}，執行步驟5。

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

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

5 實驗與系統實現

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

6 結束語

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

[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/.