基于Spark的大規(guī)模天文數(shù)據(jù)天區(qū)覆蓋生成算法

熊聰聰，田祖宸，趙 青，馮 闊，崔辰州

(1. 天津科技大學(xué)計算機科學(xué)與信息工程學(xué)院，天津 300457；2. 中國科學(xué)院國家天文臺，北京 100012)

近年來科技的發(fā)展大大提高了天文數(shù)據(jù)的采集能力，各波段的數(shù)據(jù)量呈指數(shù)級增長，天文學(xué)逐漸走向全波段巡天的“大數(shù)據(jù)”時代[1]．如此龐大的數(shù)據(jù)量，對傳統(tǒng)科學(xué)計算技術(shù)處理大規(guī)模天文數(shù)據(jù)的做法提出了新的挑戰(zhàn)．一般而言，天文數(shù)據(jù)的計算主要分為數(shù)據(jù)導(dǎo)入、數(shù)據(jù)預(yù)處理和數(shù)據(jù)計算三個階段，如果面對大規(guī)模天文數(shù)據(jù)，在預(yù)處理階段對數(shù)據(jù)沒有進(jìn)行合理的安排，會降低數(shù)據(jù)計算階段的執(zhí)行效率，制約天文發(fā)現(xiàn)的進(jìn)展．天區(qū)覆蓋是望遠(yuǎn)鏡所拍攝區(qū)域在天區(qū)上的一種表征方式，它是將大量的星體記錄以一定的規(guī)則對雜亂無序的天體數(shù)據(jù)重新整理、歸檔的過程，對后續(xù)的數(shù)據(jù)查詢、處理、分析等工作具有重要的支持作用．但望遠(yuǎn)鏡拍攝能力的逐漸提升帶來了龐大的數(shù)據(jù)量，如何提高天區(qū)覆蓋計算的效率成為另一重要問題．

天文計算具有數(shù)據(jù)量大、I/O敏感性、內(nèi)存占用高等特點，早期的計算通常在超級計算機或高性能計算機集群上完成的[2]，但這種方案費用較高，而且程序設(shè)計復(fù)雜．云計算的出現(xiàn)讓研究人員得以用廉價、高效的方式處理海量數(shù)據(jù)，其中Spark框架憑借可以在內(nèi)存中進(jìn)行計算的優(yōu)勢，具有比 Hadoop MapReduce更高的計算效率，為天文大數(shù)據(jù)處理提供了前所未有的潛力．近年來已有一些天文研究工作使用Spark分布式計算框架處理大規(guī)模天文數(shù)據(jù)，如利用決策樹實現(xiàn)星系的分類[3]、利用聚類算法實現(xiàn)光譜分類[4]、還有結(jié)合 Hadoop與 Spark利用聚類算法實現(xiàn)了星象圖片分類[5]．此外，從基于 Spark的天文研究的軟件平臺 AstroSpark[6]、高性能計算機集群 SDSC Comet[7]中也不難看出，分布式計算技術(shù)已成為海量數(shù)據(jù)下進(jìn)行天文研究工作的一種發(fā)展趨勢．HEALPix作為一種天文學(xué)常用的天區(qū)索引方法，被應(yīng)用在了如星表交叉證認(rèn)[8]、天文圖像顯示[9]等多種天文相關(guān)的場景中．由此可見，HEALPix與Spark的結(jié)合尤其適用于處理大規(guī)模天文數(shù)據(jù)．但是，現(xiàn)有的研究多是為了解決某一種天文問題，或針對計算環(huán)節(jié)進(jìn)行分析，而對天文原始數(shù)據(jù)的預(yù)處理及歸檔方面卻鮮有提及，實驗也只是直接讀取原始數(shù)據(jù)并進(jìn)行計算．通常情況下，原始天文數(shù)據(jù)往往是分散、無序存放的，而對這種雜亂數(shù)據(jù)頻繁讀取會大大降低計算效率．

鑒于此，本文在借鑒前人工作的基礎(chǔ)上，圍繞對原始天文數(shù)據(jù)的預(yù)處理及歸檔，結(jié)合文獻(xiàn)[10]和文獻(xiàn)[11]中介紹的層次化索引思想，提出一種基于 Spark天區(qū)覆蓋生成的數(shù)據(jù)預(yù)處理與歸檔方法：在HEALPix分層索引的基礎(chǔ)上，將天文數(shù)據(jù)層次化、分塊、連續(xù)存放，從而提升后期交叉證認(rèn)、漏源監(jiān)測等天文計算中對數(shù)據(jù)進(jìn)行訪問、處理的效率；利用Spark的快速計算及迭代計算的特點，將天區(qū)覆蓋生成方法設(shè)計成一系列彈性分布式數(shù)據(jù)集的轉(zhuǎn)換操作，使其可以在并行環(huán)境下處理大規(guī)模天文數(shù)據(jù)．同時，索引的結(jié)果還可以用于數(shù)據(jù)可視化，方便研究人員直觀了解各個星表中的天文數(shù)據(jù)在天區(qū)上的分布情況.

1 天區(qū)覆蓋生成算法

1.1 HEALPix層次化球面索引

天文數(shù)據(jù)以每一條天體的信息為單位，如果在天文計算中以詞為單位逐條處理，過細(xì)的粒度會大大降低對數(shù)據(jù)訪問查詢的效率，消耗不必要的資源．針對這一問題，可以采用對數(shù)據(jù)進(jìn)行分塊的方法，適當(dāng)增加處理單位數(shù)據(jù)的粒度，這樣在抽取數(shù)據(jù)的時候可以分段讀取，本文利用了一種天文領(lǐng)域常見的球面索引方法HEALPix(hierarchical equal area ISO latitude pixelation)[12]．

HEALPix根據(jù)星體坐標(biāo)和區(qū)塊塊號之間的映射關(guān)系，采用四叉樹結(jié)構(gòu)，以層層遞歸的方式對天球進(jìn)行等面積劃分(圖 1)．其具有層次化、等面積和等緯度分布等特點．HEALPix在初始層級將天球劃分成12個面積相等的基準(zhǔn)球面四邊形，每一個四邊形被看作一個 pixel(這里稱為“區(qū)塊”，下同)，從 0至11為其編號．到了第二層級，每個四邊形再被等分為4個子四邊形，隨著層級深度的增加，每層的四邊形會不斷地被細(xì)分．最終到第 k層時，對天球基準(zhǔn)四邊形任一邊細(xì)分的次數(shù)即為當(dāng)前索引的分辨率，用參數(shù)Nside表示(Nside＝2,k)[12]．

圖1 HEALPix天球分區(qū)示意圖Fig. 1 HEALPix partition of the sphere

HEALPix具有 2種編碼規(guī)則：RING和NESTED. 其中 RING是按照自西向東、自北向南的順序?qū)^(qū)塊編號；NESTED是以層次化遞歸的方式對區(qū)塊編號．因順序編碼的RING規(guī)則不適應(yīng)于天區(qū)覆蓋生成的迭代操作，這里采用NESTED編碼方法，其編碼形式如圖 2所示．每一層級的區(qū)塊都會被分配唯一的二進(jìn)制編碼，每一個區(qū)塊的塊號由父塊號和當(dāng)前層級編碼兩部分組成．當(dāng)父塊(記為 highOrder)被分為 4個子塊后，每個子塊根據(jù)其位置會被賦予00、01、10或11的二進(jìn)制編號(記為lowOrder)，也就是當(dāng)前層級編碼，父級塊號作為前綴，當(dāng)前層級編碼作為后綴，即執(zhí)行了lowOrder?2&highOrder移位操作，形成了新層級的編碼．隨著層數(shù)深度的增加，對應(yīng)層級的塊號位數(shù)也會越來越大．如果以k表示劃分的層級，那么 Nside＝12×22k表示當(dāng)前層級可以最多劃分出的區(qū)塊數(shù)．

圖2 HEALPix索引及NESTED編碼方式示意圖(Nside＝2)Fig. 2 HEALPix index and NESTED numbering schemes(Nside＝2)

1.2 天區(qū)覆蓋生成算法的基本原理

從最低層級k開始，利用上文提到的位運算對當(dāng)前層每個區(qū)塊的塊號進(jìn)行分割操作，獲得相應(yīng)的父塊塊號(highOrder)和子塊塊號(lowOrder)．當(dāng)所有區(qū)塊處理完畢后，將相同父塊塊號的區(qū)塊聚合在一起，分析相同父塊塊號下的子塊分布情況，如果這些子塊中的星體覆蓋了 00、01、10、11位置時，當(dāng)前區(qū)域內(nèi)的所有區(qū)塊的原始塊號全部由父級塊號代替，同時這些區(qū)塊會被保留至下一次迭代操作．如果這些子塊沒有滿足上一條件則保留原始塊號，舍棄并進(jìn)行其他處理，不代入下一次迭代．如此往復(fù)，每一次迭代后滿足條件的數(shù)據(jù)會聚集起來，成為下一次迭代的輸入，這一迭代操作持續(xù)到?jīng)]有可聚集的數(shù)據(jù)為止．圖 3以一組第 2層級下的小規(guī)模數(shù)據(jù)為例展示了這一變換過程．

圖3 天區(qū)覆蓋生成過程示意圖Fig. 3 The process of sky coverage generation

圖 3(a)的深色區(qū)域表示初始數(shù)據(jù)在第 2級的區(qū)塊覆蓋情況，圖 3(b)與圖 3(c)實線區(qū)域為滿足條件聚合后的區(qū)塊，即原4個子塊的HEALPix,ID更新為父塊的 HEALPix ID，虛線表示為無法繼續(xù)聚合但被保留的區(qū)塊．當(dāng)算法執(zhí)行到第 0級或者無法聚合時結(jié)束．

1.3 位運算

在聚合操作中主要通過二進(jìn)制的移位操作獲得子塊自身以及對應(yīng)的父塊信息．以層級為 2、塊號為19的區(qū)塊為例，對應(yīng)的二進(jìn)制表示為 010011，那么從低兩位11就可以知道其作為上一級的子塊時的位置，而剩余的高位 0100即為對應(yīng)上一級區(qū)塊的塊號．以此類推，如果再一次進(jìn)行移位操作，就可以得知 19這個區(qū)塊在0和1層級時對應(yīng)的塊號為1和4．具體的對應(yīng)關(guān)系見表1．

表1 某個區(qū)塊的層級對應(yīng)關(guān)系表(以層級2，塊號19為例)Tab. 1 Hierarchical correspondence of a pixel(No.,19，level 2)

2 基于Spark的算法設(shè)計與實現(xiàn)

對天文數(shù)據(jù)的處理量通常比較龐大，如果在單臺計算機上進(jìn)行處理，往往受存儲空間和處理能力的限制．由于本算法中的迭代操作會產(chǎn)生大量中間數(shù)據(jù)，如果使用Hadoop MapReduce框架執(zhí)行這一任務(wù)，系統(tǒng)會將中間變量存儲在磁盤中并產(chǎn)生大量的讀寫操作．由于磁盤的存取速度慢，頻繁的磁盤讀寫操作會制約數(shù)據(jù)的處理效率．因此，本算法使用 Spark作為計算框架．

天區(qū)覆蓋生成算法在 Spark平臺下的實現(xiàn)主要分為 2個階段，算法中涉及到的 RDD(Resilient Distributed Datasets)及算子如圖 4所示，具體實現(xiàn)代碼可以從網(wǎng)站(https://github.com/ZuChen93/Cross-Match)獲得．

2.1 第1階段

數(shù)據(jù)預(yù)處理．首先用 textFile算子將原始數(shù)據(jù)從HDFS中導(dǎo)入，轉(zhuǎn)換為RDD，再用map算子根據(jù)天體的赤經(jīng) ra、赤緯 dec添加 HealPix索引．由于從網(wǎng)站獲得的天文數(shù)據(jù)中未含有 HEALPix索引信息，所以需要根據(jù)星體的坐標(biāo)逐條生成某一給定層級 k下的HEALPix索引編號，k根據(jù)數(shù)據(jù)密度或者天區(qū)覆蓋的最小精度決定．此外，由于原始數(shù)據(jù)是未經(jīng)壓縮的文本數(shù)據(jù)，在 RDD中直接以字符串形式處理會占用大量內(nèi)存空間，而且除坐標(biāo)以外的大部分信息在實際處理過程中并未參與運算．為了提高數(shù)據(jù)的傳輸效率，在添加 HEALPix索引之后，將每條天體數(shù)據(jù)中除索引以外的全部信息以新生成的唯一ID替代．

圖4 天區(qū)覆蓋生成算法流程圖Fig. 4 Flow chart of sky coverage generation

2.2 第2階段

根據(jù)天區(qū)分布生成算法的原理，對數(shù)據(jù)執(zhí)行迭代聚合操作，并根據(jù)條件對數(shù)據(jù)進(jìn)行過濾．在本階段采用了多個 map操作，用于根據(jù)計算需要變換鍵值．combineByKey算子作為核心操作，主要用于將相同 highOrder的數(shù)據(jù)整合為一條記錄，并統(tǒng)計lowOrder的分布情況．最后使用 filter算子，當(dāng)lowOrder完全分布在 00、01、10、11的時候，對應(yīng)highOrder的記錄可以用于下一次迭代，反之會被剔除并保存至HDFS中．

需要說明的是，算法在迭代進(jìn)行前添加了一個longAccumulator累加器，用于統(tǒng)計每次迭代完成后可以繼續(xù)聚合的數(shù)據(jù)條數(shù)．當(dāng)數(shù)值為 0時，說明沒有可以繼續(xù)處理的數(shù)據(jù)，迭代終止．

另外，本階段會有一些需要重用的 RDD數(shù)據(jù)，如combineByKey執(zhí)行后生成的RDD會被filter算子調(diào)用兩次．這里需要用到Spark中特有的持久化特性，即在RDD需要被頻繁使用的時候，使用cache持久化操作將數(shù)據(jù)緩存至內(nèi)存中，這樣如果后面需要再次調(diào)用這組數(shù)據(jù)的話，就可以直接到內(nèi)存中訪問，從而加快對此 RDD訪問的速度．如果沒有這一操作的話，Spark中的 lazy特性會使RDD在每次被調(diào)用的時候，都要根據(jù)它的 DAG 映射重新開始計算[13]，這樣不但占用系統(tǒng)資源，還會降低程序運行效率．算法中對類似的RDD都進(jìn)行了持久化處理．

3 實驗與分析

3.1 數(shù)據(jù)集與集群配置情況

為了驗證算法的性能，選取在中國虛擬天文臺網(wǎng)站(http://casdc.china-vo.org/mirror/2MASS)中公開的2,MASS星表中Point Source Catalog(PSC)數(shù)據(jù)集進(jìn)行實驗．共 152,GB數(shù)據(jù)中包含 4.7億條星體信息，以文本形式存儲，每一行數(shù)據(jù)表示為一個星體，每一行數(shù)據(jù)中的各個屬性信息以豎線分割．

實驗在阿里云的 E-MapReduce平臺下進(jìn)行，集群版本是 EMR-3.4.2(Spark 2.1.1，Hadoop 2.7.2)，每個節(jié)點的配置為“通用型 n2，4核處理器，16,GB內(nèi)存，SSD 云盤 80,GB×4，series Ⅲ”，1個 Master節(jié)點，多個 Core節(jié)點，集群資源使用 YARN進(jìn)行托管．實驗數(shù)據(jù)預(yù)先存儲至集群下的 HDFS文件系統(tǒng)中，Spark以 YARN-client模式運行，通過 sparksubmit中的 num-executors參數(shù)調(diào)整計算節(jié)點數(shù)，以模擬算法的并行執(zhí)行情況．

3.2 初始層級k對天區(qū)覆蓋生成結(jié)果的影響

1.1節(jié)已提到Nside＝2,k是影響HEALPix對天區(qū)細(xì)分精度的首要因素，其中 k代表細(xì)分的層數(shù)．為了測試這一參數(shù)是否會對天區(qū)覆蓋生成算法的結(jié)果有影響，選定計算節(jié)點數(shù)為 64，令 k取值分別為 2，4，6，…，20，實驗結(jié)果如圖5所示．

圖5 初始層級k與算法生成結(jié)果的關(guān)系Fig. 5 Relation between k and results of algorithm

圖中的“層級分布”表示數(shù)據(jù)從初始層級 k不斷聚合，直到程序結(jié)束時，能夠聚合出數(shù)據(jù)的層數(shù)分布；“聚合層深”表示程序結(jié)束時的層級距離初始層級 k的深度．可以看出：當(dāng) 2＜k＜10時，“聚合層深”與k相等，其中“層級分布”為1的情況表明數(shù)據(jù)全部聚合到了第0層；當(dāng)k＞18時，“聚合層深”為 0，表示數(shù)據(jù)完全無法聚合，全部數(shù)據(jù)都停留在了初始 k層．以上兩種情況都不是最理想的聚合結(jié)果．而 k的取值為 10和 12時，數(shù)據(jù)的劃分程度更好，k取10的時候數(shù)據(jù)的聚合結(jié)果最理想，且在不同層級間均有所分布．通過分析實驗結(jié)果猜測當(dāng)k取值過低時，由于區(qū)塊范圍過大，大量數(shù)據(jù)都集中在同一區(qū)塊中；而 k取值過大時，由于區(qū)塊過小，星體分散嚴(yán)重，大量數(shù)據(jù)在最低層級就無法向上聚合．實際情況下k的取值除了要考慮處理時間和層級分布外，還應(yīng)根據(jù)數(shù)據(jù)歸檔的需要設(shè)定．

3.3 計算節(jié)點數(shù)對算法執(zhí)行時間的影響

為了比較集群節(jié)點數(shù)對計算效率的影響，分別在1、2、4、8、16、32、64 計算節(jié)點數(shù)下用相同數(shù)據(jù)測試了算法的性能，初始層級k定為10．實驗結(jié)果如圖6所示．

圖6 算法在不同節(jié)點下的耗時情況(k＝10)Fig. 6 Time elapsed on different nodes(k＝10) with the algorithm

從圖 6可以看出：隨著節(jié)點數(shù)的增加，耗時越來越少；但是從節(jié)點數(shù)為16開始，增速逐漸放緩．在本算法的迭代計算中，reduce算子會在集群中產(chǎn)生大量的數(shù)據(jù)交換操作，這種操作會造成節(jié)點間的通信更頻繁，從而產(chǎn)生大量的網(wǎng)絡(luò)開銷，并且隨著計算節(jié)點數(shù)的增加而網(wǎng)絡(luò)開銷越來越大，會在整個程序執(zhí)行流程中占據(jù)較多時間．這可能是造成計算節(jié)點達(dá)到 16后，加速比逐漸放緩的主要因素．由此可見：合理地增加計算節(jié)點數(shù)可以加快處理數(shù)據(jù)的速度，但是要想在增加計算節(jié)點后獲得更好的加速比，需要進(jìn)一步優(yōu)化算法或者集群資源分配；可以根據(jù)需要，人為規(guī)定聚合結(jié)束時停止的層級，使得聚合操作提前結(jié)束，避免耗費不必要的計算資源．

4 結(jié) 語

本文探討了一種基于 Spark的天區(qū)覆蓋生成算法，在 HEALPix分層索引的基礎(chǔ)上，對天文數(shù)據(jù)層次化、分塊、連續(xù)存放．實驗證明，本算法借助 Spark可以在較短的時間內(nèi)處理大規(guī)模數(shù)據(jù)．將整理后的數(shù)據(jù)用于后期相關(guān)天文計算，可明顯提升訪問、處理數(shù)據(jù)的效率．此外，算法處理后的數(shù)據(jù)還可以用于數(shù)據(jù)可視化，繪制天區(qū)覆蓋圖，以便直觀地了解天文數(shù)據(jù)在天區(qū)上的分布情況．

分布式計算的性能受算法優(yōu)化、參數(shù)調(diào)優(yōu)等多種因素制約，例如文獻(xiàn)[6]提到Spark對數(shù)據(jù)塊劃分的合適與否會影響計算的并行性，因此還可以從以上方面繼續(xù)優(yōu)化本文算法．