支持異構集群并行的高能物理數據處理系統

霍 菁,雷曉鳳,李 強,孫功星

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

·專欄·

支持異構集群并行的高能物理數據處理系統

霍 菁1,2,雷曉鳳1,2,李 強1,2,孫功星1

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

傳統集群計算系統無法充分利用本地磁盤的存儲資源和I/O,大量網絡I/O成為系統瓶頸,導致資源利用率降低,并造成高昂的存儲和網絡成本。使用Hadoop處理分析作業可有效利用本地磁盤存儲和I/O資源,而集群資源統一管理工具Mesos則使用輕量化的設計和高效的通信機制,能在不同計算集群之間動態共享集群資源。為此,分析高能物理數據處理的特點,利用Mesos構建異構集群間資源共享的高能物理實驗數據處理系統,實現Torque/Maui和Hadoop集群的集成。測試結果表明,該系統能夠在集群間動態分配集群資源,并利用本地存儲和磁盤I/O顯著降低網絡I/O,提高集群資源利用率。

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

1 概述

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

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

本文分析高能物理實驗數據處理的不同作業類型和特點,以及Torque/Maui集群和Hadoop集群的特性,以提高集群資源利用率和利用本地存儲空間為目標,利用集群資源管理工具Mesos集成Torque/ Maui和Hadoop框架,實現一個支持混合集群并行的高能物理實驗數據處理平臺,并應用BESIII高能物理實驗數據進行初步評估。

2 研究背景

2.1 BESIII實驗數據處理的流程和作業類型

BESIII數據處理的作業類型主要有模擬作業、重建作業和分析作業3種。

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

圖1 BESIII實驗數據處理的作業類型和步驟

2.2 Mesos集群資源統一管理框架

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

圖2 Mesos架構

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

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

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

3 系統設計與實現

經過調研,本文設計了一種新的系統架構,在同一個物理集群上并行運行 Torque/Maui集群和Hadoop集群,使用Mesos進行集群資源管理和分配。系統架構如圖3所示。

圖3 高能物理數據處理系統框架

該系統使用PC服務器和高性能服務器作為計算節點,使用由Lustre管理的磁盤陣列和由計算節點本地磁盤構建的分布式文件系統HDFS提供數據存儲。使用Mesos作為系統資源調度管理器。

該系統在應用層提供模擬和重建作業,分析作業,事例預篩選等應用。另外,提供給用戶命令行接口和Web接口來提交和查詢作業。本文使用AFS進行用戶認證與權限管理,使用Puppet[11]和Ganglia[12]進行集群管理和監控。

系統的設計和實現的關鍵技術和本文的主要工作集中在 Torque/Maui和 Mesos的集成部分。Mesos和Hadoop的集成和資源分配策略采用了開源社區提供的代碼。

3.1 Torque/Maui與Mesos的集成

Torque/Maui與Mesos的集成主要分為2個部分:(1)資源調度器Framework Scheduler的設計和實現;(2)作業執行器Executor的設計和實現。

3.1.1 Framework Scheduler的設計與實現

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

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

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

新系統中Maui和Framework Scheduler的架構設計和資源、作業的調度流程和信息交互如圖4中細實線和虛線所示。

圖4 Torque/Maui與Mesos的集成架構

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

圖5 Framework Scheduler中的資源和作業信息

3.1.2 Executor的設計與實現

Executor的功能是獲取節點上Mesos分配給它的資源,對作業進行操作和更新作業狀態。在作業執行的過程中,Mesos需要監控作業的運行狀態,并根據作業狀態決定對該作業的操作和釋放資源。

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

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

3.2 信息交互格式定義

為簡化消息處理機制,使2個系統之間的信息交互更簡單高效且易讀,本文定義了一個結構體request來傳遞消息:

結構體request中各字段的名稱和含義如表1所示,其中,req_type指定消息類型;hostname指定消息相關的節點名稱;message指定消息內容;value指定消息的值。表1中的示例為:更新節點host01上的作業job0103的狀態為6。

表1 結構體request各字段定義及示例

4 系統評估測試

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

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

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

圖6 BESIII集群CPU使用情況

圖7顯示的是采用新的系統架構后集群的整體CPU使用情況和CPU在2個框架間的動態分配情況。本文設置2個集群的資源使用比例為Hadoop: Torque=5:7。由圖7可見,當沒有Hadoop作業時, Torque集群可以獨占集群的CPU資源,當Hadoop中有作業時,資源占用比例逐漸變為5:7。當Hadoop中作業數量減少時,Torque占用資源增加,當Torque作業數量減少時,Hadoop資源占用量增加,集群資源可以根據2個集群的作業數量動態地進行調整。

圖7 Hadoop與Torque占用資源比例

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

圖8 測試集群網絡I/O速度

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

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

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

5 結束語

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

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

[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的高能物理數據分析系統[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

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

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

2014-02-17

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

中文引用格式:霍 菁,雷曉鳳,李 強,等.支持異構集群并行的高能物理數據處理系統[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.