張譯天 于炯 魯亮 李梓楊

摘 要：新型大數(shù)據(jù)流式計算框架Apache Heron默認使用輪詢調(diào)度算法進行任務(wù)調(diào)度，忽略了拓撲運行時狀態(tài)以及任務(wù)實例間不同通信方式對系統(tǒng)性能的影響。針對這個問題，提出Heron環(huán)境下流分類任務(wù)調(diào)度策略（DSC-Heron），包括流分類算法、流簇分配算法和流分類調(diào)度算法。首先通過建立Heron作業(yè)模型明確任務(wù)實例間不同通信方式的通信開銷差異;其次基于流分類模型，根據(jù)任務(wù)實例間實時數(shù)據(jù)流大小對數(shù)據(jù)流進行分類;最后將相互關(guān)聯(lián)的高頻數(shù)據(jù)流整體作為基本調(diào)度單元構(gòu)建任務(wù)分配計劃，在滿足資源約束條件的同時盡可能多地將節(jié)點間通信轉(zhuǎn)化為節(jié)點內(nèi)通信以最小化系統(tǒng)通信開銷。在包含9個節(jié)點的Heron集群環(huán)境下分別運行SentenceWordCount、WordCount和FileWordCount拓撲，結(jié)果表明DSC-Heron相對于Heron默認調(diào)度策略，在系統(tǒng)完成時延、節(jié)點間通信開銷和系統(tǒng)吞吐量上分別平均優(yōu)化了8.35%、7.07%和6.83%;在負載均衡性方面，工作節(jié)點的CPU占用率和內(nèi)存占用率標準差分別平均下降了41.44%和41.23%。實驗結(jié)果表明，DSC-Heron對測試拓撲的運行性能有一定的優(yōu)化作用，其中對接近真實應(yīng)用場景的FileWordCount拓撲優(yōu)化效果最為顯著。

關(guān)鍵詞：大數(shù)據(jù);流式計算;Apache Heron;任務(wù)調(diào)度;數(shù)據(jù)流分類;通信開銷

中圖分類號：TP311

文獻標志碼：A

文章編號：1001-9081（2019）04-1106-011

0?引言

隨著云計算、物聯(lián)網(wǎng)、移動互聯(lián)、社交媒體和人工智能等新型信息技術(shù)和應(yīng)用模式的不斷發(fā)展，數(shù)據(jù)正以前所未有的方式推動人類社會進入大數(shù)據(jù)時代[1]。國際數(shù)據(jù)公司（International Data Corporation， IDC）發(fā)布的白皮書《數(shù)據(jù)時代2025》中顯示，預(yù)計到2025年全球互聯(lián)網(wǎng)數(shù)據(jù)總量達162ZB，其中超過1/4的數(shù)據(jù)為實時數(shù)據(jù)，而物聯(lián)網(wǎng)實時數(shù)據(jù)將占這部分數(shù)據(jù)的95%以上[2]。面對實時大數(shù)據(jù)具有的實時性、易失性、突發(fā)性、無序性和無限性的新特征[3]，傳統(tǒng)的MapReduce等批處理方式不再適用，分布式大數(shù)據(jù)流式計算應(yīng)運而生。流數(shù)據(jù)處理摒棄了傳統(tǒng)批處理中對數(shù)據(jù)先存儲后計算的方式，將數(shù)據(jù)以數(shù)據(jù)流的形式在數(shù)據(jù)產(chǎn)生初期進行計算，使用可靠傳輸模式而不對計算中間結(jié)果進行存儲，在對數(shù)據(jù)分析實時性要求較高的場景中得到了廣泛的應(yīng)用，并不斷融入實時圖像識別、智慧城市等人工智能的發(fā)展中。

目前，在典型的流數(shù)據(jù)處理系統(tǒng)（例如，Storm[4]、Flink[5]）中，默認調(diào)度算法常采用靜態(tài)輪詢調(diào)度算法。該算法實現(xiàn)簡單，在拓撲提交時對拓撲進行任務(wù)分配，拓撲運行期間任務(wù)分配狀態(tài)不再發(fā)生改變;但靜態(tài)輪詢調(diào)度算法在進行任務(wù)分配時，僅考慮資源的可滿足性將任務(wù)均勻分配到各個工作節(jié)點，未考慮拓撲中任務(wù)間的關(guān)聯(lián)關(guān)系和節(jié)點間的通信開銷，會對集群性能產(chǎn)生一定影響。針對這一問題，國內(nèi)外研究人員針對不同流式計算平臺提出在線調(diào)度算法，在拓撲運行過程中通過實時監(jiān)測集群運行狀態(tài)，對已分配的任務(wù)進行重調(diào)度或調(diào)整，以使集群具有更高效的性能。針對Storm環(huán)境，文獻[6]提出自適應(yīng)調(diào)度策略，分為離線調(diào)度和在線調(diào)度。其中在線調(diào)度通過實時監(jiān)測拓撲運行過程中CPU負載和工作節(jié)點間數(shù)據(jù)流流量等數(shù)據(jù)，依次將通信開銷較大的一對任務(wù)調(diào)度到CPU負載較小的工作節(jié)點中。該策略可以較好地降低Storm的通信開銷，但實驗中使用的是自定義鏈式拓撲，缺乏一定的代表性。文獻[7]提出流量感知在線調(diào)度策略T-Storm，旨在最小化進程間和工作節(jié)點間通信開銷，同時實現(xiàn)了細粒度的任務(wù)分配控制，可以通過調(diào)整預(yù)設(shè)參數(shù)控制工作節(jié)點數(shù)量。但該策略忽略了直接通信的一對任務(wù)之間的數(shù)據(jù)流情況且調(diào)度執(zhí)行開銷較大。文獻[8]提出資源感知調(diào)度策略R-Storm，通過將CPU、內(nèi)存和網(wǎng)絡(luò)帶寬資源映射為三維空間向量，使用最小化向量距離的方法尋找任務(wù)和工作節(jié)點的分配關(guān)系，從而最大化資源利用并提高系統(tǒng)吞吐量。該策略充分考慮了集群資源的有效利用，但拓撲中各任務(wù)的資源需求由編程人員設(shè)定而非實時監(jiān)測獲得，很難應(yīng)用于資源需求變化較大的在線調(diào)度。文獻[9]提出一種異構(gòu)環(huán)境下的任務(wù)遷移策略TMSH-Storm，將超出閾值節(jié)點中的阻尼線程細粒度地遷移至滿足條件的目標節(jié)點，避免了資源溢出后的任務(wù)重部署，可以較好地降低調(diào)度時延和節(jié)點間通信開銷。針對Flink環(huán)境，文獻[10]提出流網(wǎng)絡(luò)的流式計算動態(tài)任務(wù)調(diào)度策略。該策略通過建立流網(wǎng)絡(luò)模型，基于最大流算法使模型在滿足延遲約束前提下提高集群的實際吞吐量，能在一定程度上解決輸入速率增加階段出現(xiàn)的計算延遲升高問題。

但該策略僅關(guān)注集群輸入速率急劇上升階段的性能優(yōu)化，且沒有考慮流網(wǎng)絡(luò)容量動態(tài)變化的問題。針對其他流式計算系統(tǒng)，文獻[11]從資源分配的角度出發(fā)，對多代分布式流處理系統(tǒng)的彈性資源調(diào)度機制進行對比并提出未來研究方向。文獻[12]基于拓撲任務(wù)與集群資源的分配與映射關(guān)系提出模型驅(qū)動的調(diào)度策略，為流處理系統(tǒng)提供高效的資源利用率和吞吐量。文獻[13-14]側(cè)重于流處理系統(tǒng)的穩(wěn)定性，分別提出一種基于隊列的穩(wěn)定性預(yù)測模型和一種穩(wěn)定的在線調(diào)度策略以優(yōu)化系統(tǒng)性能。

Heron[15-17]是Twitter為解決其上一代分布式流處理平臺Storm在可擴展性、可調(diào)試性、可管理性以及集群資源共享等方面問題，而構(gòu)建的新一代分布式流處理平臺[15]。Heron在2015年已經(jīng)取代Storm，成為Twitter實際使用的實時數(shù)據(jù)處理系統(tǒng)[17]，現(xiàn)已進入Apache開源項目孵化器。Heron在設(shè)計層面對Storm進行多方面優(yōu)化的同時，其默認調(diào)度策略仍使用輪詢（Round Robin，RR）算法進行任務(wù)實例分配。該算法根據(jù)拓撲中任務(wù)實例資源需求和容器資源需求，創(chuàng)建用于調(diào)度的任務(wù)分配計劃，然后交由Heron調(diào)度器（Aurora[18]、Mesos[19]、YARN[20]等）分配至工作節(jié)點。對于Twitter目前使用的Aurora調(diào)度器，Heron將一個拓撲對應(yīng)于Aurora中一個工作（Job），一個容器與其中分配的任務(wù)實例作為Aurora的一個資源分配單元，由Aurora負責將這些單元分配至集群中滿足資源需求的工作節(jié)點中運行。在配置了Aurora調(diào)度器的Heron集群中，工作節(jié)點資源分配由Mesos負責進行，其采用DRF（Dominant Resource Fairness）資源分配策略為Aurora框架中提交的任務(wù)尋找集群中注冊的合適工作節(jié)點進行任務(wù)分配。

在Heron調(diào)度策略和任務(wù)分配過程中仍存在如下問題：1）任務(wù)分配沒有考慮任務(wù)實例間通信開銷，忽略了節(jié)點間通信和節(jié)點內(nèi)通信的差異;2）任務(wù)分配僅考慮了任務(wù)實例和工作節(jié)點資源的約束關(guān)系，且算法采用資源最大化對齊的方式容易造成資源浪費;3）僅提供了靜態(tài)調(diào)度策略，無法針對運行狀態(tài)下的拓撲進行實時調(diào)度。

針對Heron默認調(diào)度策略中存在的上述問題，本文提出流分類調(diào)度策略（task scheduling strategy based on Data Stream Classification in Heron， DSC-Heron），主要工作如下：1）提出Heron作業(yè)模型，將拓撲中任務(wù)實例通信方式劃分為節(jié)點間、容器間和實例間通信，明確不同通信方式間的通信開銷差異。

2）以Heron作業(yè)模型為基礎(chǔ)，提出資源約束模型、最優(yōu)通信開銷模型和流分類模型，作為提出DSC-Heron任務(wù)調(diào)度策略的理論依據(jù)。

3）提出流分類調(diào)度策略，包括流分類算法、流簇分配算法和流分類調(diào)度算法。該策略首先根據(jù)數(shù)據(jù)流實時大小對數(shù)據(jù)流進行分類，然后以高頻數(shù)據(jù)流關(guān)聯(lián)的高頻流簇為單位進行任務(wù)調(diào)度，使得拓撲的任務(wù)分配在滿足資源約束條件的同時最小化節(jié)點間通信開銷。

4）使用Heron示例拓撲和自定義拓撲對流分類調(diào)度策略進行性能評估。實驗結(jié)果表明，相較于Heron默認調(diào)度策略，DSC-Heron在系統(tǒng)完成時延、節(jié)點間數(shù)據(jù)流大小和吞吐量方面均有一定的優(yōu)化效果。

1?Heron作業(yè)模型

在Heron中，拓撲是用戶定義流式作業(yè)的抽象，使用有向無環(huán)圖（Directed Acyclic Graph， DAG）表示，由組件和數(shù)據(jù)流構(gòu)成。組件分為Spout和Bolt兩類：Spout為數(shù)據(jù)源編程單元，可以從Kafka[21]、DistributedLog[22]或HDFS（Hadoop Distributed File System）中不間斷地讀取數(shù)據(jù)，以數(shù)據(jù)流的形式傳遞給下游組件;Bolt為數(shù)據(jù)流處理單元，用于實現(xiàn)數(shù)據(jù)處理邏輯。數(shù)據(jù)流是對組件間以元組形式進行數(shù)據(jù)傳遞的抽象，可以通過不同的流組模式定義元組的傳遞和分組方式。由此定義拓撲邏輯模型如下。

Heron中為提高系統(tǒng)并行度和數(shù)據(jù)處理速度可以為拓撲中每個組件定義運行并行度，并在拓撲提交時為每個組件創(chuàng)建相應(yīng)數(shù)量的任務(wù)實例。每個任務(wù)實例運行一個Java進程且運行在一個JVM中。由此定義拓撲實例模型如下。

根據(jù)實例分配模型可知，Heron任務(wù)實例之間的通信需要經(jīng)過所在容器中的SM進行路由。集群中各個SM彼此連接形成一個全連接網(wǎng)絡(luò)，將復(fù)雜度為O（N（N-1）/2）的N個任務(wù)實例間通信，通過M個SM簡化為O（M（M-1）/2），其中NM。因此，在Heron集群中存在三種不同的通信方式：1）工作節(jié)點間通信，即集群不同物理工作節(jié)點間任務(wù)實例的通信方式。這種通信方式中，數(shù)據(jù)流需要經(jīng)過源任務(wù)實例、源任務(wù)所屬SM、目的任務(wù)所屬SM和目的任務(wù)實例進行傳輸，會占用大量的網(wǎng)絡(luò)帶寬資源，是集群中通信開銷最大的一種通信方式。如圖3中，任務(wù)實例Ia和Id1間的通信即屬于工作節(jié)點間通信。

2）容器間通信，即同一工作節(jié)點、不同容器中任務(wù)實例之間的通信方式。這種通信方式不占用網(wǎng)絡(luò)帶寬，數(shù)據(jù)流在同一節(jié)點的不同任務(wù)間進行傳遞，但仍需要經(jīng)過源任務(wù)實例、目的任務(wù)實例以及各自所屬的SM，屬于進程間通信且通信開銷較小。如圖3中，任務(wù)實例Id2和Ic間的通信即為容器間通信方式。

3）實例間通信，即同一工作節(jié)點且同一容器中任務(wù)實例間的直接通信。這種通信方式在一個容器中進行，數(shù)據(jù)流只經(jīng)過一個SM進行傳輸，是三種通信方式中通信開銷最小的一種。由于在Heron中每個任務(wù)實例都是一個Java進程，因此這種通信方式也屬于進程間通信，但由于減少了數(shù)據(jù)流經(jīng)過的SM數(shù)量，因此通信開銷較容器間通信開銷小。如圖3中，任務(wù)實例Ib和Ig間即屬于實例間通信方式。

2?問題建模與分析

本章在Heron作業(yè)模型的基礎(chǔ)之上提出資源約束模型、最優(yōu)通信開銷模型和流分類模型。其中資源約束模型為任務(wù)分配的基礎(chǔ)條件;最優(yōu)通信開銷模型論證了節(jié)點間和節(jié)點內(nèi)通信開銷的相互關(guān)系，為最小化通信開銷的任務(wù)調(diào)度過程提供依據(jù);流分類模型定義了拓撲中數(shù)據(jù)流分類的理論基礎(chǔ)。

2.1?資源約束模型

在Heron應(yīng)用環(huán)境中，為使各個工作節(jié)點不會出現(xiàn)滿負荷運行狀態(tài)以影響集群運行性能，需要為每個工作節(jié)點預(yù)留少量的計算資源。因此，在上述資源約束中，α、 β、γ分別為集群管理人員為CPU、內(nèi)存和網(wǎng)絡(luò)帶寬資源設(shè)定的資源閾值參數(shù)，該參數(shù)可根據(jù)集群資源情況進行設(shè)置。

2.2?最優(yōu)通信開銷模型

2.3?流分類模型

3?流分類調(diào)度策略

本章基于上述模型提出流分類調(diào)度策略（DSC-Heron），包括流分類算法、流簇分配算法和流分類調(diào)度算法。其中流分類算法以流分類模型為基礎(chǔ)，以數(shù)據(jù)流實時大小為依據(jù)對數(shù)據(jù)流進行分類;流簇分配算法以高頻流簇為基本單元進行任務(wù)分配;流分類調(diào)度算法對不同類別的數(shù)據(jù)流依次進行調(diào)度，最終完成目標任務(wù)分配計劃的構(gòu)建。

3.1?流分類算法

通過使用3.4節(jié)提出的負載監(jiān)測模塊對集群中任務(wù)實例以及實例間數(shù)據(jù)流大小進行監(jiān)測，實時獲取拓撲中任務(wù)實例間數(shù)據(jù)流大小，得到拓撲的數(shù)據(jù)流集合S={s1，2，s1，3，…，sI-1，I}作為輸入;然后根據(jù)流分類模型將集合S劃分為高頻數(shù)據(jù)流集Hf、中頻數(shù)據(jù)流集Mf和低頻數(shù)據(jù)流集Nf。具體算法如算法1所示。

算法1?流分類算法。

輸出?高頻數(shù)據(jù)流集Hf;中頻數(shù)據(jù)流集Mf;低頻數(shù)據(jù)流集Nf。

步驟1?根據(jù)S中各個數(shù)據(jù)流大小vij，kl，使用式（8）計算數(shù)據(jù)流總量V;使用式（9）計算數(shù)據(jù)流平均值;使用式（10）計算sij，kl的絕對偏差值;使用式（11）計算數(shù)據(jù)流總絕對偏差ΔV;使用式（12）平均偏差Δ。

步驟2?根據(jù)vij，kl大小降序排序S并進行遍歷，進行如下判斷：若數(shù)據(jù)流sij，kl滿足式（13），將其加入高頻數(shù)據(jù)流集Hf;若滿足式（14）或式（15），將其加入中頻數(shù)據(jù)流集合Mf;若滿足式（16），將其加入低頻數(shù)據(jù)流集合Nf。

步驟3?返回Hf、Mf和Nf。

流分類算法步驟1中包括兩次對數(shù)據(jù)流集合S的遍歷。第一次遍歷數(shù)據(jù)流集合S，計算出拓撲數(shù)據(jù)流總量V，進而可以計算數(shù)據(jù)流平均值。第二次遍歷數(shù)據(jù)流集合S，由數(shù)據(jù)流平均值計算各個數(shù)據(jù)流的絕對偏差值，同時累加計算所有數(shù)據(jù)流的總絕對偏差值ΔV和平均偏差值Δ。算法步驟2～3根據(jù)流分類模型中的式（13）～（16），使用步驟1計算所得值，對數(shù)據(jù)流集合遍歷的同時進行數(shù)據(jù)流的分類并返回分類所得不同數(shù)據(jù)流集合。由此可知，該算法的時間復(fù)雜度為數(shù)據(jù)流集合S的大小，即O（S）。

3.2?流簇分配算法

由定義4可知，DSC-Heron將高頻數(shù)據(jù)流關(guān)聯(lián)的高頻流簇作為調(diào)度的基本單元，盡可能得將同一高頻流簇相關(guān)的任務(wù)實例調(diào)度到同一工作節(jié)點。在調(diào)度的過程中，需要根據(jù)當前高頻數(shù)據(jù)流關(guān)聯(lián)的任務(wù)實例在高頻流集合Hf中遞歸搜索與之關(guān)聯(lián)的流簇，然后對該流簇中的任務(wù)實例進行分配，由此得到流簇分配算法。具體算法如算法2所示。

算法2?流簇分配算法。

輸入?數(shù)據(jù)流sij，kl關(guān)聯(lián)的任務(wù)實例Iij和Ikl;高頻數(shù)據(jù)流集合Hf;目標節(jié)點nk;原始任務(wù)分配計劃PPold;當前目標任務(wù)分配計劃PPnew。

輸出?更新后的目標任務(wù)分配計劃PPnew。

步驟1?根據(jù)Iij和Ikl在Hf中搜索關(guān)聯(lián)的高頻數(shù)據(jù)流sgh，ij和skl，mn，若不存在相關(guān)聯(lián)高頻數(shù)據(jù)流或存在任務(wù)實例Igh、Imn且均已分配則遞歸結(jié)束。若存在且未分配，根據(jù)PPold判斷Igh和Imn是否分別位于目標節(jié)點nk中：如果是，判斷將該任務(wù)實例仍分配至nk中是否滿足資源約束條件：若滿足則進行分配，更新PPnew并進行步驟2;若不滿足則查找當前PPnew中負載最小的工作節(jié)點進行調(diào)度。

如果否，判斷將未分配任務(wù)實例調(diào)度到nk節(jié)點之后是否滿足資源約束條件：若滿足進行調(diào)度，更新PPnew并進行步驟2;若不滿足則查找當前PPnew中負載最小的節(jié)點進行調(diào)度。

步驟2?在Hf中遞歸查找Igh和Imn關(guān)聯(lián)的高頻數(shù)據(jù)流sef，gh和smn，op，若存在任務(wù)實例Ief，Iop未分配，重復(fù)步驟1直至遞歸結(jié)束。

流簇分配算法是一個遞歸的集合搜索過程，它將一個高頻數(shù)據(jù)流關(guān)聯(lián)的源任務(wù)實例、目的任務(wù)實例和高頻數(shù)據(jù)流集合作為輸入，分別對源任務(wù)實例和目的任務(wù)實例在高頻數(shù)據(jù)流集合中搜索相關(guān)聯(lián)的高頻數(shù)據(jù)流，并依次對搜索結(jié)果中包含的未分配任務(wù)實例進行調(diào)度，以更新目標任務(wù)分配計劃。同時，該搜索過程也是構(gòu)建該數(shù)據(jù)流高頻流簇SC的過程。若一條高頻數(shù)據(jù)流在高頻數(shù)據(jù)流集合中不存在關(guān)聯(lián)的高頻流或存在且均已經(jīng)分配完成則遞歸結(jié)束，返回對該條數(shù)據(jù)流構(gòu)建完成的目標任務(wù)分配計劃。流簇分配算法將每個高頻數(shù)據(jù)流的關(guān)聯(lián)的流簇集合整體作為任務(wù)調(diào)度對象，旨在最大化節(jié)點內(nèi)任務(wù)間數(shù)據(jù)流，根據(jù)最優(yōu)通信開銷模型即等價于最小化節(jié)點間數(shù)據(jù)流，以減少拓撲整體通信開銷。根據(jù)該遞歸算法可知，其時間復(fù)雜度為：O（Hf·SC）。其中，由于SC的最大值等于拓撲關(guān)鍵路徑長度，因此該算法的運行時間與拓撲的層數(shù)以及高頻數(shù)據(jù)流集合規(guī)模有關(guān)。

3.3?流分類任務(wù)調(diào)度算法

流分類調(diào)度算法整合了流分類算法和流簇分配算法，將流分類算法中得到的高頻流集合作為輸入，通過遍歷數(shù)據(jù)流集合S，首先對其中高頻數(shù)據(jù)流進行分配并對該數(shù)據(jù)流調(diào)用流簇分配算法進行任務(wù)調(diào)度，直至所有的高頻數(shù)據(jù)流分配完成，然后對未分配的中頻數(shù)據(jù)流和低頻數(shù)據(jù)流分別調(diào)度，完成構(gòu)建目標任務(wù)分配計劃。具體算法如算法3所示。

算法3?流分類調(diào)度算法。

輸出?目標任務(wù)分配計劃PPnew。

初始化?由負載監(jiān)測模塊獲取當前各數(shù)據(jù)流大小vij，kl和各任務(wù)實例CPU負載wIij以初始化數(shù)據(jù)流集合S與任務(wù)實例負載集合W;使用流分類算法得到各數(shù)據(jù)流分類集合Hf、Mf、Nf;初始化PPnew為空。

步驟1?根據(jù)數(shù)據(jù)流vij，kl的大小對集合S進行降序排序，遍歷集合S中各數(shù)據(jù)流sij，kl。

步驟2?如果sij，kl屬于高頻數(shù)據(jù)流集合Hf，進行如下步驟：①判斷數(shù)據(jù)流sij，kl關(guān)聯(lián)的兩個任務(wù)實例Iij和Ikl是否在PPnew中已經(jīng)重新分配，若都已經(jīng)重新分配則對該數(shù)據(jù)流的調(diào)度結(jié)束。

②若僅其中一個任務(wù)已分配，這里以Iij已分配至工作節(jié)點nk且Ikl未分配為例，進行如下步驟：（a）根據(jù)PPold判斷Iij和Ikl是否位于同一節(jié)點。若在同一節(jié)點nk，更新PPnew并調(diào)用流簇分配算法，分配sij，kl關(guān)聯(lián)的其他高頻數(shù)據(jù)流，調(diào)度結(jié)束。

（b）若Iij和Ikl位于不同節(jié)點，根據(jù)資源約束模型判斷Ikl調(diào)度到工作nk后是否滿足資源約束，若滿足則將Ikl調(diào)度到工作節(jié)點nk，更新PPnew并調(diào)用流簇分配算法，分配sij，kl關(guān)聯(lián)的高頻數(shù)據(jù)流，調(diào)度結(jié)束。若不滿足，則根據(jù)W查找當前PPnew中負載最小節(jié)點分配任務(wù)Ikl。

③若任務(wù)實例Iij和Ikl均未分配，進行如下步驟：（a）根據(jù)PPold判斷Iij和Ikl是否位于同一節(jié)點，若在同一節(jié)點nk，更新PPnew并調(diào)用流簇分配算法，分配sij，kl關(guān)聯(lián)的高頻數(shù)據(jù)流，調(diào)度結(jié)束。

（b）若Iij和Ikl位于不同節(jié)點ni和nk中，計算當前分配計劃PPnew中ni和nk已分配任務(wù)的負載，將Iij和Ikl分配到負載較小的節(jié)點中，并根據(jù)資源約束模型判斷調(diào)度過程中是否滿足資源約束，若滿足則逐個調(diào)度任務(wù)，更新PPnew并調(diào)用流簇分配算法，分配sij，kl關(guān)聯(lián)的高頻數(shù)據(jù)流，調(diào)度結(jié)束。若在調(diào)度任務(wù)實例Ikl時已不滿足資源約束條件，則將其調(diào)度至當前負載最小的節(jié)點。

步驟3?如果sij，kl屬于中頻數(shù)據(jù)流集合Mf，則重復(fù)步驟2中的①～③，但不再調(diào)用流簇分配算法，此時高頻數(shù)據(jù)流已經(jīng)調(diào)度完成。

步驟4?結(jié)束數(shù)據(jù)流集合S的遍歷，對剩下低頻數(shù)據(jù)流集合Nf中sij，kl計算當前PPnew中各個節(jié)點的任務(wù)負載情況，優(yōu)先將任務(wù)Iij、Ikl調(diào)度到負載較輕的節(jié)點中并保證滿足資源約束條件，直到全部任務(wù)調(diào)度完成。

算法步驟1中根據(jù)集合中數(shù)據(jù)流的大小對數(shù)據(jù)流集合S進行降序排序，這樣可以保證在對集合S進行遍歷時，對高頻數(shù)據(jù)流進行優(yōu)先處理。

步驟2對集合S進行遍歷，對屬于高頻數(shù)據(jù)流集合的當前數(shù)據(jù)流進行調(diào)度。該步驟中包含的①～③分別是對當前高頻數(shù)據(jù)流關(guān)聯(lián)的兩個任務(wù)實例是否在目標任務(wù)分配計劃中分配完成進行判斷，目的是為了避免對可能出現(xiàn)在不同高頻數(shù)據(jù)流中的同一任務(wù)實例進行重復(fù)調(diào)度。步驟①中，如果當前的兩個任務(wù)實例均已經(jīng)在目標任務(wù)分配計劃中分配完成，則不再進行重復(fù)調(diào)度。步驟②中，當前兩個任務(wù)實例中僅有一個任務(wù)實例已經(jīng)分配，則優(yōu)先將未分配的任務(wù)實例調(diào)度到已分配任務(wù)實例當前所在的工作節(jié)點中，此舉是為了最大化節(jié)點內(nèi)任務(wù)實例間的直接通信以最小化節(jié)點間的通信開銷。但在這種調(diào)度的過程中，需要判斷調(diào)度發(fā)生后目標節(jié)點是否滿足資源約束條件：若滿足則可以進行調(diào)度，完成最大化節(jié)點內(nèi)通信開銷;若不滿足，則需要使用當前目標任務(wù)分配計劃PPnew和任務(wù)負載集合W尋找當前集群中負載最小的工作節(jié)點分配該任務(wù)，目的是為了在未能滿足最小化通信開銷的情況下，盡量平衡集群各工作節(jié)點負載從而在負載均衡的角度優(yōu)化集群性能。步驟③中為當前兩個任務(wù)實例均未重新分配，需要依次對兩個未分配任務(wù)實例進行調(diào)度。首先根據(jù)原始任務(wù)分配計劃PPold判斷兩個任務(wù)實例是否位于同一節(jié)點，并優(yōu)先將兩個任務(wù)調(diào)度到其中當前負載較輕的工作節(jié)點中，以最小化通信開銷的同時均衡集群負載。但在對兩個任務(wù)進行依次調(diào)度時，均需要對目標工作節(jié)點計算調(diào)度完成后的資源約束情況，若不滿足資源約束，則同步驟②中的方法相同，將該任務(wù)實例分配至當前集群中負載最小的工作節(jié)點。

通過步驟1～2，算法將數(shù)據(jù)流集合S中包含的高頻數(shù)據(jù)流及其關(guān)聯(lián)高頻流簇調(diào)度完成，因此步驟3對中頻數(shù)據(jù)流集合中的數(shù)據(jù)流進行調(diào)度時，不需要考慮高頻流簇的分配，僅對中頻數(shù)據(jù)流本身繼續(xù)調(diào)度。

算法步驟4進行之前，前序算法步驟已經(jīng)對數(shù)據(jù)流集合S中高頻數(shù)據(jù)流和中頻數(shù)據(jù)流調(diào)度結(jié)束，此時只剩下低頻數(shù)據(jù)流，而低頻數(shù)據(jù)流在拓撲中對整體通信開銷的影響最小，因此僅對其進行負載均衡處理，依次將低頻數(shù)據(jù)流調(diào)度至集群當前負載較小的工作節(jié)點中，最終完成全部數(shù)據(jù)流的調(diào)度。根據(jù)該算法步驟可知，該算法在對數(shù)據(jù)流集合S遍歷的同時分別將屬于各個類別的數(shù)據(jù)流調(diào)度至符合條件的工作節(jié)點中，因此算法時間復(fù)雜度為O（S·N）（其中N為工作節(jié)點數(shù)量）。

3.4?算法部署與實現(xiàn)

Heron為編程人員提供了可擴展的Custom Scheduler[23]實現(xiàn)。為實現(xiàn)自定義調(diào)度器，需要實現(xiàn)與Heron調(diào)度器相關(guān)的IPacking、ILauncher、IScheduler和IUploader四個Java接口。在本文實驗中，DSC-Heron基于Heron中默認AuroraScheduler進行部署和實現(xiàn)，Uploader仍使用HDFS不作修改，但分別實現(xiàn)了以下三個接口：1）DSCPacking。實現(xiàn)IPacking接口，用于部署DSC-Heron以構(gòu)建目標任務(wù)分配計劃，為調(diào)度控制模塊對拓撲任務(wù)的重調(diào)度提供依據(jù)。

2）DSCLauncher。實現(xiàn)ILauncher接口，替換原有的AuroraLauncher。用于在拓撲提交后創(chuàng)建DSCScheduler對象實例并調(diào)用其onSchedule方法啟動自定義調(diào)度器。

3）DSCScheduler。實現(xiàn)IScheduler接口，替換默認AuroraScheduler，拓撲提交后將由該調(diào)度器完成拓撲的初次調(diào)度。其中部署調(diào)度觸發(fā)模塊和調(diào)度控制模塊，用于重調(diào)度的觸發(fā)以及使用DSCPacking創(chuàng)建的目標任務(wù)分配計劃更新拓撲，完成重調(diào)度過程。

在拓撲重調(diào)度的過程中，需要實時獲取各工作節(jié)點以及工作節(jié)點內(nèi)各任務(wù)實例的CPU負載。對于各任務(wù)進程的CPU資源占用信息，可以通過Java API中ThreadMXBean類的getThreadCpuTime（long id）方法獲取，其中id為各Java進程中運行任務(wù)實例的線程ID。對于各工作節(jié)點的CPU負載，可以通過對運行在該工作節(jié)點中的任務(wù)實例CPU負載進行累加求得。此外，工作節(jié)點中相關(guān)硬件參數(shù)可通過/proc目錄下的相關(guān)文件獲得。在代碼編寫完成后，使用Maven創(chuàng)建自定義調(diào)度器的jar文件，將其放置到${HERON_HOME}/lib/scheduler目錄下，并在${HERON_HOME}/conf/aurora目錄下的scheduler.yaml文件中進行配置DSCScheduler和DSCLauncher類名后即可使用。

改進后的Heron系統(tǒng)結(jié)構(gòu)如圖4所示。其中，在Heron系統(tǒng)結(jié)構(gòu)中新增的四個自定義模塊分別是：

1）負載監(jiān)測模塊。部署在各個工作節(jié)點中，負責在一定時間窗口內(nèi)監(jiān)測工作節(jié)點中運行的任務(wù)實例CPU負載、任務(wù)間數(shù)據(jù)流大小和內(nèi)存資源占用等信息，并將監(jiān)測信息實時寫入數(shù)據(jù)存儲模塊。使用該模塊，需要在拓撲中各Spout的open（）和nextTuple（）方法以及各Bolt的prepare（）和execute（）方法中調(diào)用該模塊。

2）調(diào)度觸發(fā)模塊。部署在DSCScheduler中并在該調(diào)度器對象實例化時啟動。負責在滿足重調(diào)度觸發(fā)條件時調(diào)用調(diào)度控制模塊中重調(diào)度方法完成DSC-Heron的重調(diào)度過程。

3）數(shù)據(jù)存儲模塊。存儲并實時更新負載監(jiān)測模塊獲取的任務(wù)實例監(jiān)測信息，這里使用MySQL數(shù)據(jù)庫實現(xiàn)。

4）調(diào)度控制模塊。完成流分類調(diào)度策略的核心模塊，通過調(diào)度觸發(fā)模塊調(diào)用。根據(jù)原始任務(wù)分配計劃獲取由DSC-Heron構(gòu)建的目標任務(wù)分配計劃，更新拓撲任務(wù)分配以完成任務(wù)調(diào)度過程。該模塊采用與DSCScheduler松耦合的設(shè)計范式，便于未來部署其他在線重調(diào)度算法。

4?實驗

4.1?實驗環(huán)境

實驗環(huán)境采用硬件配置相同的PC搭建一個9節(jié)點的Heron集群。其中一個主控節(jié)點運行Heron、Heron Tracker、Heron UI、Mesos Master和Aurora Scheduler;一個協(xié)調(diào)節(jié)點運行ZooKeeper和MySQL等服務(wù);一個節(jié)點運行Heron自動創(chuàng)建的拓撲管理進程（Topology Master）用于管理拓撲整個生命周期;其余節(jié)點為工作節(jié)點分別運行Mesos Agent、Aurora Observer和Aurora Executor，負責實際運行拓撲的任務(wù)實例。此外，集群中各節(jié)點共同運行HDFS作為Heron的Uploader系統(tǒng)組件并負責共享Heron Binaries文件。實驗集群的軟硬件配置如表1所示。

實驗采用Heron Github開源項目[24]提供的Sentence WordCount和WordCount示例拓撲以及自定義FileWordCount拓撲，三種拓撲中采用不同的結(jié)構(gòu)和數(shù)據(jù)源以評估DSC-Heron在不同場景中的表現(xiàn)。其中SentenceWordCount拓撲包含三層結(jié)構(gòu)：

第一層Spout組件隨機創(chuàng)建一個長度為128×1024的句子數(shù)組并隨機發(fā)射;

第二層Bolt組件（名為Split）通過空格字符分割句子產(chǎn)生單詞;

第三層Bolt組件（名為Count）接受Split中發(fā)送的單詞并進行計數(shù)。

WordCount為兩層結(jié)構(gòu)，Spout組件（名為word）隨機生成單詞并發(fā)射，由Bolt（名為consumer）組件進行統(tǒng)計。FileWordCount拓撲結(jié)構(gòu)與SentenceWordCount相同，但數(shù)據(jù)源來自原版英文歷史小說《雙城記》，格式為txt。SentenceWordCount拓撲相對于WordCount的兩層結(jié)構(gòu)，包含的數(shù)據(jù)流數(shù)量和流簇規(guī)模較大，兩者對比有利于評估系統(tǒng)性能的優(yōu)化效果。

FileWordCount拓撲數(shù)據(jù)源采用真實文本文檔，其中各單詞出現(xiàn)的頻率不盡相同，在實際的應(yīng)用場景中有一定代表性。

實驗拓撲中設(shè)置了各組件的并行度和資源需求，數(shù)據(jù)流在各組件間的傳遞模式，可用的容器數(shù)量以及資源需求，詳細的測試拓撲運行參數(shù)配置如表2所示。

表2中：topology.max.spout.pending（簡稱為pending）的值為Spout緩存隊列的最大容量，當隊列長度達到設(shè)置的容量時Spout停止發(fā)送數(shù)據(jù)，當隊列長度小于設(shè)定值時Spout持續(xù)發(fā)送數(shù)據(jù)，從而實現(xiàn)對拓撲數(shù)據(jù)傳輸速率的控制。topology.message.timeout.secs（簡稱為timeout）的值配合ATLEAST_ONCE可靠性語義模式和Acknowledgement[25]機制使用。實驗拓撲在Bolt組件中設(shè)置ack機制，唯一標識標記的元組從Spout中發(fā)射，在timeout設(shè)定的時間內(nèi)經(jīng)過各個組件處理完成后由Spout中ack方法進行確認，若沒有在該參數(shù)規(guī)定的時間內(nèi)接受到指定元組，Heron則會重新發(fā)送以保證ATLESAT_ONCE的有效進行。

對于參數(shù)pending和timeout的取值，在默認調(diào)度策略下使用SentenceWordCount拓撲經(jīng)過多次實驗得到表3所示參數(shù)取值對拓撲元組失敗率的影響，在表2所示參數(shù)下其他測試拓撲實驗結(jié)果與此類似。其中， pending的值設(shè)置為100且timeout值設(shè)置為60s時，雖然拓撲運行前5min的失敗率較低，但由于pending的值較小無法正常發(fā)揮集群運行性能且無法體現(xiàn)真實應(yīng)用場景;當pending的值設(shè)置為1000且timeout的值設(shè)置為60s時，拓撲提交后前5min內(nèi)的元組失敗率為6.6%，相對于相同pending值但timeout值為30s時拓撲運行的20%失敗率，元組失敗率明顯降低并且有少量拓撲出現(xiàn)重新發(fā)送的情況，該場景較符合真實應(yīng)用場景且集群能夠快速地趨于穩(wěn)定。而當pending的值設(shè)置為10000且timeout的值為60s時，元組失敗率較高，此時雖然提高timeout參數(shù)值可以降低元組失敗率，但會導致CPU負載過高從而使集群運行情況不可預(yù)測。因此， pending的值設(shè)置為1000，timeout的值設(shè)置為60s，在當前集群的配置下能夠較好地滿足實驗的需要。

此外，由于集群中主控節(jié)點獨立運行，拓撲管理器單獨運行于一個工作節(jié)點的容器中，ZooKeeper和MySQL等服務(wù)進程占用一個節(jié)點資源，因此集群中可分配任務(wù)實例的工作節(jié)點數(shù)量為6。在表2中將topology.stmgrs的數(shù)量設(shè)置為6，即容器數(shù)量與集群中工作節(jié)點的數(shù)量相同，意味著每個工作節(jié)點中僅運行一個容器，從而消除容器間通信帶來的開銷，重點關(guān)注節(jié)點間通信和節(jié)點內(nèi)實例間通信的轉(zhuǎn)換對集群性能的影響。

為驗證DSC-Heron的有效性，本文與Heron默認的輪詢調(diào)度策略進行了對比，表4中列出了DSC-Heron的參數(shù)設(shè)置，其中主控節(jié)點根據(jù)reschedule.timeout參數(shù)觸發(fā)重調(diào)度。工作節(jié)點中α、 β和γ為資源約束模型中設(shè)置的資源閾值參數(shù)，為避免節(jié)點滿負荷運行影響集群性能，α值設(shè)置為0.7，由于Heron容器中已為系統(tǒng)級進程留有內(nèi)存資源，因此β和γ的值設(shè)置為1。time.window.length和time.window.count為負載監(jiān)測模塊中設(shè)置的數(shù)據(jù)統(tǒng)計窗口大小和數(shù)量，即使用長度為5s的時間窗口對數(shù)據(jù)采樣3次統(tǒng)計平均值。此外Heron集群的其他配置參數(shù)與默認輪詢調(diào)度算法的參數(shù)均取默認值。