面向高性能應用的MPI大數據處理

2019-01-07 12:16:34王鵬，周巖

計算機應用 2018年12期

關鍵詞：實驗

王鵬，周巖

(西南民族大學計算機科學與技術學院，成都 610225)(*通信作者電子郵箱22954936@qq.com)

0 引言

消息傳遞接口(Message Passing Interface, MPI)是一種基于信息傳遞的并行編程接口，在規模化集群環境中具備高效的并行運算與數據處理能力。作為計算密集型的高性能計算標準，MPI在氣候模擬、高能物理、生命科學、工業仿真、人工智能、大數據處理等高性能計算和科學計算領域應用廣泛[1]。但MPI缺乏與其自身計算系統深度融合的大數據管理機制，目前的數據集中存儲模式不便于系統實現“計算向存儲遷移”的大數據思想理念。計算過程中數據分發受網絡傳輸時長的制約，限制了MPI在大數據領域的應用擴展。

目前業界在處理海量數據時多采用分布式系統模式，無論是Lustre并行文件系統、并行虛擬文件系統(Parallel Virtual File System, PVFS)，還是通用并行文件系統(General Parallel File System, GPFS)。這些并行化系統各有特點，在不同領域均有相應應用。Lustre的可用性和擴展性不錯，但需要特殊存儲設備的支持，而且沒有實現分布式的元數據服務器管理。GPFS支持串行應用和并行應用，允許任何節點的并行應用同時訪問同一文件或不同文件。PVFS將數據文件存儲于集群的不同節點，多個客戶端可以同時訪問這些數據[2]。但這類系統本身都進行過封裝，應用程序在調用時只需知道文件邏輯地址，并不關心文件具體存儲位置，無法避免運算過程中大量數據的跨節點傳輸。

作為目前大數據應用領域最為廣泛的Hadoop，有一套主從式文件系統HDFS在底層支撐其Map/Reduce的數據處理功能。主從式基礎存儲和主從式數據處理構成Hadoop的基本架構模型。它的計算系統能夠通過HDFS快速定位文件存儲位置，這讓Hadoop方便地實現“計算向數據存儲位置的遷移”，從而大大提高系統計算效率，具備處理超大數據集(large data set)應用程序的能力[3]。

Hadoop采用機制完備的分布式文件系統是為了響應以互聯網商業應用為主體的復雜數據處理場景，應對大量數據的頻繁讀寫操作。MPI的應用場景是高性能計算領域，面對的數據應用機制相對簡單，通常一次大規模運算需要的源數據相對固定，得出的計算數據相對完整，不需要進行頻繁的更新迭代。

在高性能計算領域的應用場景中，擴展MPI針對海量數據的處理能力，是本文研究的主要目標。因此，以最簡化的方式，開發設計基于MPI的數據存儲組件(MPI Data Storage Plug-in, MPI-DSP)，將其與MPI計算系統有機融合，提高其在文件管理和大數據方面的處理能力，并以實驗對組件效果進行初步驗證。MPI-DSP適用于一次寫入多次讀取的應用環境，與大多科學計算的應用場景相吻合，但對于需要頻繁寫入更新數據的計算任務，并無特別優勢，因此希望將其作為現行MPI計算體系的一個擴展和補充。

1 MPI-DSP的設計分析

MPI-DSP的設計目標是希望在高性能計算應用場景中增強MPI的大數據處理能力。與互聯網領域應用不同，高性能計算應用往往集中于某一科研領域[4]，有如下特點：1)高頻次的文件并發訪問相對較少；2)實驗源數據相對穩定，不會頻繁寫入，但數據量較大。根據高性能應用場景的特殊性，明確MPI-DSP的設計原則：1)簡化功能設計，避免文件管理的復雜功能，保持系統簡潔與輕量級；2)完成“計算向存儲遷移”，實現數據在本地讀取的核心功能。

采用簡潔的方式不影響構建在MPI上的原有應用，能夠減少系統研發和維護代價。原有MPI的集中存儲機制決定了數據的計算與存儲相分離，在MPI程序設計時，程序人員會考慮盡量增加計算操作而減少節點間的通信操作，也就是常說的“計算換通信”設計原則[5-6]。本文在不影響MPI原有系統的情況下，采用輕量級的數據存儲組件函數，調整優化MPI現有的數據集中存儲模式，使其具備“計算向存儲遷移”的能力，減少MPI實際運算時數據在網絡中的傳輸量。

從系統架構分析，MPI-DSP是現有MPI的一個擴展，它與MPI的計算系統緊密結合，這種結合通過開發新的應用程序接口(Application Program Interface, API)函數來實現。新開發的組件使MPI-DSP具備文件切割、分配和調用功能。MPI-DSP在整個系統中的定位如圖1所示。

從運行機制分析，MPI豐富的函數調用和靈活的消息傳遞機制可以讓集群各計算節點有條不紊地處理復雜運算任務[7-8]。MPI-DSP充分繼承MPI的已有優勢，不改變MPI原有函數調用和消息傳遞機制，通過開發新的功能接口函數，用最簡潔有效的方式對數據文件進行存儲管理。MPI-DSP在計算任務開始前，將任務所需目標數據以分布式形式分配到各計算節點，當執行計算任務時，在各計算節點的本地讀取存儲數據，不用跨節點讀取，可以有效規避網絡傳輸瓶頸，提高計算任務的處理效率。它借鑒了Hadoop中HDFS的大數據處理模式，同時保持MPI原有的高效運算能力和靈活的消息傳遞機制[9]。

圖1 MPI-DSP系統架構Fig. 1 MPI-DSP system architecture

2 MPI-DSP的實現

具體實現中，分兩個階段進行：第一階段，文件上傳、切分與發送；第二階段，任務處理與結果匯總。

第一階段將計算任務所需數據文件上傳到專用存儲空間。以集群子節點進程數量為依據切分數據文件，目的是平衡各節點的計算與存儲負載，減小計算誤差。將切片文件分別發送到對應的子節點上。第二階段是MPI計算任務的核心執行階段。各子節點根據任務要求讀取數據進行計算操作，完成計算任務后匯總結果到主節點。

其中文件的切分、傳輸和操作是MPI-DSP的核心部分。組件接口函數的設計有三條原則：1)不與現有MPI開發模式產生沖突，實現無縫對接；2)盡可能簡化接口函數復雜度和調用方法；3)使MPI對文件的讀取清晰明確。

2.1 文件切分

Hadoop的HDFS以64 MB作為文件切塊的標準大小，這有其優勢，但不夠靈活，作業分割力度不夠彈性[10]。Hadoop的切分方式并不合適MPI環境：如果源文件不大，切分數量不多而MPI集群子節點數目龐大，將使大量子節點閑置，造成資源浪費；如果源文件過大，將產生大量切片文件，在集群子節點數量有限的情況下會開啟大量進程處理切片文件，造成系統擁塞。因此有必要對Hadoop的切片算法進行改進，發揮MPI自有的并行特性，提高整體運行效率。

MPI-DSP以計算節點進程數作為切分依據，文件切分采用等分切割、平均分配的原則，避免了文件切分大小不均勻的問題。根據集群中子節點的進程數量，將大數據源文件連續、平均地分成若干切片文件，充分利用MPI已有的并行性，實現文件操作并行化，確保集群中每個子節點進程分配到相應文件。同時對切片文件進行順序編號，文件順序與各計算節點順序保持一致，可以保障對源文件的有效復原。切分時將切片文件的路徑信息存入主控節點數據庫，便于各子節點讀取調用。文件切分的具體操作通過MPI-DSP中的MPI_Split實現。

實現過程如下：1)主節點通過hostfile查詢MPI集群中子節點進程數目；2)找到將要切片操作的源文件，獲取源文件大小；3)將源文件大小按子節點進程數均分后設定切片文件大小；4)通過函數get_split_name傳入源文件名前綴及切片文件編號；5)將切片文件信息存儲到數據庫中；6)從源文件中按設定的切片文件大小讀取數據寫入切片文件中；7)后一次讀取位置在源文件前一次讀取切片文件大小之后，依次循環；8)直至源文件最后一部分內容寫入最后一個切片文件中，完成文件切分操作。

2.2 切片文件傳輸

MPI-DSP的運行機制是將切片文件在計算任務開始前就分別發送給各計算子節點，有效避免計算節點在執行計算任務時從主節點調用數據文件，以提高整體任務的運行效率。MPI可以通過自有的消息傳遞機制Send和Recv來發送接收文件，但對于大數據量的文件傳輸，會產生大量通信，造成通信阻塞，影響系統穩定性和運行效率。為實現切片文件的有效傳輸，MPI-DSP采用TCP協議的Socket通信方式[11]，以C/S模式，開發兩個駐留程序Master和Slave。主節點運行Master，負責數據發送；子節點駐留Slave，負責數據接收[12]。當啟動文件傳輸服務時，這兩個駐留程序將被啟動執行，分別執行發送和接收操作。

采用Socket方式，一次只能傳輸1 MB數據，無法應對大文件的傳輸，通過對分布式存儲系統文件傳輸的優化[13]和斷點續傳技術[14]可以成功突破Socket對傳輸文件大小的限制，實現海量數據的傳輸。Master程序通過自定義的get_trans_count函數獲取待傳文件(即已切分好的文件)需要傳輸的次數。文件傳輸前通過自定義函數get_filename_master將待傳文件名稱、大小、每次傳輸數據量及傳輸次數發送給接收端(子節點)Slave程序。接收端確認收到信息后，主節點的Master程序開始向接收端分批發送數據，直至待傳文件全部傳輸完畢。

2.3 切片文件讀取

“計算向存儲遷移”的關鍵點是讓計算程序能夠快速準確地定位數據文件位置，從而讀取文件進行相應操作。因此開發出MPI_Open函數。各計算子節點通過MPI_Open函數提取數據庫中的切片文件路徑，定位切片文件存儲位置，讀取切片文件。關鍵在于，讀取的是子節點本地存儲的數據，而無需跨節點網絡傳輸。

相應步驟如下：1)MPI_Open函數導入源文件名和文件指針；2)連接數據庫，通過源文件名和主機名查找對應切片文件存儲信息；3)通過數據庫中切片文件路徑信息，打開該節點對應的切片文件；4)對該節點文件進行相應計算操作；5)待各節點數據讀取、操作完畢，將各節點運算結果發送給主節點，最終由主節點統一匯總輸出結果。

MPI_Open函數設計采用二級指針作為參數，利用參數的返回值進行值傳遞。實現流程如圖2所示。

MPI_Open接口函數部分核心代碼如下:

void MPI_Open(char *file_name,FILE **fp)

{

…

node_file=(char*)malloc(100*sizeof(char));

MPI_Get_processor_name(node_name,&namelen);

//獲取節點名稱

MPI_Comm_rank(MPI_COMM_WORLD,&myid);

//獲取節點當前進程號

node_name[strlen(node_name)]=′