[發明專利]一種面向Map/Reduce型海量數據處理平臺的作業調度方法

有效

• 申請號: 201410531590.0
• 申請日: 2014-10-10
• 公開（公告）號: CN104317650B
• 公開（公告）日: 2018-05-01
• 發明（設計）人: 梁毅;王玉鳳;樊明璐;張辰
• 申請（專利權）人: 北京工業大學
• 主分類號: G06F9/48
• 分類號: G06F9/48;G06F9/50
• 代理公司: 北京思海天達知識產權代理有限公司11203
• 代理人: 張慧

1.一種面向Map/Reduce型海量數據處理平臺的作業調度方法，包括：搶占資源、作業調度模塊、應用管理模塊、任務執行模塊、節點管理模塊、任務掛起與恢復方法、面向搶占資源的作業調度方法，其特征在于包括如下步驟：

所述的搶占資源是當Reduce任務等待獲取Map任務的輸出數據時，該Reduce任務所釋放的計算資源；

所述的作業調度模塊，對Map/Reduce作業所包含的Map任務和Reduce任務進行任務掛起與恢復決策；以任務為單位進行計算資源分配，并將掛起的Reduce任務所釋放的資源分配給待調度的Map任務；

所述的作業調度模塊，保存可用搶占資源信息表、搶占資源使用信息表和待調度任務信息表，可用搶占資源信息表用于記錄平臺當前可以使用的搶占資源，格式如下：

資源號 集群節點號 可使用資源量 掛起Reduce任務號

其中，集群節點號是搶占資源所在集群節點的編號，可使用資源量是搶占資源所包含的可以使用的計算資源的數量，掛起Reduce任務號是釋放該搶占資源的Reduce任務編號；

搶占資源使用信息表用于記錄平臺當前搶占資源的使用情況，格式如下：

資源號 任務號 資源分配量

其中，任務號是使用該搶占資源的任務編號，資源分配量是任務實際占用該搶占資源所包含資源的數量，資源分配量不大于該搶占資源的當前可使用資源量；

待調度任務信息表用于記錄平臺當前需要分配計算資源的Map任務和Reduce任務，格式如下：

作業號 任務號 任務類型 資源需求量 到達時間

其中，任務號和作業號分別表示需要分配計算資源的任務編號和其所屬作業的編號，任務類型包括Map任務和Reduce任務，資源需求量表示任務運行所需要的計算資源數量，到達時間表示該任務所屬作業到達Map/Reduce平臺的時間；

所述的應用管理模塊，對Map/Reduce作業從提交到結束的整個生命周期進行管理，跟蹤記錄Map/Reduce作業所包含的Map任務及Reduce任務的狀態，包括待調度狀態、運行狀態、掛起狀態和結束狀態；

所述的應用管理模塊保存任務狀態信息表，用于記錄當前平臺中作業的狀態及作業運行相關信息，格式如下：

其中，任務號和作業號分別表示任務及其所屬作業的編號，任務類型包括Map任務和Reduce任務，集群節點號表示該任務運行所在的集群節點編號，執行代碼路徑表示任務需要運行的代碼的存儲路徑，處理數據文件表示任務需要處理數據所在文件的存儲路徑，數據起始偏移量和數據結束偏移量分別表示需要處理數據在文件中的起始和結束偏移量；

所述的任務執行模塊，可對其執行的Reduce任務執行掛起操作，針對掛起的Reduce任務記錄其還未拷貝其中間結果數據以及中間結果數據的存儲路徑信息；

所述的節點管理模塊，可對所在節點中運行的Map任務和Reduce任務進行監控，觸發Reduce任務掛起和釋放操作；

所述的任務掛起與恢復方法，包括如下步驟：

步驟1.1：作業調度模塊周期性地執行步驟1.2到步驟1.7；

步驟1.2：作業調度模塊從應用管理模塊獲得所有處于運行狀態、掛起狀態和結束狀態的Map和Reduce任務信息，包括任務的任務號、任務所屬作業的作業號、任務狀態、任務狀態記錄時間；

步驟1.3：作業調度模塊根據步驟1.2收集的任務信息，對每一個處于運行狀態的Reduce任務計算其數據拷貝剩余處理時間以及與之屬于同一Map/Reduce作業的Map任務的剩余執行時間，根據計算結果，選擇需要掛起的Reduce任務，方法如下，若不存在需要掛起的Reduce任務，則轉至步驟1.5；

步驟1.3.1：作業調度模塊遍歷所有處于運行狀態的Reduce任務信息，對每個Reduce任務執行步驟1.3.2至步驟1.3.6；

步驟1.3.2：選取所有與該Reduce任務屬于同一作業且處于運行狀態的Map任務，構成任務集SMap，對于SMap中的每一個Map任務i，計算其剩余執行時間TMLeft_i，計算方法如下：

其中TMLeft_i代表Map任務剩余執行時間，以毫秒為單位，TMExecuted_i代表Map任務已執行時間，以毫秒為單位，PTask_i代表Map任務當前執行進度，進度值在(0,1]區間內，本發明將PTask_i設定為Map任務已處理數據量與其所需處理的數據總量的比值，PTask_i和TMExecuted_i是由對應的i任務執行模塊統計得出并發送給作業調度模塊；

步驟1.3.3：統計該Reduce任務所屬的作業中處于運行狀態的Map任務的最短剩余執行時間TMleft_min，方法如下：

TMLeft_min＝min{TMLeft_i|i∈SMap}

其中，TMLeft_min代表Map任務剩余執行時間最小值，min表示為對大括號內的所有數值求最小值；

步驟1.3.4：計算該Reduce任務拷貝剩余處理時間TRLeft，方法如下：

其中TRLeft代表Reduce任務拷貝剩余處理時間，以毫秒為單位，TRFetched代表Reduce任務已執行拷貝操作的時間長度，以毫秒為單位，Num_c代表Reduce任務已經拷貝的Map任務的個數，Num_t代表與Reduce任務屬于同一Map/Reduce作業且已運行結束的Map任務的個數，TRFetched由任務執行模塊計算Reduce任務執行拷貝操作的開始時間與當前時間的差值獲得，Num_c由任務執行模塊統計已由Reduce任務完成拷貝其輸出數據的Map任務的個數獲得，TRFetched和Num_c均由任務執行模塊推送給作業調度模塊，Num_t則由作業調度模塊根據其收集的任務狀態信息，統計與Reduce任務屬于同一Map/Reduce作業且處于結束狀態的Map任務個數；

步驟1.3.5：根據該Reduce任務對應的Map任務剩余執行時間最小值TMLeft_min和該Reduce任務的拷貝剩余處理時間TRLeft，判斷該Reduce任務是否滿足掛起條件，判斷條件如下：

其中TMLeft_min代表Map任務剩余執行時間最小值，TRLeft代表Reduce任務拷貝剩余處理時間，Dsuspend代表設定的閾值，取值范圍在[0，1]區間內；

步驟1.3.6:若該Reduce任務滿足掛起條件，則將其標記為“待掛起”；

步驟1.4：作業調度模塊對每一個標記為“待掛起”的Reduce任務執行掛起操作，并釋放掛起Reduce任務所占用的計算資源，方法如下：

步驟1.4.1：作業調度模塊查找Map/Reduce平臺中標記為“待掛起“的Reduce任務，對每一個待掛起Reduce任務，通過應用管理模塊通知相應的任務執行模塊，判斷其是否處于空閑等待狀態，判斷方法為Reduce任務執行模塊查看該模塊中所有Map任務輸出數據讀取線程是否均處于空閑狀態，任務執行模塊通過應用管理模塊向作業調度模塊返回查詢結果，若是，執行步驟1.4.2至步驟1.4.6；

步驟1.4.2：作業調度模塊通過應用管理模塊向相應的任務執行模塊發送Reduce任務掛起的消息，消息格式如下：

任務掛起消息標識 作業號 任務號

其中，任務掛起消息標識表示消息類型，任務號和作業號分別表示被掛起Reduce任務的任務編號及其所屬Map/Reduce作業的編號；

步驟1.4.3：任務執行模塊接收到掛起Reduce任務的消息后，將該Reduce任務還未拷貝其輸出數據的Map任務列表以及輸出數據的存儲路徑信息，保存在其所在節點的節點管理模塊；

步驟1.4.4：任務執行模塊中止該Reduce任務的運行，并向所在節點的節點管理模塊發送Reduce任務掛起消息，消息格式與步驟1.4.2中Reduce任務掛起消息格式相同；

步驟1.4.5：節點管理模塊向應用管理模塊發送Reduce任務掛起消息，消息格式與步驟1.4.2中Reduce任務掛起消息格式相同，應用管理模塊收到消息后，將該Reduce任務的狀態修改為掛起狀態，并記錄任務的掛起時間；

步驟1.4.6：節點管理模塊向作業調度模塊發送Reduce任務掛起消息，消息格式與步驟1.4.4中Reduce任務掛起消息格式相同; 作業調度模塊接收到任務掛起消息后，在可用搶占資源信息表中，為該Reduce任務釋放的資源新增一個表記錄；

步驟1.5：作業調度模塊對Map/Reduce平臺中每一個處于掛起狀態的Reduce任務，統計自Reduce任務最后一次掛起后新增的與該Reduce任務屬于同一Map/Reduce作業且已結束的Map任務個數，根據統計結果以及與該Reduce任務屬于同一Map/Reduce作業的Map任務總數，選擇可以被釋放的Reduce任務，若不存在需要釋放的Reduce任務，則轉至步驟1.7，選擇被釋放的Reduce任務的方法如下：

步驟1.5.1：作業調度模塊遍歷所有處于掛起狀態的Reduce任務，對每一個任務執行步驟1.5.2到1.5.3；

步驟1.5.2：對該Reduce任務，判斷其是否需要恢復運行，判斷條件如下：

其中，Ns代表與該Reduce任務屬于同一作業且已執行結束的Map任務數，Nf代表與該Reduce任務屬于同一作業且Reduce任務已完成對其輸出數據拷貝的Map任務數，Nt代表與該Reduce任務屬于同一作業的Map任務總數，Dp代表設定閾值，取值范圍在[0,1]區間內；

其中，Ns-Nf以及Nt的值由應用管理模塊計算并發送給作業調度模塊，Ns-Nf的計算方法是統計在Reduce任務最后一次掛起的時間點之后所有與之屬于同一作業且處于結束狀態的Map任務的總數，Nt的計算方法是統計該Reduce任務所在作業的Map任務總數；

步驟1.5.3：若該Reduce任務滿足釋放條件，則將其標記為“待釋放”；若不滿足釋放條件，對該任務的狀態不作修改；

步驟1.6：作業調度模塊對每一個標記為“待釋放”Reduce任務，將正在使用其所釋放資源的Map任務掛起，將資源重新分配給該掛起Reduce任務，并恢復該Reduce任務的執行，方法如下；

步驟1.6.1：作業調度模塊查找所有標記為“待釋放”的Reduce任務，對每一個待釋放的Reduce任務，執行步驟1.6.2到步驟1.6.11；

步驟1.6.2：作業調度模塊根據Reduce任務號在可用搶占資源信息表中查找該Reduce任務所釋放搶占資源的資源號，根據資源號查看搶占資源使用信息表中是否存在使用該Reduce任務所釋放搶占資源的記錄，若不存在，則轉至步驟1.6.9；否則，作業調度模塊統計所有正在使用該Reduce任務所釋放的搶占資源的Map任務，對每一個Map任務，執行步驟1.6.3到步驟1.6.7；

步驟1.6.3：作業調度模塊向應用管理模塊發送中止該Map任務執行的消息，應用管理模塊將該任務中止消息發送給運行該Map任務的任務執行模塊，消息格式如下：

任務中止消息標識 作業號 任務號

其中，任務中止消息標識表示消息類型，任務號和作業號分別表示被中止的Map任務的任務號及其所屬Map/Reduce作業的作業號；

步驟1.6.4：任務執行模塊接收任務中止消息后，等待Map任務將當前正在處理的數據記錄處理完畢后，將Map任務產生的輸出數據寫入所在計算節點的本地磁盤存儲器中，計算該Map任務尚未處理的數據記錄在數據文件中的起始偏移量，并將該偏移量信息保存于本節點的節點管理模塊中，中止Map任務執行，并向本節點的節點管理模塊發送該Map任務中止消息，消息格式與步驟1.6.3相同；

步驟1.6.5：節點管理模塊接收Map任務中止消息后，向作業調度模塊發送Map任務中止消息，并將中止Map任務消息及Map任務尚未處理數據消息發送給應用管理模塊; 其中，Map任務中止消息格式與步驟1.6.3相同，Map任務尚未處理數據消息格式如下：

任務未處理數據消息標識 任務號 數據處理偏移量

其中，任務未處理數據消息標識表示消息類型，任務號表示被中止任務的編號，數據文件偏移量表示該任務尚未處理的數據記錄在數據文件中的起始偏移量；

步驟1.6.7：作業調度模塊接收到節點管理模塊發送的Map任務中止信息后，根據任務號查找搶占資源使用信息表中該Map任務對應的搶占資源使用記錄，根據資源使用記錄中資源號查找可用搶占資源信息表中對應的搶占資源記錄，并將該可用搶占資源信息記錄中可使用資源量的值修改為當前值與Map任務資源使用量值的和；從搶占資源使用信息表中刪除該Map任務對應的資源使用記錄；

步驟1.6.8：應用管理模塊接收到Map任務中止消息及尚未處理數據消息后，將該Map 任務的狀態改為待調度狀態，用所接收的數據處理偏移量信息更新既有的任務數據處理起始偏移量信息，向作業調度模塊提交一個計算資源申請消息消息格式如下：

作業號 任務號 資源需求量

其中，任務號和作業號分別表示申請計算資源的任務編號以及其所屬Map/Reduce作業的編號，資源需求量表示該任務所需要的計算資源數量；

步驟1.6.9：作業調度模塊根據資源號查看搶占資源使用信息表中是否還存在使用該Reduce任務所釋放搶占資源的記錄，若存在，則執行步驟1.6.3；若不存在，則執行步驟1.6.10到步驟1.6.11；

步驟1.6.10：作業調度模塊向管理該Reduce任務的應用管理模塊發送任務啟動消息，消息格式如下：

任務啟動消息標識 作業號 任務號 集群節點號

其中，任務啟動消息標識表示消息類型，任務號和作業號分別表示獲得Reduce任務及其所屬作業的編號，集群節點號表示Reduce任務運行所在的節點編號；

步驟1.6.11：應用管理模塊接收任務啟動消息，根據作業號和任務號，將任務狀態信息表中，對應的任務記錄中任務狀態修改為運行態，并根據集群節點號向該節點的節點管理模塊發送任務執行消息，消息格式如下：

其中，任務號和作業號分別表示需要運行的Map任務及其所屬作業的編號，任務執行代碼路徑表示任務需要運行的代碼的存儲路徑，任務處理數據文件表示該Map任務需要處理數據所在文件的存儲路徑，數據處理起始偏移量和數據以及需要處理數據在文件中的起始偏移量；

步驟1.6.12：節點管理模塊接收任務啟動消息后，啟動任務執行模塊；任務執行模塊從節點管理模塊讀取Reduce任務掛起時保存的數據拷貝信息，繼續從已執行結束且還未拷貝其輸出數據的Map任務的運行節點上拷貝輸出數據，并根據輸出數據存儲路徑信息存放所拷的數據；

步驟1.7：作業調度模塊查看是否存在可用的搶占資源；若存在，則執行一次新的面向搶占資源的作業調度；

所述的面向搶占資源的作業調度，包括如下步驟：

步驟2.1：作業調度模塊查看可用搶占資源信息表是否存在可用搶占資源記錄，若存在，對每一個可用的搶占資源記錄順序執行步驟2.2到步驟2.4，若不存在，則結束該輪調度；

步驟2.2：作業調度模塊根據待調度任務信息表，對包含待調度Map任務的作業按照到達時間進行排序，選取最早到達的作業；

步驟2.3：對步驟2.2所選取的作業，在其所包含的待調度Map任務中，按照數據本地化、數據本機柜化、數據本交換化的優先級順序選擇Map任務；為所選擇的Map任務分配資源且運行任務，直至該可用的搶占資源的可使用資源量均小于作業剩余的待調度Map任務的資源需求量，或該作業中所有待調度Map任務均已獲得所需的計算資源，方法如下：

步驟2.3.1：在步驟2.2選取作業所包含的待調度Map任務中，選取滿足數據本地化條件的Map任務集，數據本地化是Map任務所處理的數據存儲于可用資源所在計算節點上；遍歷該Map任務集，對每一個Map任務判斷該搶占資源的可使用資源量是否大于Map任務的資源需求量，若大于，則為該Map任務在搶占資源使用信息表中增加一個資源使用記錄，其中資源使用份額即為該Map任務計算資源需求的數量，并在可用搶占資源信息表中，將該搶占資源的可使用資源量值修改為當前值與Map任務資源使用份額的差值；將該任務的資源需求記錄刪除；

步驟2.3.2：在該作業所包含的待調度Map任務中，選取滿足數據本機柜化條件的Map任務集，數據本機柜化是存儲Map任務所需處理數據的節點與可用搶占資源所在的計算節點同處于一個機柜中，遍歷該Map任務集，對每一個Map任務判斷該搶占資源的可使用資源量是否大于Map任務的資源需求，若大于，則為該Map任務在搶占資源使用信息表中增加一個資源使用記錄，其中資源使用份額即為該Map任務計算資源需求的數量，并在可用搶占資源信息表中，將該搶占資源的可使用資源量值修改為當前值與Map任務資源使用份額的差值；并將該任務的資源需求記錄刪除；

步驟2.3.3：在該作業所包含的待調度Map任務中，選取滿足數據本交換化條件的Map任務集，數據本交換化是存儲Map任務處理數據的節點與可用搶占資源所在的計算節點處于不同的機柜中，但可通過同一交換機互連，遍歷該Map任務集，對每一個Map任務判斷該搶占資源的可使用資源量是否大于Map任務的資源需求，若大于，則為該Map任務在搶占資源使用信息表中增加一個資源使用記錄，其中資源使用份額即為該Map任務計算資源需求的數量，并在可用搶占資源信息表中，將該搶占資源的可使用資源量值修改為當前值與Map任務資源使用份額的差值；并將該任務的資源需求記錄刪除；

步驟2.3.4：作業調度模塊對步驟2.3.1至步驟2.3.3中所有獲得搶占資源的Map任務，向管理該Map任務的應用管理模塊發送任務啟動消息，消息格式如下：

任務啟動消息標識 作業號 任務號 集群節點號

其中，任務啟動消息標識表示消息類型，任務號和作業號分別表示獲得搶占資源的Map任務及其所屬作業的編號，集群節點號表示Map任務獲得資源所在的節點編號；

步驟2.3.5：應用管理模塊接收任務啟動消息，根據作業號和任務號，將任務狀態信息表中，對應的任務記錄中任務狀態修改為運行態，并根據集群節點號向該節點的節點管理模塊發送任務執行消息，消息格式如下：

其中，任務號和作業號分別表示需要運行的Map任務及其所屬作業的編號，任務執行代碼路徑表示任務需要運行的代碼的存儲路徑，任務處理數據文件表示該Map任務需要處理數據所在文件的存儲路徑，數據處理起始偏移量和數據以及需要處理數據在文件中的起始偏移量；

步驟2.3.6：節點管理模塊接收任務執行消息，為Map任務啟動一個任務執行模塊，任務執行模塊根據任務執行消息中所需處理數據文件及起始偏移量的信息，讀取數據文件，執行該Map任務；

步驟2.4：若該作業中所有待調度Map任務均已獲得所需的計算資源，則重復執行步驟2.2到步驟2.3，否則，結束調度。