一種中尺度數值預報模式數據并行傳輸應用研究

（重慶市氣象科學研究所，重慶 401147）

0 引言

目前中尺度數值預報模式是局部區域天氣預報業務系統中重要的組成部分，為天氣預報業務提供技術支撐。重慶中尺度數值預報模式是以WRF（Weather Research and Forecasting Model）為核心模塊開發的一種數值天氣預報模式，是重慶精細化數值天氣預報系統的重要組成部分，為重慶本地區域復雜天氣變化過程預測預報提供依據和參考。重慶中尺度數值預報模式的運行平臺是SGI Altix uv 1000系列大型機，該模式一天運行2次（0012 UTC），預報時效96 h，該模式輸出可用的業務數據是NetCDF格式的平面二維（2 Dimension）數據，大小13 GB左右/次，一天數據量約26 GB。根據業務和科研需求，輸出的平面二維數據需及時傳輸到數據管理服務器上處理及歸檔，該數據管理服務器運行了多種數據處理串行程序，數據交換頻次較高，但由于很多程序設計上的缺陷，仍有部分網絡帶寬資源和CPU核資源未被有效利用。目前，該類型數據的傳輸工作是采用基于SCP技術的單核串行程序實現，該程序能夠長期穩定運行，傳輸單次中尺度模式數據實際耗時約2 h，傳輸延時長，造成數據無法及時被氣象業務和科研人員參考和使用；且該程序也不能充分利用空閑的網絡帶寬和數據管理服務器資源。

因此，針對現有數據傳輸程序的缺點，本文設計開發一種新的并行數據傳輸程序，并對數值天氣預報模式的數據有效分治劃分，在不影響數據管理服務器中其它數據傳輸和處理程序的實際環境下，實現一種可應用業務的基于最佳數據分治策略的并行傳輸程序，有效提高中尺度數值預報模式數據的傳輸效率。

1 數據并行傳輸程序

1.1 并行傳輸網絡架構

重慶中尺度數值預報模式的相關程序在以SGI大型機和曙光服務器節點所組成的高性能集群平臺上運行；模式數據通過該集群千兆以太網絡傳輸到數據管理服務器（圖1），SGI大型機和數據管理服務器之間通過網絡交換機連接，工作模式為全雙工，網絡整體帶寬為1000Mbps。數據管理服務器是一臺曙光高性能服務器節點，管理約37種數值預報相關業務數據資料的傳輸和處理工作，也管理著科研用戶的天氣過程試驗資料；服務器架構是對稱多處理器（Symmetrical Multi-Processing，SMP）[1-2]，具備16個處理器核（即CPU Core）。服務器節點之間基于基礎的TCP/IP協議實現數據傳輸交互。

圖1 并行傳輸網絡拓撲Fig. 1 The network topology of parallel transmission

1.2 并行傳輸程序

基于現有有限的網絡和數據管理服務器資源，構建的并行傳輸主程序按照預先設計的參數構建一進程池，該進程池中每一進程只占用一個處理器核，數據傳輸任務由主程序依次分配給進程池中的進程去執行，進程池中各個進程間相互獨立，在執行任務時它們之間沒有信息交互規程，是一種易并發執行過程[3-4]；傳輸任務完成后的進程處于空閑態并返回進程池中，進程池將空閑態的進程信息反饋給主程序，主程序檢查傳輸任務列表，并將新的傳輸任務分配給空閑態的進程執行；當任務列表中無數據傳輸任務，且所有傳輸任務的進程都執行完畢，進程池將所有空閑態的進程信息反饋給主程序，由主程序回收所有進程并關閉進程池。

該并行傳輸程序采用Python語言和Rsync同步技術[5-6]實現數據傳輸流程。Python是目前較流行的一種高級解釋型編程語言，運行該語言編寫的程序調試運行的基礎是在相應的操作系統平臺安裝的Python解釋器[7]；該語言擁有眾多開源的軟件包，非常適合多領域的編程開發和科研人員使用，本程序采用multiprocessing包[8]，通過multiprocessing.Pool函數構建一個含有多個進程的進程池；Rsync是Linux系統中非常流行的數據同步傳輸和備份技術，擁有SSH（Secure SHELL）和C/S架構兩種配置應用方式，能實現數據的快速復制、遠程復制、文件傳輸和安全文件傳輸等功能，以及相同局域網內多個服務器之間的數據快速傳輸及同步功能，Rsync使用一種遠程更新協議，該遠程更新協議采用一個有效的總和檢查搜索算法，該算法允許Rsync通過網絡鏈接僅傳輸兩端之間數據的不同部分，且可以壓縮和校驗數據[5-6,9]；本程序的子功能模塊函數采用第一種配置應用方式，即在兩臺服務器之間采用SSH協議[6]實現身份驗證，身份驗證通過后，調用Rsync實現服務器端到端的數據傳輸。

該并行傳輸主程序的流程（圖2）在啟動初始化后，根據試驗測試環境要求靈活配置處理器核數和劃分分治數據包大小。因每一進程占用一個處理器核，配置處理器核數可理解為設置進程池中進程數目。從并行傳輸主程序的源碼可看出，在初始化和配置相關變量后，process_nn為處理器核數，dip_n為分治的數據包大小，主程序首先通過create_list 函數創建dip_n分治數據包大小的傳輸任務列表，接著通過Python的Pool函數創建一個進程數為process_nn的進程池；傳輸任務列表中任務由主程序分配給具體的進程，此過程由apply_async函數完成。因數據管理服務器的處理器核資源有限，劃分分治后的數據傳輸任務數目均會大于進程池中進程數目，進程池中進程并發執行后，具體執行數據傳輸的函數是rsy_2d_list，在該函數中采用Rsync檢測和傳輸數值預報模式的輸出數據；主程序一直等待進程池中出現空閑態進程，若傳輸任務列表里仍有待執行的任務，程序及時再將列表里的傳輸任務分配給進程池中的空閑態進程；并行傳輸主程序循環反復檢測傳輸任務列表和進程池中進程態，直至傳輸任務全部執行且進程池中進程均為空閑態后，并行程序關閉進程池和回收進程資源，整體的傳輸程序流程結束。

圖2 并行傳輸主程序流程Fig. 2 Main program flow of the parallel transmission

#并行傳輸主程序的源碼

if __name__ == '__main__':

startt=int(time.time())

dip_n=sys.argv[1] # 設置模式2d數據分治數據包大小值

process_nn=sys.argv[2] # 設置并行程序的CPU資源數

nn=string.atoi(dip_n)

process_n=string.atoi(process_nn)

# filename存儲2d數據包含的氣象要素種類，包含溫、壓、濕、風以及降水等

list_result=create_list(filename,nn) # 根據分治數據包大小值，通過該函數將2d數據按照氣象要素種類進行劃分，劃分后返回一個列表對象，供下面使用

print list_result #輸出列表，檢查2d數據劃分是否準確

log_file=log_rsync_data2d.log # 指定數據傳輸的日志記錄文件

len_result=len(list_result)-1 # 確定列表最后一個元素的下標，作為任務分配終止邏輯判斷條件參數

pool=multiprocessing.Pool(processes=process_n) #利用python的multiprocessing創建進程池，該進程池包含了prpcess_n個進程

print "start allocate processes to job "

while len_result >= 0: # 未到達劃分分治列表的最后時，循環向進程池分配數據傳輸任務

print "----rsync_2d_list-----"

pool.apply_async(rsy_2d_list,(list_result[len_result],)) # 將劃分列表中的元素傳參給傳輸任務，將傳輸任務分配到進程池中的一個進程，并開始啟動進程

len_result=len_result-1

print list_result[len_result]

pool.close() # 當所有的數據傳輸任務進程完成后，關閉進程池

pool.join() # 釋放進程池所占資源，回調到主程序中，繼續執行主程序下面代碼

endtt=int(time.time())

print "sub processes done and exit "

print "total run: %d seconds "%(endtt-startt)

if os.path.isfile(log_file):

output_f=open(log_file,'a')

output_f.write("%d core %d elment%d "%(process_n,nn,endtt-startt)) # 計算并行的運行時間，并將時間寫入日志記錄中

output_f.close()

2 數據劃分與分析

數據及問題的劃分分治在并行程序設計中占有重要地位，是并行程序開發過程中首要解決的問題[3]，合理的數據劃分分治將大幅提高并行程序的工作效率，本節重點開展針對重慶中尺度數值預報模式數據劃分分治及對比分析。

2.1 劃分分治

重慶中尺度數值預報模式輸出的數據格式為NC格式二進制的2d（2 dimension）數據文件，該模式一次輸出的數據量約13 GB，整體數據由18444左右個小文件組成，文件數量龐大而無序，每一小文件約700 K左右，表示不同時空分辨率下的一種氣象要素場（如溫、壓濕、風、降水）數據，該數據共包含氣象要素場100多種，包含業務需求的兩種空間分辨率（27 km和3 km）的數據，27 km類型的預報時間分辨率為3 h，3 km類型的時間分辨率為1 h，因此，針對同一種氣象要素場，空間分辨率3 km的數據量要大于空間分辨率27 km的數據量。根據上述的數據特征，本文曾依照空間分辨率劃分數據，劃分后的數據包粒度太粗，且3 km分辨率數據傳輸時間始終會大于27 km分辨率數據傳輸時間，實際并行程序時間以3 km分辨率傳輸時間為準，程序整體運行負載不均衡；按照文件總數進行劃分，又無法快速識別業務上優先級別高的氣象要素場數據是否完整可用。最終尋求將數據的氣象要素場種類作為劃分的依據開展數據的劃分分治。

數據分治劃分流程（圖3）和劃分程序源代碼詳細闡述了模式數據的劃分的過程。其中基礎的是獲取模式2d數據氣象要素場的類型文件，此文件包含了傳輸的模式數據的所有要素場名稱。劃分分治的功能函數是create_list，該函數需設置兩個參數，模式數據的氣象要素場種類文件參數filename和劃分分治的數據包大小參數n；在初始化列表變量后，讀取和處理氣象要素場類型文件中的要素場，生成一個初始的列表存放在內存，按照變量n值的大小，計算劃分分治數icount和分治后的余數rest_n，通過初始列表的下標值nex_i和pre_i，盡可能均等劃分初始列表中的氣象要素場，并將劃分后的結果依次存儲于傳輸任務列表中，最后將劃分分治后的傳輸任務列表返回給主程序。

#數值模式數據劃分功能——劃分程序源代碼

def create_list(filename,n):

# 初始化該程序中所要使用的變量。

L_re=[]

new_L_re=[]

new_L_re01=[]

if os.path.isfile(filename): # 判斷2d數據要素文件是否存在。filename提取和存儲模式2d數據包含的所有的氣象要素。

print "the file is exist "

圖3 數據劃分流程Fig. 3 The flow of data partition

# 讀取氣象要素的數據到內存中，通過兩個循環將要素數據有效存儲于初始列表中，去除異常字符，便于劃分操作

ffile=open(filename,'r')

for line in ffile:

line=line.strip(" ").split(",")

L_re.append(line)

ffile.close()

for list_e in L_re:

str_e=str(list_e).strip("[]").strip("''")

new_L_re.append(str_e)

icount=len(new_L_re)/n # 根據設置的劃分數據包大小，確定劃分分治數

rest_n=len(new_L_re)%n # 劃分后，確定最后一項數據包大小

while icount > 0:

if rest_n == 0: # 當余數為0時，2d數據會被均等劃分

nex_i=icount*n

pre_i=(icount-1)*n

new_L_re01.append(new_L_re[pre_i:nex_i])

if rest_n != 0: #當余數不為0時，最后一項數據包大小值不等于n，而是rest_n,但rest_n只能小于n，所以整體不會延長數據并行傳輸時間。

pre_i=icount*n

nex_i=pre_i+rest_n

new_L_re01.append(new_L_re[pre_i:nex_i])

rest_n=0

icount=icount-1

else:

print "the file not exist"

del L_re

del new_L_re

return new_L_re01 # 返回根據n劃分好的傳輸任務列表

依據模式數據氣象要素場的類型，劃分原則為一方面劃分的數據包包含的氣象要素場類型不能太多，避免數據包粒度變粗，影響傳輸程序執行效率；另一方面盡可能實現均等劃分，確保傳輸并行程序負載均衡；基于上述原則本文將模式數據劃分為4種不同粗細粒度的分治數據包，形成4種數據分治策略（表1），其中分治數據包大小是指每個分治策略下分治數據包包含的氣象要素場類型數目，劃分分治數是指每個分治策略下總共劃分多少個分治數據包，4種分治策略中最細粒度的是分治數據包含1種氣象要素場，最粗力度的分治數據包包含6種氣象要素場；下面將對比分析這4種不同數據分治策略在不同的處理器核數下的運行趨勢。

表1 數據分治策略Table 1 The data partition strategy

串行單核的數值模式傳輸程序平均運行時間約7810s，數據傳輸時間很長，這主要是因為現有數據管理服務器從一個物理網絡接口每天傳輸交互約37種數值預報模式相關的數據，物理網絡接口負載過大的導致的，但網絡帶寬并沒有最大化的利用。本文將4種不同數據分治策略應用到并行傳輸程序中，通過運行時間、加速比和并行效率三種指標對比分析，綜合選擇最佳分治策略；加速比（S）是衡量并行傳輸程序一個重要指標，是指單核運行時間（Ts）和多核并行運行時間（Tp）的比值，該值越大表明并行運行時間越短[2,10-11]；并行效率E是在加速比上的基礎上引入了每次并行使用的處理器核數n，實質上是單核運行時間和多核并行運行時間乘以核數n的比值，再乘以100%，主要描述并行程序運行過程中系統開銷時間的占比[4,11]。

2.2 試驗和對比分析

試驗測試環境為實際的業務應用環境，在不影響數據管理服務器其他數據傳輸交互業務的實際情況下，極大化的占用空閑態的處理器核資源和網絡帶寬。在實際應用環境下測試，發現當并發執行進程數超過8時，數據傳輸會出現大量丟包現象，進程數設置超過10時，數據傳輸完全中斷失敗，這說明進程數設置為8以上的值時，便無服務器資源和網絡帶寬可供分配使用，且逐漸開始占用其他數據傳輸處理程序的資源，所以為了實現業務應用環境下本并行傳輸程序改善的實際效果，設置的進程數的閾值為8。服務器節點操作系統均為SUSE 11 Linux x86_64，服務器節點均安裝Python，并實現了SSH身份驗證，試驗分為三種類型，為了保證試驗效果和試驗數據的可信度，試驗過程中處理器核數遞增間隔為2；對比分析過程中，各個指標的對比分析圖中將離散試驗數據用不同的類型曲線連接，以便直觀展示各分治策略的變化趨勢。

不同數據分治策略下的并行運行時間均優于單核串行程序（圖4）。各分治策略隨著核數增加，并行運行時間逐漸縮小，當處理器核數從2到4時，運行時間降幅較大，運行時間差2000多秒，從4增至8時，并行運行時間降幅度明顯縮小，時間差均在300 s左右，從各分治策略的轉折線可看出，核數的增加并不意味著運行時間會以相同的速度減少，主要是隨著核數的增加，系統在分配任務的進程數增多到一定程度，進程間通信開銷大幅度增長，但系統總線只有一條，所以才導致并行運行時間降低力度縮小，運行效率開始漸漸降低；分治策略1、2、4運行時間趨勢一致且幾乎重合，在超過4個處理器核后，趨勢線的斜率有明顯一致的變化。分治策略3有點不同，在處理器核數由2增至6時，趨勢線斜率變化相對而言變化不大，增至到8時，趨勢線斜率和其他策略保持一致，這說明隨著核數增加，分治策略3的運行時間降低力度明顯，且運行時間開始逐漸優于其他分治策略。

圖4 數據分治策略運行時間變化趨勢Fig. 4 The data partition strategies runtime versus number of cores

分治策略的加速比和并行時間變化趨勢有一定的關聯（圖5）。隨著處理器核數的增加，各分治策略下的加速比均呈現非線性的上升趨勢。從圖3中折線的斜率可看出，分治策略1、2、4的處理器核數從2增至4時，加速比變化幅度較大，數據傳輸效率提升效果明顯，核數從4增加到8時，加速比變化趨勢發生了轉折，加速比變化幅度相對減小；分治策略3的加速比變化趨勢有點不同，當處理器核數增加到6時，加速比變化趨勢最大，核數從6增至8時，加速比趨勢趨于平緩；分治策略3使用處理器核數增加到4時，加速比值均低于其他分治策略，隨著處理器核數不斷增加，加速比值逐漸提高，并最終優于其他分治策略，在超過4個處理器核數后，該分治策略對數據的傳輸效率改善效果最大。加速比趨勢變化和上述的并行運行時間的趨勢變化類似，說明并行運行時間變化趨勢決定加速比變化趨勢。

圖5 數據分治策略加速比變化趨勢Fig. 5 The data partition strategies acceleration ratio versus number of cores

最后，對比分析一下不同分治策略的并行效率（圖6），隨著處理器核數增加，分治策略的并行效率均呈現不同的下降趨勢。主要因為即隨著處理器核數資源占用越來越多，進程數越多，并行程序需耗費更多系統開銷時間來完成資源調度分配，導致數據并行傳輸處理時間所占比重降低；分治策略1、2、4呈現的趨勢較為一致，并行效率從90%左右逐漸降低至70%以下，且是非線性的變化趨勢，處理器核數增至4個以上時，并行效率降低明顯；而分治策略3卻呈現另外一種變化趨勢，它的并行效率雖然整體上也是逐漸降低，但在核數增至到6時，并行效率開始大幅提升，這一點和其策略其它兩個指標有一定的關聯，該分治策略前兩個指標可以看出，當核數從4增至6時，該分治策略發生較大變化，指標值從最差變為最佳，系統開銷時間占比相對于整體并行傳輸時間最低；從整體并行效率變化趨勢看，分治策略3并行化效率變化范圍控制在70%～85%，整體變化范圍比較穩定，其他分治策略并行效率的變化范圍和降低力度相對較大。

圖6 數據分治策略并行效率變化趨勢Fig. 6 The data partition strategies parallel efficiency versus number of cores

為充分利用現有空閑網絡帶寬和服務器處理器核資源，實現數據傳輸運行時間最短和并行傳輸程序穩定和最佳并行效率，綜合對比上述并行運行時間、加速比和并行效率三個指標，分治策略3是并行數據傳輸應用的最佳策略，基于該分治策略的并行傳輸程序能夠能將傳輸時間從7810 s縮小至1297 s，加速比高達6.02，網絡帶寬利用率提高了6倍左右，且并行效率維持在75%左右，數據管理服務器的處理器核利用率達到50%。

3 結論

以重慶中尺度數值天氣預報模式為例，中尺度數值預報模式的數據并行傳輸程序在現有網絡和計算機資源基礎上，采用Python和Rsync數據傳輸同步技術構建和實現；根據模式數據的特征和實際業務應用環境，將模式數據劃分為4種分治策略，對比分析4種不同的分治策略，綜合選擇最佳的數據分治策略，最終實現最優的模式數據并行傳輸。最優數據并行傳輸一方面極大減少模式數據并行傳輸時間，提高模式數據傳輸效率，另一方面也較充分利用數據管理服務器中空閑的處理器核和網絡帶寬，為中尺度數值預報模式海量數據高效率傳輸備份提供一種解決方案。但重慶精細化數值天氣預報系統不僅包含重慶中尺度數值預報模式，按照建設要求還將包含其他不同尺度多類型數值天氣預報模式，傳輸的數據量也將大幅增大，未來將進一步研究基于MPI的進程組間無阻塞通信方式的更加通用的并行傳輸應用技術，以滿足重慶精細化數值天氣預報系統的更大容量數據高效率傳輸需求。