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

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

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

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

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

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

1.1 HEALPix層次化球面索引

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

HEALPix根據(jù)星體坐標(biāo)和區(qū)塊塊號(hào)之間的映射關(guān)系，采用四叉樹(shù)結(jié)構(gòu)，以層層遞歸的方式對(duì)天球進(jìn)行等面積劃分(圖 1)．其具有層次化、等面積和等緯度分布等特點(diǎn)．HEALPix在初始層級(jí)將天球劃分成12個(gè)面積相等的基準(zhǔn)球面四邊形，每一個(gè)四邊形被看作一個(gè) pixel(這里稱(chēng)為“區(qū)塊”，下同)，從 0至11為其編號(hào)．到了第二層級(jí)，每個(gè)四邊形再被等分為4個(gè)子四邊形，隨著層級(jí)深度的增加，每層的四邊形會(huì)不斷地被細(xì)分．最終到第 k層時(shí)，對(duì)天球基準(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ū)塊編號(hào)；NESTED是以層次化遞歸的方式對(duì)區(qū)塊編號(hào)．因順序編碼的RING規(guī)則不適應(yīng)于天區(qū)覆蓋生成的迭代操作，這里采用NESTED編碼方法，其編碼形式如圖 2所示．每一層級(jí)的區(qū)塊都會(huì)被分配唯一的二進(jìn)制編碼，每一個(gè)區(qū)塊的塊號(hào)由父塊號(hào)和當(dāng)前層級(jí)編碼兩部分組成．當(dāng)父塊(記為 highOrder)被分為 4個(gè)子塊后，每個(gè)子塊根據(jù)其位置會(huì)被賦予00、01、10或11的二進(jìn)制編號(hào)(記為lowOrder)，也就是當(dāng)前層級(jí)編碼，父級(jí)塊號(hào)作為前綴，當(dāng)前層級(jí)編碼作為后綴，即執(zhí)行了lowOrder?2&highOrder移位操作，形成了新層級(jí)的編碼．隨著層數(shù)深度的增加，對(duì)應(yīng)層級(jí)的塊號(hào)位數(shù)也會(huì)越來(lái)越大．如果以k表示劃分的層級(jí)，那么 Nside＝12×22k表示當(dāng)前層級(jí)可以最多劃分出的區(qū)塊數(shù)．

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

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

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

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

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

1.3 位運(yùn)算

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

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

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

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

天區(qū)覆蓋生成算法在 Spark平臺(tái)下的實(shí)現(xiàn)主要分為 2個(gè)階段，算法中涉及到的 RDD(Resilient Distributed Datasets)及算子如圖 4所示，具體實(shí)現(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)逐條生成某一給定層級(jí) k下的HEALPix索引編號(hào)，k根據(jù)數(shù)據(jù)密度或者天區(qū)覆蓋的最小精度決定．此外，由于原始數(shù)據(jù)是未經(jīng)壓縮的文本數(shù)據(jù)，在 RDD中直接以字符串形式處理會(huì)占用大量?jī)?nèi)存空間，而且除坐標(biāo)以外的大部分信息在實(shí)際處理過(guò)程中并未參與運(yùn)算．為了提高數(shù)據(jù)的傳輸效率，在添加 HEALPix索引之后，將每條天體數(shù)據(jù)中除索引以外的全部信息以新生成的唯一ID替代．

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

2.2 第2階段

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

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

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

3 實(shí)驗(yàn)與分析

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

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

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

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

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

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

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

3.3 計(jì)算節(jié)點(diǎn)數(shù)對(duì)算法執(zhí)行時(shí)間的影響

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

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

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

4 結(jié) 語(yǔ)

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

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