支持異構(gòu)集群并行的高能物理數(shù)據(jù)處理系統(tǒng)

霍 菁,雷曉鳳,李 強(qiáng),孫功星

(1.中國科學(xué)院高能物理研究所,北京100049;2.中國科學(xué)院大學(xué),北京100049)

·專欄·

支持異構(gòu)集群并行的高能物理數(shù)據(jù)處理系統(tǒng)

霍 菁1,2,雷曉鳳1,2,李 強(qiáng)1,2,孫功星1

(1.中國科學(xué)院高能物理研究所,北京100049;2.中國科學(xué)院大學(xué),北京100049)

傳統(tǒng)集群計算系統(tǒng)無法充分利用本地磁盤的存儲資源和I/O,大量網(wǎng)絡(luò)I/O成為系統(tǒng)瓶頸,導(dǎo)致資源利用率降低,并造成高昂的存儲和網(wǎng)絡(luò)成本。使用Hadoop處理分析作業(yè)可有效利用本地磁盤存儲和I/O資源,而集群資源統(tǒng)一管理工具M(jìn)esos則使用輕量化的設(shè)計和高效的通信機(jī)制,能在不同計算集群之間動態(tài)共享集群資源。為此,分析高能物理數(shù)據(jù)處理的特點,利用Mesos構(gòu)建異構(gòu)集群間資源共享的高能物理實驗數(shù)據(jù)處理系統(tǒng),實現(xiàn)Torque/Maui和Hadoop集群的集成。測試結(jié)果表明,該系統(tǒng)能夠在集群間動態(tài)分配集群資源,并利用本地存儲和磁盤I/O顯著降低網(wǎng)絡(luò)I/O,提高集群資源利用率。

高能物理;集群資源管理;資源共享;Mesos工具;Hadoop平臺;Torque/Maui系統(tǒng)

1 概述

高能物理實驗是一項龐大的系統(tǒng)工程,數(shù)據(jù)處理是其中的關(guān)鍵步驟。高能物理實驗產(chǎn)生的數(shù)據(jù)量非常龐大,歐洲核子中心(CERN)使用的大型強(qiáng)子對撞機(jī)LHC(Large Hadron Collider),每年產(chǎn)生約25 PB的實驗數(shù)據(jù);北京正負(fù)電子對撞機(jī)(BEPCII)和北京譜儀(BESIII)在經(jīng)過改造以后,2012年產(chǎn)生的數(shù)據(jù)量超過了過去幾年的總和,總規(guī)模超過3 PB。

高能物理實驗數(shù)據(jù)處理有多種不同的類型作業(yè)。傳統(tǒng)集群使用的批作業(yè)處理系統(tǒng)Torque[1]/Maui[2]采用計算資源和數(shù)據(jù)存儲相分離的系統(tǒng)架構(gòu),適用于蒙特卡洛模擬等計算密集型應(yīng)用。而Hadoop[3]平臺使用MapReduce[4]的方式,能夠利用計算節(jié)點本地磁盤組建HDFS[5],使分析作業(yè)讀取本地磁盤數(shù)據(jù),減輕網(wǎng)絡(luò)I/O壓力,更適合數(shù)據(jù)密集型的分析作業(yè)[6]。由于高能物理數(shù)據(jù)處理具有隨時間分布的特征,2個集群的作業(yè)分布密度不同,在集群間動態(tài)共享資源能夠有效提高集群資源利用率。集群資源統(tǒng)一管理工具M(jìn)esos[7],使用輕量化的設(shè)計和高效的通信機(jī)制,可以在不同的集群之間動態(tài)共享資源,比虛擬機(jī)技術(shù)更加高效。目前Mesos已支持多種應(yīng)用,例如Hadoop,Spark,Hypertable等。已有多家著名公司使用Mesos管理集群資源,包括國外著名公司Twitter和國內(nèi)著名視頻網(wǎng)站愛奇藝等,但在高能物理實驗中還沒有應(yīng)用。

本文分析高能物理實驗數(shù)據(jù)處理的不同作業(yè)類型和特點,以及Torque/Maui集群和Hadoop集群的特性,以提高集群資源利用率和利用本地存儲空間為目標(biāo),利用集群資源管理工具M(jìn)esos集成Torque/ Maui和Hadoop框架,實現(xiàn)一個支持混合集群并行的高能物理實驗數(shù)據(jù)處理平臺,并應(yīng)用BESIII高能物理實驗數(shù)據(jù)進(jìn)行初步評估。

2 研究背景

2.1 BESIII實驗數(shù)據(jù)處理的流程和作業(yè)類型

BESIII數(shù)據(jù)處理的作業(yè)類型主要有模擬作業(yè)、重建作業(yè)和分析作業(yè)3種。

實驗數(shù)據(jù)由對撞機(jī)產(chǎn)生,生成原始數(shù)據(jù)(RAW Data)后存儲在 Lustre[8]中。為了驗證數(shù)據(jù)的正確性,需要一定量的模擬數(shù)據(jù)來進(jìn)行對比,因此,使用蒙特卡洛模擬產(chǎn)生模擬數(shù)據(jù)。原始數(shù)據(jù)和模擬數(shù)據(jù)在經(jīng)過重建(Reconstruction)后,生成可以供物理學(xué)家進(jìn)行分析的DST數(shù)據(jù)。對DST數(shù)據(jù)的分析是典型的一次寫多次讀的應(yīng)用。不同的物理學(xué)家使用自己編寫的分析程序分析DST數(shù)據(jù),挑選自己感興趣的物理事例,最后生成圖表等結(jié)果。本文設(shè)計的混合集群的目的是使用傳統(tǒng)集群計算系統(tǒng)處理主要消耗CPU資源的模擬和重建作業(yè),使用Hadoop處理主要消耗I/O資源的分析作業(yè),如圖1所示。

圖1 BESIII實驗數(shù)據(jù)處理的作業(yè)類型和步驟

2.2 Mesos集群資源統(tǒng)一管理框架

Mesos同大多數(shù)分布式系統(tǒng)一樣,采用的是Master-Slave的架構(gòu),如圖2所示。該架構(gòu)主要由4個部分構(gòu)成:Master,Slave,Framework和Executor。Master負(fù)責(zé)資源管理和分配,Slave負(fù)責(zé)匯報資源和管理本地資源,并負(fù)責(zé)啟動作業(yè)執(zhí)行器Executor。在Mesos中,所有接入系統(tǒng)的分布式計算框架都成為Framework,這些框架在Slave節(jié)點上用來運行作業(yè)的執(zhí)行器稱為Executor。

圖2 Mesos架構(gòu)

Mesos的Master在設(shè)計上實現(xiàn)了輕量化,僅保存Framework和Slave的部分狀態(tài)信息,這些信息通過注冊的方式由Framework和Slave匯報給Master,因此Mesos使用ZooKeeper[9]解決Master的單點失效問題,提高系統(tǒng)容錯能力。Mesos的各組成部分之間通過LibProcess和Protocol Buffers組成的高效的通信系統(tǒng)。

在Mesos中,資源以Resource Offer的形式進(jìn)行調(diào)度。Resource Offer是一種基于資源數(shù)量的調(diào)度機(jī)制,不同于Hadoop中的基于Slot的調(diào)度機(jī)制, Mesos中的資源可以根據(jù)不同應(yīng)用的資源需求靈活的進(jìn)行資源分配。同時,Mesos采用一種支持多維資源向量的Max-Min Fair資源分配算法Dominant Resource Fairness[10],使不同應(yīng)用的不同作業(yè)需求能夠得到公平的資源分配。

實現(xiàn)集群間資源動態(tài)分配需要建立一個全局的資源和作業(yè)信息系統(tǒng),因此,任何一個分布式計算框架,如果想接入Mesos獲取資源運行作業(yè),需要開發(fā)2個組件:用于獲取Mesos分配的資源信息的資源調(diào)度器 Framework Scheduler和用于管理作業(yè)并向Mesos同步作業(yè)信息的作業(yè)執(zhí)行器Executor。

3 系統(tǒng)設(shè)計與實現(xiàn)

經(jīng)過調(diào)研,本文設(shè)計了一種新的系統(tǒng)架構(gòu),在同一個物理集群上并行運行 Torque/Maui集群和Hadoop集群,使用Mesos進(jìn)行集群資源管理和分配。系統(tǒng)架構(gòu)如圖3所示。

圖3 高能物理數(shù)據(jù)處理系統(tǒng)框架

該系統(tǒng)使用PC服務(wù)器和高性能服務(wù)器作為計算節(jié)點,使用由Lustre管理的磁盤陣列和由計算節(jié)點本地磁盤構(gòu)建的分布式文件系統(tǒng)HDFS提供數(shù)據(jù)存儲。使用Mesos作為系統(tǒng)資源調(diào)度管理器。

該系統(tǒng)在應(yīng)用層提供模擬和重建作業(yè),分析作業(yè),事例預(yù)篩選等應(yīng)用。另外,提供給用戶命令行接口和Web接口來提交和查詢作業(yè)。本文使用AFS進(jìn)行用戶認(rèn)證與權(quán)限管理,使用Puppet[11]和Ganglia[12]進(jìn)行集群管理和監(jiān)控。

系統(tǒng)的設(shè)計和實現(xiàn)的關(guān)鍵技術(shù)和本文的主要工作集中在 Torque/Maui和 Mesos的集成部分。Mesos和Hadoop的集成和資源分配策略采用了開源社區(qū)提供的代碼。

3.1 Torque/Maui與Mesos的集成

Torque/Maui與Mesos的集成主要分為2個部分:(1)資源調(diào)度器Framework Scheduler的設(shè)計和實現(xiàn);(2)作業(yè)執(zhí)行器Executor的設(shè)計和實現(xiàn)。

3.1.1 Framework Scheduler的設(shè)計與實現(xiàn)

在資源調(diào)度器Framework Scheduler的設(shè)計中,主要實現(xiàn)2個功能:(1)從Mesos處獲得可用資源信息,并將這些信息傳遞給 Torque的作業(yè)調(diào)度器 Maui; (2)Maui對分配到的資源進(jìn)行處理后,將需要執(zhí)行的作業(yè)信息傳遞給Mesos,以便分配資源和啟動作業(yè)。

在原系統(tǒng)中Maui使用Torque提供的API(定義在頭文件pbs_ifl.h中)從pbs_server中獲取進(jìn)行調(diào)度所需要的信息,包括:隊列信息QueueInfo,節(jié)點信息NodeInfo,作業(yè)信息JobInfo。然后對根據(jù)節(jié)點狀態(tài),隊列和作業(yè)的優(yōu)先級等信息,把作業(yè)跟可用資源進(jìn)行匹配,并將匹配的結(jié)果返回給 pbs_server,由pbs_server發(fā)命令到相應(yīng)slave節(jié)點上的pbs_mom來執(zhí)行作業(yè),流程如圖4中粗實線和細(xì)實線所示。

為了將Torque/Maui與Mesos集成,必須建立一個全局的資源及作業(yè)信息系統(tǒng),因此,修改Maui中的資源查詢函數(shù),使其改向資源調(diào)度器Framework Scheduler查詢可用的資源信息,并在作業(yè)調(diào)度結(jié)束后,將作業(yè)的調(diào)度結(jié)果反饋給資源調(diào)度器。資源調(diào)度器接收到作業(yè)信息后,由Mesos確定其需要的資源數(shù)量,為各節(jié)點上的pbs_mom分配資源。由于Maui是使用C語言編寫的,而Mesos使用C++編寫,因此本文使用Socket在2個程序間交換信息。

新系統(tǒng)中Maui和Framework Scheduler的架構(gòu)設(shè)計和資源、作業(yè)的調(diào)度流程和信息交互如圖4中細(xì)實線和虛線所示。

圖4 Torque/Maui與Mesos的集成架構(gòu)

由于Mesos資源分配拉(pull)的模式和Maui作業(yè)分配推(push)的模式有沖突,本文設(shè)計了一個Hashmap作為緩沖數(shù)組來接收Mesos分配的資源,在資源接受完畢后,由Maui查詢并進(jìn)行作業(yè)分配,將每個資源分配到的作業(yè)信息寫入數(shù)組中,作業(yè)信息用鄰接鏈表來保存。對于沒有分配作業(yè)的offer,調(diào)用offer.decline()方法回收資源,分配給另外的框架使用。資源數(shù)組和作業(yè)信息結(jié)構(gòu)如圖5所示。

圖5 Framework Scheduler中的資源和作業(yè)信息

3.1.2 Executor的設(shè)計與實現(xiàn)

Executor的功能是獲取節(jié)點上Mesos分配給它的資源,對作業(yè)進(jìn)行操作和更新作業(yè)狀態(tài)。在作業(yè)執(zhí)行的過程中,Mesos需要監(jiān)控作業(yè)的運行狀態(tài),并根據(jù)作業(yè)狀態(tài)決定對該作業(yè)的操作和釋放資源。

在Executor的設(shè)計中,只需要讓Executor啟動pbs_ mom,具體作業(yè)的執(zhí)行和操作由pbs_mom來進(jìn)行。pbs_mom在作業(yè)執(zhí)行的過程中,會向pbs_server匯報作業(yè)狀態(tài)信息,本文利用這個動作來更新Mesos中的作業(yè)狀態(tài)。在這里仍然使用Socket進(jìn)行進(jìn)程間的通信。

Executor在啟動作業(yè)時,為每一個作業(yè)都創(chuàng)建了一個線程負(fù)責(zé)更新該作業(yè)的狀態(tài),并通過一個全局變量的數(shù)組來保存正在運行的作業(yè)的狀態(tài)。pbs_ mom在修改作業(yè)狀態(tài)的同時,會通過Socket把作業(yè)的狀態(tài)發(fā)送給本地的Executor。Executor在接收到作業(yè)狀態(tài)信息后,更新數(shù)組中對應(yīng)作業(yè)的狀態(tài),并觸發(fā)相應(yīng)的作業(yè)線程去更新Mesos中的作業(yè)狀態(tài),或觸發(fā)完成作業(yè)釋放資源、殺掉作業(yè)等操作。Executor的結(jié)構(gòu)設(shè)計如圖4右下角所示,這樣的設(shè)計可以把2個系統(tǒng)之間的通信分散在各個計算節(jié)點上,保證系統(tǒng)的穩(wěn)定性。

3.2 信息交互格式定義

為簡化消息處理機(jī)制,使2個系統(tǒng)之間的信息交互更簡單高效且易讀,本文定義了一個結(jié)構(gòu)體request來傳遞消息:

結(jié)構(gòu)體request中各字段的名稱和含義如表1所示,其中,req_type指定消息類型;hostname指定消息相關(guān)的節(jié)點名稱;message指定消息內(nèi)容;value指定消息的值。表1中的示例為:更新節(jié)點host01上的作業(yè)job0103的狀態(tài)為6。

表1 結(jié)構(gòu)體request各字段定義及示例

4 系統(tǒng)評估測試

系統(tǒng)的實現(xiàn)基于 HDFS2.0,MapReduce0.20, Torque2.5.5,Maui3.2.6和Mesos0.14.0。通過部署在X86架構(gòu)的服務(wù)器,每個節(jié)點的配置為8× 2.4 GHz CPU,24 GB內(nèi)存,千兆以太網(wǎng)卡,2×2 TB的SATA硬盤,形成一個由1個服務(wù)器節(jié)點、6個計算節(jié)點組成的混合集群。

系統(tǒng)測試采用真實的BESIII實驗數(shù)據(jù)和真實的高能物理實驗數(shù)據(jù)處理軟件BOSS[13]和BEAN,運行真實的模擬作業(yè)、重建作業(yè)和分析作業(yè)作為測試程序。

圖6顯示的是Torque/Maui集群在運行真實物理作業(yè)的CPU資源使用情況,可以看到,CPU核平均有約20%左右的空置。

圖6 BESIII集群CPU使用情況

圖7顯示的是采用新的系統(tǒng)架構(gòu)后集群的整體CPU使用情況和CPU在2個框架間的動態(tài)分配情況。本文設(shè)置2個集群的資源使用比例為Hadoop: Torque=5:7。由圖7可見,當(dāng)沒有Hadoop作業(yè)時, Torque集群可以獨占集群的CPU資源,當(dāng)Hadoop中有作業(yè)時,資源占用比例逐漸變?yōu)?:7。當(dāng)Hadoop中作業(yè)數(shù)量減少時,Torque占用資源增加,當(dāng)Torque作業(yè)數(shù)量減少時,Hadoop資源占用量增加,集群資源可以根據(jù)2個集群的作業(yè)數(shù)量動態(tài)地進(jìn)行調(diào)整。

圖7 Hadoop與Torque占用資源比例

圖8顯示的是測試集群分別運行Torque/Maui和混合集群時的網(wǎng)絡(luò)I/O速度。由圖8可見,僅運行Torque/Maui時,6個計算節(jié)點的總I/O速度約為200 MB/s,平均每個節(jié)點33 MB/s。如果以現(xiàn)在BESIII集群1 000個節(jié)點的規(guī)模估算,總帶寬約為33 GB/s,所以BESIII集群的帶寬壓力很大。而運行混合集群時,由于Hadoop集群基本上都從本地讀取數(shù)據(jù),混合集群的網(wǎng)絡(luò)I/O速度下降了40%,因此大大降低了帶寬壓力。

圖8 測試集群網(wǎng)絡(luò)I/O速度

圖9顯示的是測試集群分別運行Torque/Maui和混合集群時的磁盤I/O速度。

圖9 測試集群本地磁盤I/O速度

可以看到,因此Torque集群運行時,作業(yè)數(shù)據(jù)全部從網(wǎng)絡(luò)讀取,所以磁盤I/O速度很低,共20 MB/s左右。但是Hadoop集群由于需要從本地磁盤讀取數(shù)據(jù),因此磁盤I/O速度較高,平均有200 MB/s左右。這個數(shù)值是圖8中網(wǎng)絡(luò)I/O下降數(shù)值的2倍,也說明了使用Hadoop處理分析作業(yè)時的數(shù)據(jù)讀取效率比Torque/ Maui集群要高。

5 結(jié)束語

本文在分析高能物理數(shù)據(jù)處理的實際需求和集群特點后,利用Mesos構(gòu)建了一個支持集群間資源動態(tài)分配的高能物理實驗數(shù)據(jù)處理框架。經(jīng)過初步測試,證明該系統(tǒng)架構(gòu)可以在不同的框架間動態(tài)地分配集群資源,有效降低網(wǎng)絡(luò)I/O壓力,利用計算節(jié)點本地磁盤擴(kuò)充系統(tǒng)存儲容量,顯著提高了資源利用率。

在實際應(yīng)用中會遇到更多問題,在資源分配上還需要進(jìn)一步改進(jìn),例如同一集群中的機(jī)器性能和配置上可能有較大差異,如何為不同的應(yīng)用分配合適的機(jī)器是資源分配時需要考慮的因素之一。

[1] Staples G.TORQUE Resource Manager[C]//Proceedings of 2006ACM/IEEE ConferenceonSupercomputing. New York,USA:ACM Press,2006.

[2] Adaptive Computing.Maui[EB/OL].[2014-02-15].http:// www.adaptivecomputing.com/products/open-source/maui/.

[3] Yahoo.Apache Hadoop[EB/OL].[2014-02-15]. http://hadoop.apache.org/.

[4] Dean J,GhemawatS.MapReduce:SimplifiedData Processing on Large Clusters[J].Communications of the ACM,2008,51(1):107-113.

[5] Shvachko K,Kuang H,Radia S,etal.The Hadoop Distributed File System[C]//Proceedings of the 26th IEEE Symposium on Mass Storage Systems and Technologies. Incline Village,USA:IEEE Press,2010:1-10.

[6] 臧冬松,霍 菁,梁 棟,等.基于MapReduce的高能物理數(shù)據(jù)分析系統(tǒng)[J].計算機(jī)工程,2014,40(2):1-5.

[7] Hindman B,Konwinski A,Zaharia M,et al.Mesos:A Platform for Fine-grained Resource Sharing in the Data Center[C]//Proceedings of NSDI’11.Berkeley,USA: USENIX Association,2011:22-22.

[8] Schwan P.Lustre:Building a File System for 1000-node Clusters[C]//Proceedings of 2003 Linux Symposium. Ottawa,Canada:[s.n.],2003:380-386.

[9] Hunt P,Konar M,Junqueira F P,et al.ZooKeeper:Waitfree Coordination forInternet-scale Systems[C]// Proceedings of 2010 USENIX Conference on USENIX Annual Technical Conference.[S.l.]:USENIX Association,2010:11.

[10] Ghodsi A,Zaharia M,Hindman B,et al.Dominant Resource Fairness:Fair Allocation of Multiple Resource Types[C]//Proceedings of NSDI’11.Berkeley,USA: USENIX Association,2011:323-336.

[11] Puppet Labs.What is Puppet?[EB/OL].[2014-02-15]. https://puppetlabs.com/puppet/what-is-puppet/.

[12] IBM.Ganglia Monitoring System[EB/OL].[2014-02-15]. http://ganglia.info/.

[13] Li Weidong,Liu Huaiming,Deng Ziyan,et al.The Offline Software for the BESIII Experiment[C]//Proceedings of CHEP’06.Mumbai,India:[s.n.],2006.

編輯 金胡考

High Energy Physics Data Processing System with Parallel Heterogeneous Clusters

HUO Jing1,2,LEI Xiaofeng1,2,LI Qiang1,2,SUN Gongxing1
(1.Institute of High Energy Physics,Chinese Academy of Sciences,Beijing 100049,China; 2.Graduate University of Chinese Academy of Sciences,Beijing 100049,China)

The traditional cluster computing system can not make best of the local disks and disk I/O resources, therefore the network becomes the bottleneck of the whole system.And this is the reason of low utilization of the cluster resources and high cost on data storage and network equipment.Using Hadoop to process analysis can significantly reduce the pressure on network I/O by using the local disks as a distributed file system.Mesos is a cluster resource manager with light-weight design and efficient communication mechanisms that can dynamically share resources among clusters.This paper introduces the features of High Energy Physics(HEP),data processing,presents a new HEP data processing system by using Mesos to provide dynamic resource sharing among clusters,and implements integration of Toruqe/Maui and Hadoop which can avoid the disadvantages.The test result shows that the new system can dynamic distribute the cluster resource,and reduce the network I/O,improve the resource utilization.

High Energy Physics(HEP);cluster resource management;resource sharing;Mesos tool;Hadoop platform; Toruqe/Maui system

1000-3428(2015)01-0001-05

A

TP391

10.3969/j.issn.1000-3428.2015.01.001

國家自然科學(xué)基金資助項目(11375223,11375221);國家自然科學(xué)基金A3前瞻計劃基金資助項目(61161140454)。

霍 菁(1985-),男,博士研究生,主研方向:分布式計算,集群資源管理;雷曉鳳、李 強(qiáng),博士研究生;孫功星,研究員。

2014-02-17

2014-03-20 E-mail:huojing@ihep.ac.cn

中文引用格式:霍 菁,雷曉鳳,李 強(qiáng),等.支持異構(gòu)集群并行的高能物理數(shù)據(jù)處理系統(tǒng)[J].計算機(jī)工程,2015, 41(1):1-5.

英文引用格式:Huo Jing,Lei Xiaofeng,Li Qiang,et al.High Energy Physics Data Processing System with Heterogeneous Clusters[J].Computer Engineering,2015,41(1):1-5.