汪志峰 ,趙宇海* ,王國(guó)仁

（1.東北大學(xué)計(jì)算機(jī)科學(xué)與工程學(xué)院，沈陽(yáng)，110169；2.北京理工大學(xué)計(jì)算機(jī)學(xué)院，北京，100081）

近年來，由于數(shù)據(jù)量呈指數(shù)式增長(zhǎng)，處理大數(shù)據(jù)［1］的方式也發(fā)生了很大變化，迄今已經(jīng)歷三代引擎的改進(jìn)：第一代以Hadoop［2］為代表，利用MapReduce［3］進(jìn)行大數(shù)據(jù)處理；第二代以Spark［4］為代表，是基于RDD（基于內(nèi)存計(jì)算）的微批流處理框架；第三代是以Flink 為代表的面向流，保證Exactly Once 的實(shí)時(shí)流處理框架.Flink［5］的計(jì)算平臺(tái)可以實(shí)現(xiàn)毫秒級(jí)延遲下每秒處理上億次的消息或者事件；同時(shí)，F(xiàn)link 還提供一種Exactlyonce［6］的一致性語(yǔ)義，保證數(shù)據(jù)的正確性，使Flink大數(shù)據(jù)引擎可以提供金融級(jí)的數(shù)據(jù)處理能力.高吞吐、低延遲的性能使Flink 成為目前流處理的首選.與此同時(shí)，各個(gè)公司處理大數(shù)據(jù)的方式也發(fā)生了很大變化，例如阿里巴巴、滴滴出行、美團(tuán)都大規(guī)模使用Flink 集群，阿里巴巴還開源自己的Blink 給Flink 社區(qū)，給Flink 帶來了更好的性能優(yōu)化以及方便的SQL 環(huán)境.如圖1 所示的流程中，上游是事務(wù)處理、日志、點(diǎn)擊事件等，經(jīng)過Flink 的流處理到達(dá)下游；下游是處理之后的數(shù)據(jù)存儲(chǔ)到數(shù)據(jù)庫(kù)里或直接被應(yīng)用所利用.Flink 在其中可以提供低延遲、高吞吐、Exactly Once 的處理.

圖1 Flink 應(yīng)用場(chǎng)景Fig.1 Flink application scenario

任務(wù)調(diào)度是Flink 很重要的功能.Flink 通過JobManger 任務(wù)調(diào)度器管理Slot，把任務(wù)分配到合適的Slot 等待TaskManager 執(zhí)行.Flink 的任務(wù)調(diào)度圖如圖2 所示，F(xiàn)link 集群?jiǎn)?dòng)后首先會(huì)啟動(dòng)一個(gè)JobManger 和一個(gè)或多個(gè)TaskManager，Client提交任務(wù)給JobManager，JobManager 再調(diào)度任務(wù)到各個(gè)TaskManager 去執(zhí)行.在這個(gè)過程中TaskManager 把心跳和任務(wù)處理信息匯報(bào)給Job-Manager，TaskManager 之間以流的形式進(jìn)行數(shù)據(jù)傳輸.在集群運(yùn)行的過程中，如果流任務(wù)失敗則利用快照加檢查點(diǎn)的形式恢復(fù)，如果批任務(wù)失敗則重新開始.在把任務(wù)分配到具體的Slot 的過程中，優(yōu)先選擇符合Local 屬性的節(jié)點(diǎn).如果有Slot-SharingGroup 限制，則在SlotSharingGroup 里再創(chuàng)建一個(gè)SimpleSlot；如果有CoLocationGroup 限制，則必須在同一個(gè)CoLocationGroup 里創(chuàng)建一個(gè)SimpleSlot；如果沒有上述限制，則從Slot 集合里挑一個(gè).

圖2 Flink 任務(wù)調(diào)度圖Fig.2 Flink task scheduling diagram

所以，F(xiàn)link 的默認(rèn)調(diào)度策略是輪詢，每個(gè)任務(wù)需要去可用的Slot 集合順序選擇一個(gè)Slot 分配給這個(gè)Task.在調(diào)度過程中，屬于同一個(gè)Task 的所有SubTask 不能分配在Slot 里面，因?yàn)槿蝿?wù)需要分布式運(yùn)行，所以不同的子任務(wù)必須分配在不同的Slot 里.

Flink 默認(rèn)的調(diào)度策略把所有節(jié)點(diǎn)認(rèn)作同等性能（這里的性能指CPU 的處理能力、內(nèi)存等），但在實(shí)際的集群搭建部署過程中，集群中的節(jié)點(diǎn)可能性能相差較大.若在這種異構(gòu)的集群中采用輪詢的調(diào)度策略就可能因?yàn)闆]有考慮每個(gè)節(jié)點(diǎn)的不同負(fù)載能力，使性能較差的節(jié)點(diǎn)和性能較好的節(jié)點(diǎn)被分配同樣的任務(wù).性能較差的節(jié)點(diǎn)負(fù)載過多的Task 會(huì)影響整體Job 的運(yùn)行效率，使集群的吞吐降低、延時(shí)增加，延長(zhǎng)整個(gè)Job 的運(yùn)行時(shí)間，因此在異構(gòu)集群中根據(jù)集群各個(gè)節(jié)點(diǎn)的不同性能調(diào)整任務(wù)的分配成為提升異構(gòu)集群整體性能的關(guān)鍵.基于此，本文提出自適應(yīng)負(fù)載均衡算法，提升系統(tǒng)的吞吐量、降低延時(shí)、減少Job 的運(yùn)行時(shí)間，使集群的整體性能有較大的提升.

本文的主要貢獻(xiàn)：

（1）通過對(duì)Hadoop 和Spark 的研究發(fā)現(xiàn)，在異構(gòu)集群中應(yīng)用默認(rèn)的輪詢調(diào)度策略沒有考慮節(jié)點(diǎn)的處理性能，導(dǎo)致集群執(zhí)行效率的下降.本文通過實(shí)驗(yàn)證明Flink 系統(tǒng)中也會(huì)出現(xiàn)這種現(xiàn)象，即異構(gòu)Flink 集群中可能存在由于任務(wù)分配不均衡導(dǎo)致Job 完成時(shí)間被拖長(zhǎng)、吞吐量降低、延時(shí)增加.

（2）提出異構(gòu)集群中平滑加權(quán)輪詢?nèi)蝿?wù)調(diào)度算法（Smooth Weighted Polling Task Scheduling，SWPTS）.根據(jù)集群中每個(gè)節(jié)點(diǎn)的性能權(quán)重Ew，按從大到小的次序，依次給每個(gè)節(jié)點(diǎn)的Slot 分配SubTask.與此同時(shí)，記錄節(jié)點(diǎn)當(dāng)前的權(quán)重Cw，降低Cw最大的節(jié)點(diǎn)來避免有效權(quán)重大的節(jié)點(diǎn)被連續(xù)多次選中，從而使集群在開始調(diào)度時(shí)就能保持負(fù)載均衡.

（3）提出基于蟻群算法的任務(wù)調(diào)度算法（Task Scheduling Algorithm Based on Ant Colony Algorithm，ACTS）.在集群運(yùn)行過程中，當(dāng)集群資源的使用高于預(yù)設(shè)值ε時(shí)則使用ACTS 尋找全局最優(yōu)任務(wù)分配方案，使每個(gè)SubTask 被分配到合適的Slot 上，讓整個(gè)集群的效率達(dá)到最高.

（4）在真實(shí)數(shù)據(jù)集和人工數(shù)據(jù)集上進(jìn)行了實(shí)驗(yàn)分析和驗(yàn)證，結(jié)果表明，在Flink 集群里應(yīng)用SWPTS 和ACTS 確實(shí)對(duì)縮短Job 運(yùn)行時(shí)間、提高吞吐量、降低延時(shí)有很好的效果.

1 相關(guān)工作

在流計(jì)算框架中任務(wù)調(diào)度是很重要的一個(gè)模塊，負(fù)責(zé)把Task 調(diào)度給Slave 的Slot 執(zhí)行，不同的流計(jì)算引擎盡管實(shí)現(xiàn)方式不同，但實(shí)現(xiàn)的功能都是把任務(wù)根據(jù)某種算法分配給指定的Slave 執(zhí)行.

Hadoop 提供可插拔的任務(wù)調(diào)度器，它根據(jù)用戶的需求選擇合適的調(diào)度方案，并可以隨時(shí)切換.Hadoop 的任務(wù)調(diào)度算法有三種.（1）FIFO 調(diào)度器［7］.FIFO 調(diào)度器是最開始被集成到Hadoop里的，Task 按FIFO 的順序進(jìn)入大工作隊(duì)列，Job-Tracker 從工作隊(duì)列里取最先到達(dá)的Task.這種調(diào)度策略沒有考慮作業(yè)的優(yōu)先級(jí)和作業(yè)的大小，但這種策略最容易實(shí)現(xiàn)，也是有效率的.（2）公平調(diào)度器（Fair Scheduler）［8］.公平調(diào)度是一種分配作業(yè)資源的策略，目的是讓所有的作業(yè)隨著時(shí)間的推移都能平均地獲取等同的共享資源，所有Job 享有相同的資源.（3）計(jì)算能力調(diào)度器（Capacity Scheduler）［9］支持多個(gè)隊(duì)列，每個(gè)隊(duì)列可配置一定的資源量并采用FIFO 調(diào)度策略.為防止同一個(gè)用戶的作業(yè)獨(dú)占隊(duì)列中的資源，該調(diào)度器會(huì)對(duì)同一用戶提交的作業(yè)所占的資源量進(jìn)行限定，優(yōu)先執(zhí)行占用最小、優(yōu)先級(jí)高的作業(yè).

Lee and Chung［10］提出一個(gè)針對(duì)Hadoop 的截止時(shí)間約束調(diào)度算法，使用統(tǒng)計(jì)學(xué)方法來測(cè)量數(shù)據(jù)節(jié)點(diǎn)的性能，并基于該信息創(chuàng)建檢查點(diǎn)來監(jiān)視作業(yè)的進(jìn)度；根據(jù)每個(gè)檢查點(diǎn)的作業(yè)進(jìn)度，算法將任務(wù)分配給不同的作業(yè)隊(duì)列.Kc and Anyanwu［11］提出基于工作執(zhí)行代價(jià)模型來滿足用戶規(guī)定的數(shù)據(jù)處理截止日期的調(diào)度算法，這些工作執(zhí)行代價(jià)包括map 和reduce 兩個(gè)階段的運(yùn)行時(shí)間、map 和reduce 輸入數(shù)據(jù)的規(guī)模和分布.

Spark 是當(dāng)下十分流行的流計(jì)算框架.默認(rèn)情況下，Spark 調(diào)度器按照FIFO（先進(jìn)先出）［12］的形式來調(diào)度任務(wù)，每個(gè)工作被分為多個(gè)“階段”（如map 和reduce 語(yǔ)句）.對(duì)于所有可用的資源，第一個(gè)工作的優(yōu)先級(jí)最高，其任務(wù)即被啟動(dòng)；之后是第二個(gè)，依次類推.如果集群不需要隊(duì)列頭的工作，后面的工作將被立刻啟動(dòng)；如果隊(duì)列頭的工作很大，則后面的工作可能被大大推遲.Spark 后來的版本模仿Hadoop 的公平調(diào)度器［13］也添加了公平的調(diào)度策池，不同的工作可以被分到不同的組，每組對(duì)應(yīng)一個(gè)任務(wù)池，不同的任務(wù)池設(shè)置不同的調(diào)度選項(xiàng)（權(quán)重）.

Mao et al［14］提出用機(jī)器學(xué)習(xí)領(lǐng)域的增強(qiáng)學(xué)習(xí)和神經(jīng)網(wǎng)絡(luò)，在無須手動(dòng)設(shè)置最小工作完成時(shí)間的情況下得到一種在指定工作負(fù)載情況下的調(diào)度算法，在這個(gè)過程中設(shè)計(jì)可擴(kuò)展的RL 模型，并發(fā)明RL 訓(xùn)練方法處理連續(xù)到來的隨機(jī)Job.Ren et al［15］設(shè)計(jì)了第一個(gè)分散感知調(diào)度器Hopper，為了提供可拓展性和可預(yù)測(cè)性，Hopper 被設(shè)計(jì)成一個(gè)分散的Job 調(diào)度器，它把資源分配給Job 的同時(shí)也能預(yù)測(cè)拖慢工作進(jìn)度的子任務(wù)，從而采取合理算法降低系統(tǒng)的延時(shí).

Storm 里也有很多對(duì)Storm 的調(diào)度器進(jìn)行改進(jìn)的工作.Peng et al［16］在2015 年提出R-Storm 資源感知調(diào)度器，通過增加最大化資源利用率來提升總吞吐量，同時(shí)最小化網(wǎng)絡(luò)延遲，在任務(wù)調(diào)度時(shí)R-Storm 可以滿足軟資源和硬資源的限制以及最小化組件之間的網(wǎng)絡(luò)通信代價(jià)，在標(biāo)準(zhǔn)的Yahoo基準(zhǔn)測(cè)試下吞吐量提高30%～47%，CPU 利用率提高69%～350%.Chen and Lee［17］發(fā)現(xiàn)Storm 默認(rèn)使用輪詢的算法來分配任務(wù)，這對(duì)異構(gòu)計(jì)算環(huán)境不是最佳，于是提出G-Storm 調(diào)度算法.GStorm 利用集群節(jié)點(diǎn)GPU 來加速整體性能，實(shí)驗(yàn)結(jié)果表明，與原始Storm 調(diào)度算法相比，G-Storm在輕量級(jí)和高負(fù)載拓?fù)渖峡梢詫?shí)現(xiàn)1.65 倍至2.04 倍的性能提升.

目前針對(duì)異構(gòu)集群的負(fù)載均衡算法沒有通用的高效算法，大部分還是通過感應(yīng)集群中某種資源來作出動(dòng)態(tài)調(diào)度，或利用集群的其他計(jì)算能力（如GPU）來負(fù)載均衡，通常存在三個(gè)問題：（1）這些算法只從單一的性能指標(biāo)去實(shí)現(xiàn)負(fù)載均衡，沒有綜合多個(gè)性能指標(biāo)來思考；（2）雖然其中有加權(quán)輪詢的算法，但這種加權(quán)在選擇節(jié)點(diǎn)時(shí)沒有考慮平滑，導(dǎo)致有些權(quán)重大的節(jié)點(diǎn)被連續(xù)選中，造成短時(shí)間內(nèi)局部負(fù)載過重，影響節(jié)點(diǎn)效率；（3）這些算法只考慮負(fù)載均衡，沒有考慮集群負(fù)載均衡算法后已使用的資源高于設(shè)定的閾值時(shí)的處理.

針對(duì)以上問題，本文改進(jìn)和實(shí)現(xiàn)如下：（1）考慮CPU 利用率、內(nèi)存利用率和總內(nèi)存這三個(gè)關(guān)鍵性能指標(biāo)，而非單一指標(biāo)，并且動(dòng)態(tài)地監(jiān)控三個(gè)性能指標(biāo)來作出調(diào)度決策；（2）不僅考慮多個(gè)性能指標(biāo)的加權(quán)，而且在加權(quán)過程中進(jìn)行平滑處理，即避免讓權(quán)重大的節(jié)點(diǎn)連續(xù)被選中造成局部短時(shí)間內(nèi)負(fù)載過重；（3）在集群已使用的資源高于設(shè)定閾值后采用ACTS，能夠在一定的迭代次數(shù)內(nèi)得到全局最優(yōu)任務(wù)分配方案，按照這個(gè)方案調(diào)度任務(wù)來重新負(fù)載均衡，也能使集群處于最佳狀態(tài).

本研究考慮Flink 集群CPU 利用率和內(nèi)存利用率等性能指標(biāo)，在初始調(diào)度階段采用SWPTS，在集群已使用資源高于設(shè)定閾值的情況下采用ACTS，經(jīng)過實(shí)驗(yàn)驗(yàn)證，能夠在Job 的運(yùn)行時(shí)間、延時(shí)、吞吐量等性能指標(biāo)上有明顯提升.

2 基本概念

定義1 內(nèi)存使用率(Mu)這是描述計(jì)算機(jī)的重要性能指標(biāo)之一，可描述集群中已使用的內(nèi)存占用總內(nèi)存的比重.定義如下：

其中，memfree表示剩余的內(nèi)存，memtotal表示總內(nèi)存大小.

定義2 CPU 使用率(Cu)它描述CPU 當(dāng)前被占用的情況，使用率越大表示CPU 越忙，在這種情況下CPU 很難空出時(shí)鐘周期運(yùn)行即將提交該節(jié)點(diǎn)的任務(wù).定義如下：

定義3 有效權(quán)重(Ew)這是在初始調(diào)度時(shí)根據(jù)CPU 使用率、內(nèi)存使用率等性能指標(biāo)得到的一個(gè)綜合權(quán)重，可衡量節(jié)點(diǎn)的總體資源使用情況.計(jì)算方式如下：

定義4 當(dāng)前權(quán)重(Cw)它可以用來衡量運(yùn)行過程中每個(gè)節(jié)點(diǎn)被選中的權(quán)重，每經(jīng)過一次調(diào)度當(dāng)前權(quán)重都會(huì)發(fā)生變化，可以用來挑選最合適的節(jié)點(diǎn).定義如下：

其中，Cwlast表示上次調(diào)度時(shí)的有效權(quán)重.

定義5 任務(wù)執(zhí)行時(shí)間(timeMatrix[ i ][ j ])它表示任務(wù)i分配給節(jié)點(diǎn)j所需的時(shí)間，在算法執(zhí)行的初始時(shí)刻初始化.定義如下：

定義6 最大信息素概率(maxphe)它主要用來確定螞蟻的臨界下標(biāo)（indexbound），即確定一個(gè)臨界螞蟻編號(hào)，在這個(gè)編號(hào)之前的螞蟻選擇信息素最大的節(jié)點(diǎn)從而把任務(wù)分配給該節(jié)點(diǎn)，而這個(gè)節(jié)點(diǎn)之后的螞蟻選擇一個(gè)隨機(jī)的節(jié)點(diǎn)，把任務(wù)分配給該節(jié)點(diǎn).定義如下：

其中，pheMatrix[i]是節(jié)點(diǎn)i的信息素.臨界下標(biāo)的定義如下：

3 算法描述

本節(jié)詳細(xì)介紹集群資源監(jiān)控、SWPTS 和ACTS.集群資源監(jiān)控主要負(fù)責(zé)監(jiān)控集群中每個(gè)節(jié)點(diǎn)的CPU 使用率、內(nèi)存使用率，并把這些數(shù)據(jù)通過Http 協(xié)議［18］發(fā)送給Flink 的JobManager 作為性能數(shù)據(jù)來確定權(quán)重.與傳統(tǒng)的負(fù)載均衡算法不同，SWPTS 和ACTS 是基于集群資源的動(dòng)態(tài)負(fù)載均衡算法.

3.1 集群資源監(jiān)控集群資源監(jiān)控主要負(fù)責(zé)為調(diào)度器提供性能數(shù)據(jù)，F(xiàn)link 集群的JobManager根據(jù)性能數(shù)據(jù)來確定每個(gè)Slave 的權(quán)重，調(diào)度器利用權(quán)重來實(shí)現(xiàn)整體的調(diào)度負(fù)載均衡.集群資源監(jiān)控整體的架構(gòu)如圖3 所示，每個(gè)slave 節(jié)點(diǎn)利用正則表達(dá)式解析/proc 目錄下的cpuinfo 和meminfo 這兩個(gè)文件得到每個(gè)時(shí)刻CPU 的使用率和內(nèi)存使用率，然后通過Socket 通信把數(shù)據(jù)發(fā)送給Master.Master 中的Redis［19］完成性能數(shù)據(jù)的持久化處理，同時(shí)通過Web 技術(shù)對(duì)外提供Http 接口，F(xiàn)link 集群可以通過RPC 調(diào)用獲取集群中各個(gè)節(jié)點(diǎn)的實(shí)時(shí)性能數(shù)據(jù).

圖3 資源調(diào)度架構(gòu)圖Fig.3 The diagram of resource scheduling architecture

3.2 SWPTS 總覽初始化bestInstance和total-Weight.其中bestInstance用來保存最后選中的最好的DataMetric，它在每次循環(huán)過程中都記錄當(dāng)前節(jié)點(diǎn)為止Cw最大的DataMetric.totalWeight則在每次循環(huán)過程中累加每個(gè)DataMetric的Ew，循環(huán)結(jié)束時(shí)，totalWeight保存所有節(jié)點(diǎn)Ew的和.

在每次循環(huán)過程中，每個(gè)DataMetric的Cw都會(huì)加上它自身的Ew，并用total累加當(dāng)前Ew.如果當(dāng)前DataMetric的Cw大于bestInstance的Cw，則記錄bestInstance到目前為止具有最大Cw的DataMetric.

在循環(huán)完成之后，找到當(dāng)前的最優(yōu)的Data-Metric，它有最大的Cw.為使該DataMetric不被重復(fù)選中，需要降低它的Cw.即用bestInstance的Cw減去total，并返回這個(gè)最優(yōu)的bestInstance.具體見算法1.

SWPTS 主要是在Job 運(yùn)行的初始階段把任務(wù)分配給TaskManager 的Slot.為減少頻繁Rpc帶來的時(shí)間影響，在初始的調(diào)度過程中JobManager 沒有一直請(qǐng)求集群資源監(jiān)控的性能數(shù)據(jù)，而是在第一次請(qǐng)求后緩存數(shù)據(jù)，在后續(xù)任務(wù)到來時(shí)SWPTS 利用初始緩存的性能數(shù)據(jù)，再用ACTS選擇一個(gè)節(jié)點(diǎn)的Slot 分配給Task.

以一個(gè)有三個(gè)Slave 節(jié)點(diǎn)的集群為例，假設(shè)weight（Slave1）∶weight（Slave2）∶weight（Slave3）=3∶1∶2，集群的前六次調(diào)度如表1 所示.

可以看到，Slave1 的權(quán)重最大，被調(diào)度的次數(shù)為3；Slave3 的權(quán)重次之，被調(diào)度的次數(shù)為2；Slave2 的權(quán)重最小，被調(diào)度的次數(shù)為1.上述調(diào)度體現(xiàn)了加權(quán)，即權(quán)重大的節(jié)點(diǎn)被調(diào)度選中的次數(shù)也多；同時(shí)也體現(xiàn)了平滑的特點(diǎn)，權(quán)重大的節(jié)點(diǎn)沒有被連續(xù)選中，而是被間斷地選中.綜上所述，SWPTS 能使集群調(diào)度負(fù)載均衡且避免權(quán)重較大的節(jié)點(diǎn)在局部短時(shí)間內(nèi)負(fù)載過重.

表1 Flink 任務(wù)調(diào)度過程Table 1 Flink task scheduling process

3.3 ACTS 總覽SWPTS 主要用在初始調(diào)度，但當(dāng)集群中已使用資源高于設(shè)定閾值ε時(shí)，如果不考慮全局的最優(yōu)調(diào)度，則資源有可能進(jìn)一步降低，負(fù)載持續(xù)不均衡.此時(shí)采用ACTS，可以得到一種全局的任務(wù)分配方案，按照這個(gè)方案把Task分配給指定的Slot 運(yùn)行能使集群處于最佳運(yùn)行狀態(tài)，此時(shí)集群資源的利用率最好.

蟻群算法（Ant Colony Algorithm）是一種模擬螞蟻覓食行為的模擬優(yōu)化算法，由意大利學(xué)者Dorigo and Gambardella［20］于1991 年首先提出，并首先使用在解決TSP（旅行商問題）［21］上.蟻群算法的基本原理如下：

（1）螞蟻在隨意行走的路徑上釋放信息素［22］，這些信息素有利于后面的螞蟻繼續(xù)尋找路徑.

（2）碰到?jīng)]走過的路口就隨機(jī)挑選一條路走，同時(shí)釋放與路徑長(zhǎng)度有關(guān)的信息素.

（3）信息素濃度與路徑長(zhǎng)度成反比.后來的螞蟻再次碰到該路口時(shí)，選擇信息素濃度較高的路徑.

（4）隨著螞蟻覓食過程中信息素的不斷累積，最優(yōu)路徑上的信息素濃度越來越大，讓后續(xù)的螞蟻更多地選擇信息素濃度高的這條路徑.

（5）一段時(shí)間后，蟻群找到最優(yōu)覓食路徑，是一條全局的最短路徑，即距離食物源最近.

圖4 蟻群算法示意圖Fig.4 Ant colony algorithm

ACTS 就是根據(jù)蟻群算法改造的.經(jīng)典的蟻群算法是尋找一條最短路徑，而ACTS 的目標(biāo)是尋找一種全局最優(yōu)分配方案，使Task 的完成時(shí)間最短、資源使用率最好、集群重新負(fù)載均衡.ACTS 以Task 的運(yùn)行時(shí)間為衡量信息素增減的標(biāo)準(zhǔn)，由每只螞蟻把每個(gè)任務(wù)分配給對(duì)應(yīng)的節(jié)點(diǎn).由于分布式并行的特點(diǎn)，每只螞蟻取運(yùn)行時(shí)間最長(zhǎng)的Task 所需要的時(shí)間作為Job 的完成時(shí)間.

算法的整體框架如算法2 所示，主要包括三部分：第一部分是初始化任務(wù)集合，第二部分是初始化信息素矩陣，第三部分是迭代搜索.

步驟1，初始化任務(wù)矩陣.timeMatrix[i][j]表示任務(wù)i分配給節(jié)點(diǎn)j所需的時(shí)間.根據(jù)任務(wù)集合tasks 和節(jié)點(diǎn)集合nodes，應(yīng)用式（5）計(jì)算每個(gè)任務(wù)在每個(gè)節(jié)點(diǎn)完成需要的時(shí)間.表2 是一個(gè)示例，Task[ 1 ]，Task[ 2 ]，Task[ 3 ]對(duì)應(yīng)的數(shù)據(jù)規(guī)模大小分別是10，8，6；Node[ 1 ]，Node[ 2 ]，Node[ 3 ]對(duì)應(yīng)的處理能力分別為2，2，1；每個(gè)任務(wù)在每個(gè)節(jié)點(diǎn)完成需要的時(shí)間如表2 所示.

表2 初始化任務(wù)矩陣Table 2 Initialization task matrix

步驟2，初始化信息素矩陣.將負(fù)載均衡調(diào)度過程中的一次任務(wù)分配看作一條路徑，因此pheromoneMatrix[i][j]表示將任務(wù)i分配給節(jié)點(diǎn)j這條路徑的信息素濃度.初始化信息素矩陣，將所有值置為1，因?yàn)槌跏紩r(shí)所有的路徑都沒有螞蟻選擇，默認(rèn)為1.

步驟3，迭代搜索.這是ACTS 最關(guān)鍵的一步，由三部分組成：迭代分配任務(wù)、計(jì)算每只螞蟻的任務(wù)處理時(shí)間、更新信息素.整個(gè)迭代搜索的算法框架如算法3 所示.

算法3 中，第2 行表示初始化Path_AllAnt，它是一個(gè)三維數(shù)組，用來存保第iteratorCount次迭代過程中第antCount只螞蟻將i個(gè)任務(wù)分配給j個(gè)節(jié)點(diǎn)處理.第4 行是初始化Path_OneAnt，表示第antCount只螞蟻將i任務(wù)分配給j節(jié)點(diǎn)處理.第5～8 行表示循環(huán)每個(gè)任務(wù)，為每個(gè)任務(wù)通過任務(wù)分配函數(shù)asignTask 分配到一個(gè)節(jié)點(diǎn)node，給相應(yīng)的Path_OneAnt，Path_AllAnt賦值.第9 行和第10 行是計(jì)算本次迭代中所有螞蟻的任務(wù)處理時(shí)間，并將所有螞蟻的任務(wù)處理時(shí)間加入總結(jié)果集resultData.第11 行是更新信息素.最后返回resultData.

在整個(gè)蟻群算法中，共進(jìn)行iteratorNum次迭代.每次迭代都會(huì)產(chǎn)生當(dāng)前的最優(yōu)分配策略，即“局部最優(yōu)解”，迭代的次數(shù)越多，局部最優(yōu)解就越接近全局最優(yōu)解.但迭代次數(shù)過多會(huì)造成調(diào)度器大量的時(shí)間和性能開銷，無法滿足海量任務(wù)的調(diào)度；而迭代次數(shù)太少，則得到的可能不是全局最優(yōu)解.本文采用固定迭代次數(shù)為100 次.

任務(wù)分配函數(shù)負(fù)責(zé)將一個(gè)指定的任務(wù)按照某種策略分配給某一節(jié)點(diǎn)處理.分配策略有兩種：（1）按信息素濃度分配，即是將任務(wù)分配給本行中信息素濃度最高的節(jié)點(diǎn)處理.例如：當(dāng)前的任務(wù)編號(hào)是taskCount，當(dāng)前的信息素濃度矩陣是phe-romoneMatrix，則任務(wù)會(huì)分配給pheromoneMatrix[taskCount]這一行中信息素濃度最高的節(jié)點(diǎn).（2）隨機(jī)分配，即將任務(wù)隨意分配給某個(gè)節(jié)點(diǎn)處理，一般根據(jù)螞蟻的編號(hào)antNum來選擇.boundPointMatrix[i]=5 表示編號(hào)為0～5 的螞蟻在分配任務(wù)i的時(shí)候采用“按信息素濃度”的方式分配，即將任務(wù)i分配給信息素濃度最高的節(jié)點(diǎn)處理；而編號(hào)為6～9 的螞蟻在分配任務(wù)i時(shí)，采用隨機(jī)分配策略.

當(dāng)閥門結(jié)構(gòu)是非平衡式軟密封單座結(jié)構(gòu)時(shí)，閥芯導(dǎo)向在閥門套筒內(nèi)且全行程導(dǎo)向，從而可以保證閥門打開的時(shí)候介質(zhì)在閥芯側(cè)向流動(dòng)，該處閥芯外表面不參與密封。軟密封閥芯鑲嵌在閥芯體內(nèi)，閥芯密封面在凹槽底部，閥座密封面使用的是一個(gè)凸臺(tái)，兩者接觸時(shí)為平面密封。當(dāng)閥門關(guān)到底部處于關(guān)閉狀態(tài)時(shí)，閥座的凸起表面嵌入軟密封環(huán)中，該處的密封材料變形包圍住閥座凸起表面，由此確?？煽康拿芊庑Ч?，同時(shí)閥桿也不用承受過大推力。由于整個(gè)密封墊片鑲嵌在凹槽內(nèi)，可以使得閥芯閥座接觸后的軟密封材料PTFE變形得到有效控制，從而保證閥芯閥座在有效行程內(nèi)密封可靠。

蟻群算法有三個(gè)問題需要注意：

（1）計(jì)算任務(wù)處理問題.由于Flink 集群里任務(wù)都是并行運(yùn)行的，因此在計(jì)算任務(wù)處理時(shí)通常以最晚完成的任務(wù)的時(shí)間為整個(gè)Job 的完成時(shí)間.

（2）更新信息素問題.將所有路徑的信息素降低m%，表示信息素的揮發(fā)；找出所有螞蟻的最短路徑，則該路徑的信息素提升n%，表示該路徑是最短路徑，信息素不斷提升.

（3）更新boundPointMatrix問題.bound-PointMatrix表示臨界螞蟻下標(biāo)集合，在該下標(biāo)之前的螞蟻選擇信息素濃度最高的節(jié)點(diǎn)分配，在該下標(biāo)之后的螞蟻選擇隨機(jī)一個(gè)節(jié)點(diǎn)分配，計(jì)算方式如式（6）和式（7）所示.

4 實(shí)驗(yàn)分析

本文的調(diào)度器及算法均已在Flink 中實(shí)現(xiàn).通過修改Flink runtime 包下面的調(diào)度模塊，可添加數(shù)據(jù)HTTP Api 數(shù)據(jù)訪問和數(shù)據(jù)解析器.整個(gè)系統(tǒng)依賴Redis 和maskmonitor 性能監(jiān)視應(yīng)用.使用SWPTS 和使用默認(rèn)調(diào)度算法的Flink 系統(tǒng)在延時(shí)、吞吐量、運(yùn)行時(shí)間方面做了對(duì)比和分析，并在不同的數(shù)據(jù)集和不同并行度上進(jìn)行了實(shí)驗(yàn)驗(yàn)證.實(shí)驗(yàn)使用Flink 自帶的實(shí)例WordCount.由于臨界資源閾值是個(gè)經(jīng)驗(yàn)值，不同閾值效率對(duì)比如圖5.根據(jù)圖5 可以看出閾值為80%效率最高，因此本文實(shí)驗(yàn)?zāi)J(rèn)閾值設(shè)定為80%.

圖5 Flink 臨界閾值效率對(duì)比Fig.5 Comparison of Flink critical threshold efficiency

4.1 數(shù)據(jù)集針對(duì)WordCount 使用真實(shí)數(shù)據(jù)集和模擬數(shù)據(jù)集.真實(shí)數(shù)據(jù)集是TPC-C［23］，用九個(gè)表生成模擬五種事務(wù)處理，產(chǎn)生三個(gè)大數(shù)據(jù)集，模擬真實(shí)場(chǎng)景下的批計(jì)算.此外，自己寫程序生成三個(gè)只包括字符串的模擬數(shù)據(jù)集，用以擴(kuò)充實(shí)驗(yàn)測(cè)試.數(shù)據(jù)集的來源和規(guī)模如表3 所示.

表3 數(shù)據(jù)集規(guī)模Table 3 The size of datasets

4.2 實(shí)驗(yàn)環(huán)境及配置SWPTS 和ACTS 均基于Flink1.4.2，編程語(yǔ)言為Java.實(shí)驗(yàn)所用的集群硬件配置和參數(shù)如下：

實(shí)驗(yàn)環(huán)境：分布式集群由一臺(tái)服務(wù)器和兩臺(tái)虛擬主機(jī)構(gòu)成，兩臺(tái)虛擬主機(jī)模擬異構(gòu)集群的效果，服務(wù)器主機(jī)為Master 節(jié)點(diǎn)，兩臺(tái)虛擬機(jī)為Slave 節(jié)點(diǎn).

Master 節(jié)點(diǎn)的配置如下：CPU 為Intel（R）Core（TM）i7-6700，4 Core，主頻3.40 GHz；內(nèi)存為64 GB 2133 MHz；機(jī)械硬盤為WDC WD10 EZEX -08WN4A0 1 TB；編程環(huán)境為IntelliJ IDEA 2018，Maven，Git；操作系統(tǒng)為Ubuntu 16.04.5；Flink 版本1.4.2，JDK 版本1.8.0_151，Hadoop 版本2.7.5.

Slave1 的配置如下：CPU 為Intel（R）Core（TM）i5-4690，4 Core，主頻3.50 GHz；內(nèi)存為4 GB 2133 MHz；機(jī)械硬盤60 GB；編程環(huán)境為IntelliJIDEA 2018，Maven，Git；操作系統(tǒng)為Ubuntu 16.04.5；Flink 版本1.4.2，JDK 版本1.8.0_151，Hadoop 版本2.7.5.

Slave2 的配置如下：CPU 為Intel（R）Core（TM）i5-4690，2 Core，主頻3.50 GHz；內(nèi)存為2 GB 2133 MHz；機(jī)械硬盤60 GB；編程環(huán)境為IntelliJ IDEA 2018，Maven，Git；操作系統(tǒng)為Ubuntu 16.04.5；Flink 版本1.4.2，JDK 版本1.8.0_151，Hadoop 版本2.7.5.

4.3 實(shí)驗(yàn)結(jié)果及分析通過修改Flink1.4.2 的默認(rèn)任務(wù)調(diào)度器為自適應(yīng)任務(wù)調(diào)度器，新增SWPTS 和ACTS 兩種負(fù)載均衡調(diào)度算法.在異構(gòu)集群中不同的數(shù)據(jù)集下分別與Flink1.4.2 在運(yùn)行時(shí)間、延時(shí)、吞吐量［24］方面做性能對(duì)比，可以看出改進(jìn)之后的算法在三個(gè)方面都有所提升.

4.3.1 運(yùn)行時(shí)間對(duì)比分析通常運(yùn)行時(shí)間是一個(gè)可以讓用戶對(duì)比算法性能的最直觀的指標(biāo).運(yùn)行快說明算法節(jié)省了Job 的完成時(shí)間，加快了用戶響應(yīng)時(shí)間，對(duì)提升性能非常重要.SWPTS 和ACTS 能將任務(wù)盡可能多地分配給資源豐富的節(jié)點(diǎn)，應(yīng)用到自適應(yīng)任務(wù)調(diào)度器中能使資源豐富的節(jié)點(diǎn)更多地接收Task，比Flink 默認(rèn)調(diào)度器的負(fù)載更均衡，運(yùn)行時(shí)間也相應(yīng)縮短.

在并行度為8 和16 時(shí)，在自己程序生成的數(shù)據(jù)集（圖6）和TPC-C 數(shù)據(jù)集（圖7）上進(jìn)行對(duì)比實(shí)驗(yàn)，可以看到，應(yīng)用自適應(yīng)任務(wù)調(diào)度器（LoadbalanceFlink）之后的運(yùn)行時(shí)間比默認(rèn)任務(wù)調(diào)度器（NativeFlink）平均減少8%左右，這是因?yàn)榍罢邔⑷蝿?wù)均勻分配，資源多的節(jié)點(diǎn)可以完成盡可能多的Task，縮短了整體的完成時(shí)間，而默認(rèn)任務(wù)調(diào)度器讓資源少的節(jié)點(diǎn)完成和其他節(jié)點(diǎn)同等的任務(wù)，拖慢了整體工作的完成進(jìn)度.

4.3.2 吞吐量對(duì)比分析吞吐量（Throughput）指單位時(shí)間內(nèi)由計(jì)算引擎成功處理的數(shù)據(jù)量，反映系統(tǒng)的負(fù)載能力.吞吐量常用于資源規(guī)劃，也能協(xié)助分析系統(tǒng)性能瓶頸，從而進(jìn)行相應(yīng)的資源調(diào)整以保證系統(tǒng)達(dá)到用戶要求的處理能力.

圖6 Flink 運(yùn)行時(shí)間對(duì)比(自定義數(shù)據(jù)集)Fig.6 Flink runtime on custom dataset

圖7 Flink 運(yùn)行時(shí)間對(duì)比(TPC-C 數(shù)據(jù)集)Fig.7 Flink runtime on TPC-C dataset

實(shí)驗(yàn)使用阿里巴巴提供的advertising 工具，它是標(biāo)準(zhǔn)流測(cè)試工具yahoo stream benchmark 的簡(jiǎn)化版.測(cè)試原理是隨機(jī)產(chǎn)生兩個(gè)廣告流，把a(bǔ)d_id 相同的join 起來存放到Redis 里，通過單位時(shí)間內(nèi)在Redis 讀到多少條數(shù)據(jù)來計(jì)算吞吐量.計(jì)算如式（8）所示：

其中，currentNum代表當(dāng)前讀到的數(shù)據(jù)編號(hào)，即已經(jīng)讀到多少條數(shù)據(jù)；lastNum表示前一次讀到的數(shù)據(jù)編號(hào)；currentTime表示當(dāng)前時(shí)間，lastTime表示上一次的時(shí)間.

從實(shí)驗(yàn)結(jié)果（圖8）可以看出，改進(jìn)之后的Flink 自適應(yīng)負(fù)載均衡算法比默認(rèn)任務(wù)調(diào)度算法的吞吐量更高，原因同前，本文算法能使負(fù)載均衡，即資源多的節(jié)點(diǎn)負(fù)載較多的任務(wù)，則整個(gè)集群在單位時(shí)間內(nèi)能完成更多的任務(wù)，吞吐量增大.

圖8 Flink 不同并行度下的吞吐量對(duì)比Fig.8 Throughput of Flink with different parallelism

4.3.3 延時(shí)對(duì)比分析延時(shí)（latency）指數(shù)據(jù)從進(jìn)入數(shù)據(jù)窗口的時(shí)間到真正被處理的時(shí)間間隔，單位為毫秒（ms），反映系統(tǒng)處理的實(shí)時(shí)性.金融交易分析等大量實(shí)時(shí)計(jì)算業(yè)務(wù)對(duì)延遲要求較高，因?yàn)檠訒r(shí)越小，數(shù)據(jù)實(shí)時(shí)性越強(qiáng).

實(shí)驗(yàn)使用阿里巴巴提供的advertising 工具，原理是隨機(jī)產(chǎn)生兩個(gè)廣告流，把a(bǔ)d_id 相同的join起來存放到Redis 里.計(jì)算如式（9）所示：

其中，handleTime表示某條記錄實(shí)際被處理的時(shí)間，windowTime表示流里面該記錄屬于的時(shí)間窗口開始時(shí)間.

從實(shí)驗(yàn)結(jié)果（圖9）可以看出，本文改進(jìn)之后的Flink 自適應(yīng)負(fù)載均衡算法比默認(rèn)的調(diào)度算法延時(shí)更小，原因亦同前，本文算法能夠?qū)崿F(xiàn)負(fù)載均衡，使資源多的節(jié)點(diǎn)能負(fù)載較多的任務(wù)，則每個(gè)任務(wù)在被處理之前需要等待的時(shí)間也相應(yīng)變短，即延時(shí)變小.

圖9 Flink 不同并行度的延時(shí)對(duì)比Fig.9 Delay of Flink with different parallelism

上述在運(yùn)行時(shí)間、吞吐量、延時(shí)三方面的實(shí)驗(yàn)表明：SWPTS 和ACTS 改變了任務(wù)的默認(rèn)分配策略，可以盡量按動(dòng)態(tài)資源的大小將任務(wù)優(yōu)先分配給資源較多的節(jié)點(diǎn)，解決異構(gòu)集群負(fù)載不均衡的問題.

5 總結(jié)

本文提出的自適應(yīng)負(fù)載均衡算法由平滑加權(quán)輪詢?nèi)蝿?wù)調(diào)度算法（SWPTS）和基于蟻群算法的任務(wù)調(diào)度算法（ACTS）組成.經(jīng)過實(shí)驗(yàn)驗(yàn)證，在異構(gòu)Flink 集群的環(huán)境下，自適應(yīng)負(fù)載均衡算法的運(yùn)行時(shí)間、吞吐量和延時(shí)與默認(rèn)調(diào)度算法相比都有所提升.在運(yùn)行初期，利用SWPTS 負(fù)載均衡，使得任務(wù)在初始分配的時(shí)候負(fù)載均衡.在運(yùn)行過程中，當(dāng)集群已使用資源高于設(shè)定閾值時(shí)，采用ACTS 尋找一種全局最優(yōu)分配方案，也能重新均衡負(fù)載.等已使用資源低于設(shè)定閾值時(shí)，則繼續(xù)采用之前的算法進(jìn)行調(diào)度.