一種優化的并行數據流調度算法

葛茂松 富春巖 支援 李微娜 周虹

摘要：很多需要對數據流進行實時處理、快速響應。本文對已有的MapReduce調度器進行了分析，結合它們的優缺點，對MapReduce調度算法進行了優化。實驗表明，該優化算法可進行精準的估算，運行效率較高。

關鍵詞：數據流;調度算法;優化查詢

中圖分類號：TP391? ? ? 文獻標識碼：A

文章編號：1009-3044（2019）22-0003-02

開放科學（資源服務）標識碼（OSID）：

1 引言

目前，在電力、電信、金融、網絡安全等很多領域，都需要對數據流進行在線實時處理。數據流無須存儲，而需要通過數據流查詢被及時、快速地處理。為了應對不斷增大的數據流規模，出現了分布式數據流處理技術，根據數據流速率的變化可動態調整數據流查詢所需的計算資源。

MapReduce是一個用于大規模數據集的并行運算模型，廣泛地應用在分布式查詢、分布式排序、web訪問日志分析、機器學習等方面。它有FIFO、Capacity Scheduler和Fair Scheduler三種調度器，這三種調度器它們有各自的優點。但是，共同的缺點是：在作業提交前，要求對系統的參數進行預設。也就是說，一旦作業提交預先設定完成，MapReduce系統給每一個任務的資源分配策略就已經確定，再無法根據實際情況進行動態的調整，更不要說自適應的調整。因此，本文對MapReduce調度算法進行了優化，改進后的算法能夠基于正在進行的任務的進度對任務完成的時間進行預估，這樣就可以根據每個任務進行的情況，自適應地動態的分配資源，從而提高系統資源的利用率。

2 相關定義

為了準確描述優化的MapReduce調度算法，本文有如下定義，見表1。

其中，作業m中第i項任務的任務完成時間由任務已運行時間和任務剩余時間兩部分組成，因此有，[αmi=βmi+δmi]，作業集合[tmi]由[Cm]、[Um]和[Rm]三部分組成。

3 任務調度策略

任務調度策略就是要根據作業性能估算后得到的任務平均完成時間，通過相應的公式推導出當前作業所需的任務執行單元的數量，通過這個數量來確定當前作業的優先級。調度器將根據所確定的優先級分配相適應的合理資源。該任務調度策略由給作業分配合適的優先級和分配算法兩部分組成。

3.1 作業的優先級

在MapReudce中，正在運行的某作業，需要進行調度時，調度器會根據任務調度策略給其分配合適數量的任務執行單元。

因為要根據在規定時間內完成作業所需要分配的執行單元數而算出來的作業的優先級，因此，需要估算出某項作業m中，正在執行的任務數和尚未執行的任務數。如果，每個任務執行單元需要花費[μm]時間，我們可對作業m當前所需任務執行單元數[smreq]進行估算，見如下公式：

其中，[Tmgoal]為目標完成時間，[Tcurr]為當前時間。

計算后，調度器將依據得到的[smreq]值作為作業m的權重，將當前作業集合放于優先隊列中。

但是，由于任務調度策略是依據以往的任務運行情況對任務完成時間進行估算的，因此，在一些特殊情況下，任務調度策略無法準確地進行估算。例如，一些已超過目標完成時間的作業，優先級會被賦予的較高，使其盡快得到資源完成任務。另外，作業剛提交的時，調度器還未收集到信息，無法估算作業最終完成時間，也就無法計算出所需任務執行單元數[smreq]。這種情況下的作業將獲得較高的優先級，以方便調度器盡快調度該任務。

因此，調度器要通過以下幾個因素對任務的優先級進行整體的評估計算：第一，判斷作業是否已超過目標完成時間;第二，看調度器是否收集了作業的狀態信息;第三，估算出的任務執行單元數目。

任務調度策略中規定調度器分配資源的順序為：已經超過目標完成時間的作業、未被調度器收集到信息的作業、所需任務執行單元數較大的作業。

3.2 優化的調度算法

確定了優先級之后就要確定基于作業的優先級的分配算法。調度算法主要完成在作業服務器中維護優先隊列的功能。優化后的調度算法定義了三個函數，分別是初始化函數、添加作業到優先隊列函數和調度優先隊列已有作業函數。

初始化作業優先隊列函數的主要功能是創建一個優先隊列。其中隊列的元素為Job類型，包含作業的Id、作業的狀態和當前任務執行單元數等。作業又包括UDEAD、NODATA、ADJUST這3種狀態。

用戶在提交新的作業時，開始調用添加作業到優先隊列函數，這個函數的功能是將新作業加入到優先隊列中去。輸入參數為Job類型，Job類型參數的初始狀態為NODATA，偽代碼如下：

當工作服務器收到來自作業服務器的信息時，將執行調度優先隊列中已有作業函數。在函數中，先遍歷作業服務器中返回的任務列表，如果該任務已在優先隊列中，就更新作業優先隊列中相關信息。若該任務不在優先隊列中，就將該任務所屬作業加入優先隊列中。然后，再按照作業的優先級依次地取出作業，進行相應操作。若該作業狀態是UNDEAD狀態或NODATA狀態，那么，說明該作業已超出了作業的目標完成時間;或者此作業處于剛剛提交狀態，無信息可循，那么，將給該作業分配最大任務執行器數目maxSlot;若作業當前處于ADJUST狀態，就依據公式1進行計算，得出作業所需任務執行單元數，并進行分配。

4 結論

本文對MapReduce調度算法中的任務調度策略進行了優化。調度器可根據記錄的作業及任務信息數據，估算出任務完成時間。優化后的算法可計算出當前作業所需資源，并給出當前作業對應的優先級，形成優先隊列。

通過模擬真實場景下多個作業提交的情況，對算法進行了實驗驗證，作業運行的前30%時間里，因為收集到的關于該作業的信息數據少，所以估算的完成時間與實際作業的完成時間差距較大。但在作業運行了30%的時間以后，由于調度器收集到了足夠的關于該作業的信息數據，估算的完成時間與實際完成時間基本相符。

在實驗過程中，初始的幾次實驗運行時間較長，但實驗次數增加后，調度器收集到的信息數據量足夠大之后，就可以對作業的完成時間做出比較精準的估算，分配給作業的任務執行單元數也更為合適，運行效率較高。

參考文獻：

[1] Bonnet P， Gehrke JE， Seshadri P. Towards sensor database systems[C]//Tan K-L， Franklin MJ， Lui JCS， eds. Proceedings of the 2nd International Conference on Mobile Data Management. Hong Kong： Springer-Verlag， 2010. 3-14.

[2] 富春巖，葛茂松，張立銘，李微娜，趙佳彬，等. 一種準實時MapReduce調度算法的改進與實現[J]. 電腦知識與技術，2016（12）：3-4.

【通聯編輯：梁書】