基于Spark平臺的地理數據并行裝載技術

賀夢潔，朱美正，初 寧，楊 崗

(1. 中國電子科技集團第十五研究所，北京 100000；2. 中國電子設備系統工程公司研究所，北京 100000)

基于Spark平臺的地理數據并行裝載技術

賀夢潔1，朱美正1，初 寧2，楊 崗2

(1. 中國電子科技集團第十五研究所，北京 100000；2. 中國電子設備系統工程公司研究所，北京 100000)

隨著地理信息系統研究的不斷深入發展，其應用領域不斷擴張，地理數據規模越來越大，高性能的地理數據處理成為迫切并且必然的要求。為了適應地理信息系統發展的需要，本文設計了基于Spark平臺的地理數據并行處理框架，并最終設計實現了地理數據并行裝載技術，使得地理數據裝載效率得到顯著提高，為今后實現地理數據實時更新、訪問、分析等奠定了良好的基礎。

Spark；地理信息系統；并行裝載

0 引言

隨著信息時代的來臨，人們對地理數據的依賴性越來越強。如今線上貿易滲透到生活的每個角落，物流公司需要利用地理數據節約成本、提高效率；經銷商需要參考地理數據做出重要的商業決策；人們出行也基本不再使用紙質地圖，有地圖應用可以給出詳細的路線指導。可見，衣、食、住、行無不和地理數據相關，因此對地理數據的需求量逐年呈爆炸式增長，稱為海量亦不為過。與此同時，對地理數據處理的性能需求也越來越高。地理數據處理包括裝載、組織、查詢、分析等等，傳統的單個CPU的計算資源，單進程、集中式的數據處理方式已經遠遠無法滿足目前的性能需求，為了突破傳統處理方式的限制，考慮在高并發空間大數據庫的基礎上，采用多機、多進程、分布式的數據并行處理方式，充分利用并行計算巨大的數值計算和數據處理能力的優勢，實現海量地理數據的快速裝載、組織、查詢、分析、更新和訪問。

地理數據并行裝載技術是地理數據并行處理的第一步，也是后續處理的基礎。完整、正確且合理平均的將相關地理數據裝載到相應的分布式存儲系統，才能保證后續檢索、更新和訪問等處理過程的效率。因此地理數據并行裝載技術對于海量地理數據高效處理的意義尤為重要。

本文將從Spark分布式并行計算平臺、基于Spark平臺的地理數據并行處理框架、地理數據并行裝載技術等方面進行詳細介紹。

1 基于Spark平臺的地理數據并行處理框架

1.1 Spark分布式并行計算平臺

Spark于2009年誕生于加州大學伯克利分校AMPLab[1]，目前已經成為大數據領域應用最廣泛、最高效的通用集群計算平臺。Spark是基于MapReduce思想實現的分布式并行計算框架，繼承了Hadoop的MapReduce的優點，但相比MapReduce有更多拓展應用和更高效的計算模型。

Spark創造性地提出了分布式內存存儲結構彈性分布式數據集RDD的概念，在此基礎上可以在Spark這個統一的框架下高效地支持包括批處理、迭代算法、交互式查詢、流處理等在內的多種計算模式。

大數據計算流程往往分為多個階段，在MapReduce中，不同計算階段之間重用數據，需要將上一個階段的計算結果保存到外部存儲系統（如分布式文件系統HDFS）中[10]，由此導致了大量的數據復制、磁盤I/O、序列化、反序列化等開銷[2]，大大降低了計算效率。而Spark將執行模型抽象為通用的有向無環圖執行計劃（DAG），可以將有多個計算階段的任務串聯或者并行執行，計算階段的中間結果用分布式內存存儲結構彈性分布式數據集RDD的形式存儲在內存中，因此不同計算階段之間只需讀寫內存，無需讀寫磁盤[8]。在內存空間不足的情況下，也可以像Hadoop一樣存儲在磁盤上。相對于MapReduce上的批量計算、迭代型計算以及基于Hive的SQL查詢，Spark可以帶來上百倍的性能提升[1]。

Spark所提供的接口非常豐富。除了提供基于Python、Java、Scala和SQL的簡單易用的API以及內建的豐富的程序庫之外，Spark還能和其他大數據工具密切配合使用[3]。AMPLab開發以Spark為核心的伯克利數據分析棧（BDAS）的目標是在一套軟件棧內完成各種大數據計算任務。BDAS涵蓋支持結構化數據SQL查詢與分析的查詢引擎Spark SQL和Shark，提供機器學習功能的系統MLbase及底層的分布式機器學習庫MLlib、并行圖計算框架GraphX、流計算框架Spark Streaming、采樣近似計算查詢引擎BlinkDB、內存分布式文件系統Tachyon、資源管理框架Mesos等組件[1]，這些組件和Spark緊密集成，共享數據，因此可以構建出無縫整合不同處理模型的應用[9]。

以上這些優點和特性使得Spark成為目前大數據處理首選的計算平臺，也是本文將Spark應用于海量地理數據處理系統的原因。

1.2 適用于并行處理的地理數據模型

數據模型是并行計算的基礎，空間數據組織結構是否合理直接決定數據并行處理的性能，決定數據部署、維護、檢索、獲取的效率，尤其是對于海量空間數據的處理更是重要。為了實現地理數據的高效處理，需要重點研究并行計算環境下地理數據模型應該滿足的要求，如空間實體對象的完整性特征，空間實體對象之間不存在緊耦合狀態，彼此相互獨立，其拓撲、度量等關系可動態創建等等[6]。地理數據模型及數據結構應適用于并行I/O操作，可實現基于這類數據模型的并行快速查詢和迅速處理。

考慮以上特征，設計了面片，這種非結構化地理數據模型作為地理數據并行處理的對象單元。首先，地理數據按數據類別（如矢量數據、影像數據等）分類存儲。某一類數據分N個面片數據集存于分布式數據庫系統的各個物理存儲節點上。每個面片數據集有固定的空間參考、面片劃分方案、資源類型、面片類型、應用類型、面片格式。面片數據集內部又按不同級別對面片分類，每種級別對應一個確定大小的比例尺，級別越大對應的分辨率越高、比例尺越大，支持創建金字塔數據。每個級別數據集里又分為N個桶或桶集，桶集包含N個桶，按桶組織可以避免因面片數目巨大而造成管理低效，更易于部署、更新、遷移。桶里可以按多個版本區分數據，版本表示數據來源、年份等等含義。版本數據集再包含面片數據，面片數據是數據存儲的最小粒度單位。桶集有桶分布位圖，反映哪些桶有數據；桶有面片分布位圖，反映某個桶的某個版本的面片分布情況。面片的邏輯模型如圖1所示。

圖1 面片的邏輯模型Fig.1 Piece Logical Model

面片這種數據模型互相獨立，可存儲于分布式數據庫，適用于并行I/O操作，在此基礎上可實現地理數據分布式并行處理。

1.3 地理數據并行處理框架

1.3.1 Spark開發環境及其分布式集群搭建

目前Apache Spark 支持三種分布式部署方式，分別是standalone、spark on Mesos和spark on YARN[7]。Standalone模式，即獨立模式，其內部實現了容錯性和資源管理，無需依賴任何其他資源管理系統。后兩種則是在企業級應用和工業生產中常用的模式，部分容錯性和資源管理交由統一的資源管理系統完成，讓Spark運行在一個通用的資源管理系統之上，這樣可以和其他計算框架共用一個集群資源，最大的好處是降低運維成本和提高資源利用率。從一定程度上來說，standalone模式是后兩種模式的基礎。借鑒spark開發模式，我們可以得到一種開發新的計算框架的一般思路：為了快速開發，可以暫不考慮服務（比如master/slave）的容錯性，先設計出它的standalone模式（目前spark在standalone模式下沒有單點故障問題，這是借助zookeeper實現的），之后在開發相應的wrapper，將standalone模式下的服務原封不動地部署到資源管理系統mesos或yarn上，由資源管理系統負責服務本身的容錯。由于在項目初期，本文設計的地理數據并行處理框架就是在spark的standalone模式下實現的。

本文的實驗測試環境為兩臺Linux測試機。集群包括一個master節點和兩個worker節點，兩個worker節點分別部署在兩臺機器上，其中一臺機器同時作為master節點。每個worker節點有多個CPU，以達到多機、多進程并行處理的目的。測試使用的Spark集群如圖2所示。

圖2 本文測試使用的Spark集群模型及其硬件信息Fig.2 Spark Cluster Model of Test and Its Hardware Information

1.3.2 基于Spark的地理數據并行處理框架

Spark應用是用戶提交的應用程序。Driver進程是Spark應用的主控進程，負責應用的解析、切分Stage并調度Task到Executor執行，包含DAGScheduler等重要對象。Spark應用的提交包含兩種方式：其一，Driver進程運行在客戶端，對應用進行管理監控；其二，主節點指定某個Worker節點啟動Driver進程，負責整個應用的監控。本文實驗采取的是第一種方式。

應用執行流程如圖3所示。用戶啟動客戶端，之后客戶端運行用戶程序，啟動Driver進程。在Driver中啟動或實例化DAGScheduler、SparkContext等組件。客戶端的Driver向Master注冊。Worker 向Master注冊，Master命令Worker啟動Executor。Worker通過創建ExecutorRunner線程，在ExecutorRunner線程內部啟動ExecutorBackend進程。ExecutorBackend啟動后，向客戶端Driver進程內的SchedulerBackend注冊，這樣Driver進程就能找到計算資源。

圖3 Driver進程在Client端的應用執行流程圖Fig.3 The Flow Chart of Application Execution When Driver Progress in Client End

彈性分布式數據集（Resilient Distributed Dataset，RDD）是Spark的核心數據結構，可以通過一系列算子進行操作。一個操作執行完畢，RDD變轉換為另一個RDD。Spark為了系統的內存不至于快速用完，使用延遲執行的方式執行。只有當RDD遇到Action算子時，將之前的所有算子形成一個有向無環圖RDD DAG，由Spark Action（如count、collect等）算子觸發整個RDD DAG執行。這樣的過程稱為一個Job。

Spark應用提交后經過一系列轉換變成Job、Stage、Task幾個層次被調度，轉換過程如圖4所示。在Spark應用程序內部，用戶通過不同線程提交的Job可以并行運行。默認情況下，Spark的調度器以FIFO（first in first out）方式調度Job。RDD的Action算子觸發Job提交到Spark后得到的RDD DAG，由DAGScheduler根據RDD的寬依賴關系將其分解為Stage DAG，每個Stage中產生相應的Task集合，再由TaskScheduler將各個Task分發到Executor執行。每個Task對應相應的一個數據塊，使用用戶定義的函數處理數據塊。

圖4 Spark應用轉換流程圖Fig.4 Spark Application Convertion Flow Chart

2 地理數據并行裝載技術

2.1 基于Spark的地理數據并行裝載實現步驟

如前文所說，本文設計了面片，這種非結構化地理數據模型作為地理數據并行處理的對象單元。因此，地理數據并行裝載技術要實現的目的就是把源數據轉換成面片這種數據類型，并將這些面片集存入分布式數據庫或分布式文件系統中。源數據通常按不同級別（對應不同的比例尺）以圖幅數據文件為單位存儲。

地理數據包括矢量數據、影像數據、數字高程模型數據、數字地面模型數據等等常用的數據類型，種類繁多，并行轉載步驟略有差異。以矢量數據為例，在本文設計的基于Spark的地理數據并行處理框架上實現矢量數據并行裝載基本步驟流程圖如圖5所示。

用戶提交矢量入庫任務并輸入需要入庫的矢量源數據路徑。程序解析源數據路徑，獲得源數據包括的所有圖幅文件路徑。按一定的規模（地理范圍大小）為單位將這些圖幅文件分堆，由Driver進程隨機分配給各個Worker，由各個Worker并行進行接下來的入庫流程。每個Worker各自選取一堆圖幅文件，加載這些圖幅文件并按面片規則對其做切片處理，得到面片碎片結構的數據集，然后按照面片規則將面片各部分合并得到面片碎片集結構的數據集。判斷面片碎片集是否完整，若完整，合并面片并入庫；若不完整，緩存。將各節點的不完整面片碎片集再次合并，判斷面片碎片集是否完整，若完整，合并面片并入庫；若不完整，緩存。重復上個步驟，直到全部面片入庫，入庫流程結束。

圖5 矢量數據并行裝載流程圖Fig.5 Vector Data Parallel Loading Flow Chart

2.2 基于Spark的地理數據并行裝載算法實現思路

本文的地理數據并行裝載算法使用Java語言編程，結合基于Spark的地理數據并行處理框架和其算子RDD進行綜合設計。算法實現思路主要分為一下兩步：

Step1：源數據部分入庫，得到不完整面片：使用parallelize將源數據圖幅文件以JavaRDD 的形式分布到多個機器上。

使用flatMap(JavaRDD)對源數據圖幅文件切片，得到完整面片和面片碎片。將完整面片入庫，返回面片碎片集JavaRDD>。

使用mapToPair(JavaRDD>)得到以面片碼為Key的面片碎片鍵值對數據集JavaPairRDD。

使用filter(._1().startsWith(“ERROR:”)).collect-AsMap()過濾得到其中的錯誤信息并對其做錯誤處理。

Step2：面片碎片合并，完整面片入庫，不完整面片做相應處理：

使用reduceByKey(_,_)將相同面片碼的面片碎片合并，得到面片碎片集JavaPairRDD。

判斷面片是否完整，使用兩個filter分別得到完整面片碎片集和不完整面片碎片集。對于完整面片碎片集，使用mapValues將碎片合并成完整面片并入庫，使用Filter().collectAsMap得到錯誤信息集Map并對其進行相應處理；對于不完整面片碎片集，使用mapPartitions()對其進行處理，在緩存中尋找各個不完整面片缺少的碎片，如果某個面片缺少的碎片全部找到則合并完整入庫，如果不能全部找到則緩存，返回錯誤信息集JavaRDD>，再使用collect()得到錯誤信息集List>并做錯誤處理。

基于Spark的矢量數據并行裝載算法流程如圖6所示。

圖6 基于Spark的矢量數據并行裝載算法流程圖Fig.6 The Flow Chart of Vector Data Parallel Loading Algorithm Based on Spark

3 矢量數據裝載實驗

為了測試本文提出的地理數據并行裝載技術的性能，選取部分矢量源數據做裝載測試。實驗環境包括單機單進程矢量數據裝載環境和多機多進程矢量數據并行裝載環境。單機單進程矢量數據裝載測試用到的機器是linux系統、16G內存、8CPU，用Java編寫的矢量單機入庫程序，測試時后臺只啟動了一個應用服務進程；多機多進程矢量數據并行裝載測試用到的分布式集群如上文圖2所示，集群包括一個master節點和兩個worker節點，兩個worker節點分別部署在兩臺Linux測試機上，一臺機器16G內存、8CPU同時作為master節點，另一臺機器12G內存、8CPU，在Spark并行計算框架下用Java編寫的矢量并行入庫程序，每個worker啟動三個應用服務進程。測試用到的數據有：1:100萬比例尺下112萬平方公里的全要素矢量數據、1:25萬比例尺下2萬平方公里的全要素矢量數據、1:5萬比例尺下24萬平方公里的全要素矢量數據。

測試用的數據和測試結果如表1所示。

表1 測試數據及單機單核、雙機三核并行入庫時間

從測試結果可以看出，本文提出的地理數據并行裝載技術明顯提高了地理數據裝載的速度，并且在數據規模越大的情況下，并行裝載的優勢越顯著。在實際應用中，集群的規模可在此基礎上數十倍甚至上百倍，可以預見并行裝載技術的效率將非常高。

4 結束語

本文的主要成果是設計了基于Spark的地理數據并行處理框架和實現了地理數據并行裝載技術。相較于傳統的單機單進程的地理數據裝載方式，利用地理數據并行裝載技術可以使得地理數據裝載效率得到顯著提高，大大縮短了地理數據裝載所需時間，為今后實現地理數據實時更新、訪問、分析等奠定了良好的基礎。

[1] 高彥杰. Spark大數據處理: 技術、應用與性能優化[M]. 機械工業出版社, 2015.01.

[2] 王迅, 馮瑞. 基于Spark的海量圖像檢索系統設計. 微型電腦應用, 2015, 31(11): 11-17.

[3] Karau.H等著, 王道遠譯. Spark快速大數據分析[M]. 北京:人民郵電出版社, 2015.09.

[4] 崔鑫. 海量空間數據的分布式存儲管理及并行處理技術研究[D]. 國防科學技術大學研究生院, 2010.

[5] 金翰偉. 基于Spark的大數據清洗框架設計與實現[D]. 浙江大學計算機科學與技術學院, 2015.

[6] 張廣第, 分布式環境下海量空間數據的存儲和并行查詢技術研究[D], 南昌: 江西理工大學, 2012.

[7] 溫馨、羅侃、陳榮國等, 基于Shark/Spark的分布式空間數據分析框架[J]. 地球信息科學, 2015.4, 17(4): 401-407.

[8] 王迅、馮瑞等, 基于Spark的海量圖像檢索系統設計[J]。微型電腦應用, 2015, 31(11): 11-17.

[9] 方金云等, 基于Spark的空間數據實時訪存技術的研究[J]。地理信息世界, 2015.12, 22(6): 24-31.

[10] 霍紅衛, 林帥, 于強, 等. 基于MapReduce的模體發現算法[J]. 新型工業化, 2012, 2(9): 18-30.

Geographic Data Parallel Loading Technology Based on Spark

HE Meng-jie1, ZHU Mei-zheng2, CHU Ning3, YANG Gang3
(1. No.15 Institute of China Electronics Technology Corporation, Beijing, 100000; 2.No.15 Institute of China Electronics Technology Corporation,Beijing, 100000; 3. Institute of China Electronics System Engineering Company, Beijing, 100000)

With the deepening development of GIS research，its application domain expands unceasingly, geographic data scale grows so fast that high-performance of geographic data processing becomes a urgent and inevitable requirement. In order to meet the needs of GIS development, in this paper, geographic data parallel computing framework was designed based on Spark parallel computing platform and geographic data parallel loading was realized, on this basis, the loading efficiency of geographic data improves significantly, geographic data update, access, analysis in real time becomes possible.

Spark; GIS; Parallel loading

TP311

A

10.3969/j.issn.1003-6970.2016.12.016

賀夢潔(1992-)，女，碩士研究生，主要研究方向：計算機軟件與理論；朱美正(1965-)，男，研究員級高級工程師，主要研究方向：地理信息系統平臺技術與數據共享；初寧(1972-)，男，高級工程師，主要研究方向：指揮自動化；楊崗(1972-)，男，主要研究方向：指揮自動化。

本文著錄格式：賀夢潔，朱美正，初寧，等. 基于Spark平臺的地理數據并行裝載技術[J]. 軟件，2016，37（12）：63-68