基于Spark的空間范圍查詢索引研究

陳業斌　劉　娜　徐　宏　劉　敏

(安徽工業大學計算機科學與技術學院　安徽 馬鞍山 243032)

0　引　言

近年來，隨著信息技術的不斷發展，空間數據增長迅速，傳統的數據處理方式不能滿足大量數據的分析處理需求，為了滿足日益增長的信息需求，選擇合適的大數據分析處理平臺尤其重要。傳統的數據處理系統將數據存儲于外存中，再進行數據處理解決問題，當傳統數據處理平臺處理大數據時，其計算能力的不足會導致系統性能隨著數據量的增大而急劇下降，在實際應用中并不適用。

Spark是分布式內存計算框架，具有高效性、高可靠性與高性能的特征，適用于大數據的數據分析和分布式并行處理。依靠HBase的分布式處理能力將單個任務分解為多個任務，使用集群計算的方式將被分解的任務分配給每個節點在內存中進行數據的并行處理，保證高可靠、高并發、低延遲地處理空間數據。Spark在大數據處理領域提出了彈性分布式數據集(RDD[1])的概念，將部分RDD數據集緩存于內存中，在重用計算時具備一定程度的容錯糾正機制[2]。因此，基于Spark的大數據處理平臺在計算處理數據時具有高可靠性與高性能。Spark中的Spark SQL組件使用SQL和DataFrame API兩類接口來處理結構化數據，通過使用Spark SQL組件操作SQL語句處理數據能夠更方便地梳理數據間的關系[10]。

對于空間大數據的研究不斷發展，技術逐漸成熟。2004年，Google公司發表了其內部設計的大數據處理平臺MapReduce[7,12]分布式計算模型、GFS[8]分布式文件系統、BigTable[9]分布式數據庫的論文，奠定了大數據處理框架的核心技術基礎[3,13-14]。

斯坦福大學Chu等在2006年NIPS會議發表的論文，對常用機器學習算法的形式進行分析，基于簡單通用具有自動容錯機制的MapReduce抽象提出了適用于海量數據的機器學習算法通用計算框架[4]。然而分布式數據處理框架可能存在著單點故障和磁盤IO效率低的情況，從而影響數據處理的效率。

2013年加州大學伯克利大學AMPLab實驗室發布了新一代基于內存計算模型的Spark分布式框架。該計算框架在開源社區的共同努力下，形成了和Hadoop相互促進的Spark生態圈，該生態圈包括Hadoop的HDFS分布式文件系統、實時計算框Streaming、機器學習包MLlib、圖計算GraphX、Spark SQL等，同時和Hadoop一樣支持HBase分布式數據庫[5]。基于Spark框架能夠高效處理實時數據[11]。

為了提高數據索引的效率，本文提出了基于Spark數據處理框架結合SIMBA的思路對數據索引進行優化。基于RDD機制實現數據的自動容錯機制，使用現有的分布式內存查詢分析引擎Spark SQL執行查詢操作，基于Spark的優化機制處理大數據下的查詢操作，高效準確地完成查詢操作。

1　相關工作

本文基于Spark分布式計算處理框架與分布式空間數據分析系統SIMBA[6](Spatial In-Memory Big data Analytics)的思想優化數據索引。Spark通過啟用內存分布數據集，構建了提供交互式查詢和優化迭代工作負載的計算框架。Spark基于RDD的實現將分析處理產生的中間結果顯式地留在內存中，改變了Hadoop大數據處理框架中出現的磁盤IO效率低的情況。

使用現有的Spark SQL查詢引擎中DataFrame數據組織處理結構化大數據的模塊，Spark DataFrame以RDD為基礎，同時具有Schema信息，處理分布式大數據時通過將數據庫對象分類的方式，提高了數據批處理的效率。

對數據通過采用分區策略優化數據分布，基于DataFrame將數據庫對象分類，提供完備的操作符用于處理數據集的方式，能夠提高查詢效率與改善查詢性能。

為了優化在大數據下的數據索引，依據SIMBA的思想基礎，在RDD數據集中采取了過濾優化和局部索引的策略，對Spark的內核不作任何修改，以保證系統能夠無縫嵌入新版本的Spark生態系統。在基于Spark的大數據處理框架下展開索引，能夠高效地確定數據集范圍內的候選點。對計算機應用與研究領域的大規模數據與復雜計算有極大的研究價值。

2　解決方案

在大數據索引中，為了改善系統性能，提出來索引優化策略，采用SIMBA項目的思路在Spark中提出優化策略，基于SQL創建索引并保存到內存中，從而提高查詢性能和查詢吞吐量。在大數據研究領域，基于Spark系統存取數據能夠有效查找數據集中的數據。在空間范圍中展開索引，為了提升索引效率和查詢性能，使用Spark SQL解析器執行空間查詢操作，并吸納SIMBA開源項目的思路，將空間進行分區，提出全局過濾和局部空間索引兩種優化策略，基于這些優化策略提升了索引性能。

2.1　分　區

在Spark集群計算系統中，采用數據分區的方式優化數據分布并采用操作符的方式處理數據，從而提高查詢效率，改善查詢性能。

對于分布式計算，采用Spark中Partitioner抽象類的RangePartitioner分區策略進行分區，如圖1所示，使用分區的方式處理數據需要滿足分區大小、數據局部性、負載均衡三個條件[11]。首先采用RangePartitioner方式接受RDD數據集和數據集的范圍邊界傳值。接下來，基于調用預定義采樣率的采樣方法大致估算出數據分布。隨后將原始的RDD數據集劃分為多個分區，當分區所包含的數據項超過了平均數量時，RangePartitioner將對當前分區進行重采樣。基于RangePartitioner分區能夠盡量保證分區中數據量分布均衡，分區之間排列有序。基于這種采樣分區的方式進行數據集分區，可以滿足系統分區的負載均衡，同時將空間位置相近的數據盡量相鄰，滿足數據分區的局部性。

圖1　數據分區策略

2.2　全局過濾

對數據集采用數據分區的方式將數據集分為多個大小相近的分區，雖然能夠改善查詢效率，然而在空間查詢時依然需要采用全局掃描的方式對數據進行篩選。為了能夠更大程度地改善查詢性能，提高查詢效率，使用全局過濾的方式過濾數據，減少需要掃描的分區數量。使用RangePartitioner分區函數計算每個分區的邊界，并將每個分區的上界保存在一個邊界數組中，在空間查詢時，通過判斷查詢點所落在的分區邊界進行分區剪枝，通過這種全局過濾的方式大幅度地降低掃描分區的數量，從而減少索引時間提升查詢效率。

2.3　局部索引

在空間查詢時，Spark數據處理框架通過在各分區數據集中建立線段樹的方式進行空間索引，基于線段樹壓縮空間，快速查找到空間數據[14]。

使用線段樹索引，線段樹作為一種二叉搜索樹，其索引方式如圖2所示，首先依據分區中的邊界值根據上下界端點的特征計算出中間值，將其作為根節點，再根據中間值可以將空間范圍分為三部分：第一部分為左集合，包含右邊界小于中間值的空間范圍間隔；第二類為右集合，包含左邊界大小中間值的空間范圍間隔；第三類為中間集合，包含中間值落在其邊界內的空間范圍間隔。使用線段樹進行范圍查詢，通過對節點值進行比較，小于節點值則進入左子樹中進行查找，大于節點值則進入右子樹中查找，依次迭代比較，直到匹配到對應的節點值。在定位到對應葉節點后，線性遍歷葉節點中的RDD數據集，相較于全局線性遍歷提高了查詢效率，降低了時間復雜度。

圖2　線段樹索引示意圖

2.4　優化策略

基于Spark處理框架的空間范圍查詢，如圖3流程所示。首先輸入查詢語句，使用DataFrames查詢引擎將查詢語句轉化為抽象語法樹。再使用Catalyst組件自帶規則解析查詢語句中對應的表屬性及其查詢條件，從而追蹤到查詢數據源，并生成分析后的邏輯計劃。通過緩存管理機制Cache Manager和索引管理機制Index Manager檢查內存中是否存在數據和索引，如果存在已經建立好的索引則讀入，最后獲得滿足條件的查詢結果集。

圖3　查詢步驟

本系統基于Spark數據處理框架，將創建的索引容器均封裝為RDD抽象，所以具備RDD的容錯機制，因此該系統在進行空間查詢時具備一定的容錯性。

3　查詢方法設計

3.1　集群計算

本系統基于Spark分布式計算框架，基于時空數據庫進行參數化查詢，系統利用通用編程語言，通過集群內存計算實現高效的數據處理及分析應用。

本系統基于Spark的實現數據索引，能將處理分析產生的中間結果顯式地駐留在內存中，通過分區策略來優化數據分布，采用全局過濾與局部索引的方式完成數據索引。在空間范圍查詢時，對查詢進行分析，嵌入空間索引優化策略，利用系統的分布式計算方法處理時空數據庫，提高查詢的性能與效率。

3.2　系統架構

本系統基于Spark 系統，依據Spark SQL執行范圍查詢，具有高吞吐量和低延遲特性。

在本系統中通過使用Spark應用程序的方式執行空間范圍查詢，具體系統架構如圖4所示，在Spark框架下使用Spark SQL解析器將查詢語句解析為抽象語法樹，使用Spark Catalyst優化器執行邏輯計劃生成。

圖4　Spark系統架構

在物理計劃執行階段，新建索引管理模塊Cache Manager和Index Manager用于執行內存和磁盤索引，對完成分區優化的數據執行查詢操作。本系統基于Spark中的Spark Catalyst和Spark SQL組件完成邏輯計劃，基于數據分區與索引管理模塊完成物理計劃，能夠更加高效、高性能地完成空間查詢。

3.2.1查詢方式

本系統基于Spark Catalyst展開空間查詢，并通過索引管理機制對Spark SQL內核的解析器執行空間查詢，使得本系統在空間索引時能夠支持必要的關鍵字查詢。

索引時通過使用關鍵字確定數據集中的查詢對象，將查詢拓展為空間點查詢與空間范圍查詢，通過建立線段樹的方式提高查詢性能與查詢效率。

3.2.2關鍵字接口

在空間范圍查詢時，使用空間范圍關鍵字選擇結構化表中查詢語言，使用空間點關鍵字表達空間對象，在查詢通過空間范圍的表達形式，在WHERE語句中優先匹配進行查詢。對于空間范圍查詢的表達形式如下所示：

POINT(point1,round1)

在系統中開展的空間范圍查詢操作，空間范圍查詢為查詢數據集中空間屬性落在指定空間范圍內的所有記錄，具體索引表達形式如下：

SELECT * from locations

WHERE POINT(locations.point)

IN POINT({34,23},40)

3.2.3復合索引查詢

本系統空間范圍查詢基于Spark SQL的語法，通過索引管理機制在時空數據集上建立空間范圍索引，不但能夠執行空間查詢還能夠執行包含關鍵字的復合查詢。在執行空間查詢中為了能夠提高查詢性能與查詢吞吐量，提出來建立線段樹的方式完成索引，具體查詢表達形式如下：

CREATE INDEX pointIndex ON locations(point)

USER INTERVALTREE

在復合查詢時，本系統基于空間關鍵字進行查詢，能夠查詢時空數據集中落在空間范圍內的所有記錄，并對結果集基于屬性id做GROUP操作，復合查詢表達形式如下：

SELECT locations.point from locations

WHERE POINT(locations.point)

IN POINT(point1,round)

GROUP BY locations.point

3.3　索引機制

本系統通過提供索引管理機制提升索引效率。基于索引管理模塊在時空數據集上建立空間索引，創建索引并保存在內存中，建立包含空間查詢與空間查詢建立所依賴的RDD數據集。基于索引管理機制進行空間索引，能夠快速定位數據集位置，從而實現較高的查詢吞吐量與較低的查詢延遲。

采用線段樹的思想設計索引。通過二分法構造相應區間，并構造區間內的子區間，將其存儲為新的節點。通過上下界的兩個端點能夠將每個節點區間定位到數據集中，通過SQL接口在各個分區數據集中建立對應的空間索引。基于線段樹進行空間索引，能夠高效查詢到所有覆蓋在給定空間范圍的數據。

4　實驗環境

本文的測試平臺為十個節點構建的集群運行環境，在集群上運行對比實驗，其中的8Xeon E5-2620，2.00 GHz處理器，96 GB內存。在這十個節點中均使用Ubuntu系統，在Ubuntu 14.04.2 LOS系統下搭建Apache Hadoop 2.4.1大數據處理框架，建立Apache Spark 1.5.2集群計算平臺，選擇具有2.00 GHz處理器和96 GB內存的2臺機器中的一臺作為主節點，剩余機器作為從節點。構建的Spark集群將以Standalone模式運行，并配置所有的從節點最高可以使用15 GB內存。

實驗使用的真實數據集采用紐約出租車接單數據集為測試數據，包括2013年全年紐約市所有出租車的接單數據，對數據進行去重處理和數據格式檢測，獲得包含1.7億條出租車接單記錄元組的有效實驗數據集，每一條元組包含，出租車接單ID、接單開始時間和接單結束時間。在實驗中測試系統的空間點查詢和空間范圍查詢兩種查詢操作的效率，在本文中使用OS(Optimized Spark)表示優化之后的Spark，將OS與對應原生Spark(未被優化的Spark)的基準查詢效率進行比較，通過比較查詢時的吞吐量和查詢延遲來評價優化后的系統對數據索引性能的改善。每個測試基于500次實驗進行查詢，查詢效率按照吞吐量和查詢延遲進行衡量：吞吐量=查詢數量/總執行時間(min)；查詢延遲=總執行時間(s)/查詢數量。

通過比較在數據集大小不同的情況下，原生Spark系統與OS系統進行空間查詢時的查詢延遲與吞吐量的變化，對兩個系統進行性能與效率的評價。如圖5所示，原生Spark系統與OS系統在數據集大小不斷增加的時候，查詢延遲的時間也不斷增加，相比于OS系統，原生Spark系統的延遲時間增長率要遠遠高于OS系統。如圖6所示，是在不同數據集大小的情況下吞吐量的變化，隨著數據量的增長，兩個系統的吞吐量都發生了一定程度的下降，然而OS系統依舊優異于原生Spark系統。在空間點查詢時，OS系統的查詢操作包含全局過濾和分區查詢兩部分優化，因此OS系統在查詢性能與效率上優于原生Spark系統。

圖5　數據集大小對于查詢延遲的影響

圖6　數據集大小對于吞吐量的影響

對空間范圍查詢中，通過比較在不同數據量大小的情況下原生Spark系統與OS系統的查詢延遲時間與吞吐量判斷系統的性能差異。首先需要生成一個落在全局空間范圍的空間標識，基于泰森多邊形將整個查詢空間劃分為多個多邊形區域。空間范圍查詢類似于空間點查詢，OS系統的查詢操作時依舊保證全局過濾和分區查詢兩部分優化，如圖7、圖8所示，在執行空間范圍查詢時，本系統的查詢效率依舊同樣優于原生Spark SQL查詢策略。

圖7　數據集大小對于查詢延遲的影響

圖8　數據集大小對于吞吐量的影響

當分區較大時，空間查詢的結果集可能覆蓋多個分區，可以通過固定數據集大小測試不同分區數量的情況下兩個系統的效率。如圖9、圖10所示，為不同分區下兩個系統的查詢延遲和吞吐量變化。由圖9、圖10可以觀察得出，OS系統在一定分區大小范圍內，隨著分區數量的增大，查詢延遲呈現降低趨勢，吞吐量大小增長。當分區數量增大，分區大小過小時，空間范圍查詢覆蓋大量的分區，那么需要遍歷的數據量將會增多，因為導致了查詢延遲的增長，吞吐量下降，查詢性能的降低。原生Spark系統的查詢延遲和吞吐量的變化與分區數量無明顯關系，因為使用原生Spark系統進行查詢需要對整個數據集進行遍歷，因此分區大小的改變不會對查詢性能產生影響。

圖9　分區大小對于查詢延遲的影響

圖10　分區大小對于吞吐量的影響

實驗主要分為三組對比實驗，第一組圖5、圖6，針對不同大小的數據集，在執行空間查詢時OS系統與Spark SQL的執行效率對比；第二組圖7、圖8，針對不同大小的數據集，在執行空間范圍查詢時OS系統與原生Spark SQL程序的執行效率對比；第三組圖9、圖10，針對不同大小的分區數量，在執行空間查詢時本系統與原生Spark SQL程序的執行效率對比。

5　結　語

本文提出了基于Spark的空間查詢擴展系統OS，基于OS完成空間范圍查詢。使用Spark系統中的Spark Catalyst組件的SQL解析器的操作方式進行查詢。在Spark 中嵌入索引管理機制，將其封裝在RDD內，用于提高查詢效率。通過建立線段樹存儲數據的方式提高數據檢索的效率。對于數據預處理時采用RangePartitioner分區策略的方式對數據進行分區，基于全局過濾和局部索引進行查詢。保證該系統在進行查詢操作時能夠保持高吞吐量和低延遲特性，提高查詢效率。

將在OS與Spark下進行查詢實驗的性能進行比較，由實驗結果分析，隨數據集大小的增加，基于該拓展系統(OS)完成的查詢操作在查詢延遲與吞吐量檢測上均優于原生Spark系統，當分區數量發生變化時，該拓展系統(OS)在性能上依舊優于原生Spark系統。

雖然該拓展系統取得了初步的研究成果，但是在未來的工作當中，本系統仍有較多需要拓展改進的部分，在系統中可以增加時態查詢，時態查詢也是時空數據庫中的研究重點。與此同時，系統優化也是未來的研究中不可或缺的部分，通過嵌入更多的優化策略提高索引的效率與性能。

[1] Zaharia M,Chowdhury M,Das T,et al.Resilient distributed datasets:a fault-tolerant abstraction for in-memory cluster computing[C]//Usenix Conference on Networked Systems Design and Implementation.USENIX Association,2012:2-2.

[2] 高官濤,鄭小盈,宋應文,等.基于Spark MapReduce框架的分布式渲染系統研究[J].軟件導刊,2013,12(12):26-29.

[3] Dean J,Ghemawat S.MapReduce:simplified data processing on large clusters[C]//Conference on Symposium on Opearting Systems Design & Implementation.USENIX Association,2008:10-10.

[4] Chu C T,Kim S K,Lin Y A,et al.MapReduce for machine learning on multicore[C]//Conference and Workshop on Neural Information Processing Systems,NIPS,2006,6:281-288.

[5] Zaharia M,Das T,Li H,et al.Discretized streams:fault-tolerant streaming computation at scale[C]//Twenty-Fourth ACM Symposium on Operating Systems Principles.ACM,2013:423-438.

[6] Xie D,Li F,Yao B,et al.Simba:Efficient In-Memory Spatial Analytics[C]//International Conference on Management of Data.ACM,2016:1071-1085.

[7] 張宇,程久軍.基于MapReduce的矩陣分解推薦算法研究[J].計算機科學,2013,40(1):19-21.

[8] Scott M L,Peterson L L.Proceedings of the 19th ACM Symposium on Operating Systems Principles 2003,SOSP 2003,Bolton Landing,NY,USA,October 19-22,2003[C]//Acm Symposium on Operating Systems Principles,2003(53):113-126.

[9] 趙宇蘭,柳欣.基于連接依賴信息的分布式連接查詢優化算法[J].現代電子技術,2016,39(460):28-32.

[10] 金澈清,錢衛寧,周敏奇.數據管理系統評測基準:從傳統數據庫到新興大數據[J].計算機學報,2015,38(1):18-34.

[11] Xin R S,Gonzalez J E,Franklin M J,et al.GraphX:a resilient distributed graph system on Spark[C]//International Workshop on Graph Data Management Experiences and Systems.ACM,2013:1-6.

[12] Armbrust M,Xin R S,Lian C,et al.Spark SQL:Relational Data Processing in Spark[C]//ACM SIGMOD International Conference on Management of Data.ACM,2015:1383-1394.

[13] Zhong Y,Zhu X,Fang J.Elastic and effective spatio-temporal query processing scheme on Hadoop[C]//ACM Sigspatial International Workshop on Analytics for Big Geospatial Data.ACM,2012:33-42.

[14] Tan H,Luo W,Ni L M.CloST:a hadoop-based storage system for big spatio-temporal data analytics[C]//ACM International Conference on Information and Knowledge Management.ACM,2012:2139-2143.