編者按

伏羲（Fuxi）是十年前最初創(chuàng)立飛天平臺(tái)時(shí)的三大服務(wù)之一（分布式存儲(chǔ) Pangu，分布式計(jì)算 MaxCompute，分布式調(diào)度 Fuxi），當(dāng)時(shí)的設(shè)計(jì)初衷是為了解決大規(guī)模分布式資源的調(diào)度問題（本質(zhì)上是多目標(biāo)的最優(yōu)匹配問題）。

隨阿里經(jīng)濟(jì)體和阿里云豐富的業(yè)務(wù)需求（尤其是雙十一）和磨練，伏羲的內(nèi)涵不斷擴(kuò)大，從單一的資源調(diào)度器（對(duì)標(biāo)開源系統(tǒng)的YARN）擴(kuò)展成大數(shù)據(jù)的核心調(diào)度服務(wù)，覆蓋數(shù)據(jù)調(diào)度（Data Placement）、資源調(diào)度（Resouce Management）、計(jì)算調(diào)度（Application Manager）、和本地微（自治）調(diào)度（即正文中的單機(jī)調(diào)度）等多個(gè)領(lǐng)域，并在每一個(gè)細(xì)分領(lǐng)域致力于打造超越業(yè)界主流的差異化能力。

過去十年來，伏羲在技術(shù)能力上每年都有一定的進(jìn)展和突破（如2013年的5K，15年的Sortbenchmark世界冠軍，17年的超大規(guī)模離在/在離混布能力，2019年的 Yugong 發(fā)布并論文被VLDB接受等等）。本文試從面向大數(shù)據(jù)/云計(jì)算的調(diào)度挑戰(zhàn)出發(fā)，介紹各個(gè)子領(lǐng)域的關(guān)鍵進(jìn)展，并回答什么是“伏羲 2.0”。
?

1. 引言

過去10年，是云計(jì)算的10年，伴隨云計(jì)算的爆炸式增長，大數(shù)據(jù)行業(yè)的工作方式也發(fā)生了很大的變化：從傳統(tǒng)的自建自運(yùn)維hadoop集群，變成更多的依賴云上的彈性低成本計(jì)算資源。海量大數(shù)據(jù)客戶的信任和托付，對(duì)阿里大數(shù)據(jù)系統(tǒng)來說，是很大的責(zé)任，但也催生出了大規(guī)模、多場景、低成本、免運(yùn)維的MaxCompute通用計(jì)算系統(tǒng)。

同樣的10年，伴隨著阿里年年雙11，MaxCompute同樣支撐了阿里內(nèi)部大數(shù)據(jù)的蓬勃發(fā)展，從原來的幾百臺(tái)，到現(xiàn)在的10萬臺(tái)物理機(jī)規(guī)模。

雙線需求，殊途同歸，海量資源池，如何自動(dòng)匹配到大量不同需求的異地客戶計(jì)算需求上，需要調(diào)度系統(tǒng)的工作。本文主要介紹阿里大數(shù)據(jù)的調(diào)度系統(tǒng)FUXI往2.0的演進(jìn)。先給大家介紹幾個(gè)概念：
?

• 首先，數(shù)據(jù)從哪里來？數(shù)據(jù)往往伴隨著在線業(yè)務(wù)系統(tǒng)產(chǎn)生。而在線系統(tǒng)，出于延遲和容災(zāi)的考慮，往往遍布北京、上海、深圳等多個(gè)地域，如果是跨國企業(yè)，還可能遍布?xì)W美等多個(gè)大陸的機(jī)房。這也造成了我們的數(shù)據(jù)天然分散的形態(tài)。而計(jì)算，也可能發(fā)生在任意一個(gè)地域和機(jī)房。可是網(wǎng)絡(luò)，是他們中間的瓶頸，跨地域的網(wǎng)絡(luò)，在延遲和帶寬上，遠(yuǎn)遠(yuǎn)無法滿足大數(shù)據(jù)計(jì)算的需求。如何平衡計(jì)算資源、數(shù)據(jù)存儲(chǔ)、跨域網(wǎng)絡(luò)這幾點(diǎn)之間的平衡，需要做好“數(shù)據(jù)調(diào)度”。
  ?
• 其次，有了數(shù)據(jù)，計(jì)算還需要CPU，內(nèi)存，甚至GPU等資源，當(dāng)不同的公司，或者單個(gè)公司內(nèi)部不同的部門，同時(shí)需要計(jì)算資源，而計(jì)算資源緊張時(shí)，如何平衡不同的用戶，不同的作業(yè)？作業(yè)也可能長短不一，重要程度不盡相同，今天和明天的需求也大相徑庭。除了用戶和作業(yè)，計(jì)算資源本身可能面臨硬件故障，但用戶不想受影響。所有這些，都需要“資源調(diào)度”。
  ?
• 有了數(shù)據(jù)和計(jì)算資源，如何完成用戶的計(jì)算任務(wù)，比如一個(gè)SQL query？這需要將一個(gè)大任務(wù)，分成幾個(gè)步驟，每個(gè)步驟又切分成成千上萬個(gè)小任務(wù)，并行同時(shí)計(jì)算，才能體現(xiàn)出分布式系統(tǒng)的加速優(yōu)勢。但小任務(wù)切粗切細(xì)，在不同的機(jī)器上有快有慢，上下步驟如何交接數(shù)據(jù)，同時(shí)避開各自故障和長尾，這些都需要“計(jì)算調(diào)度”。
  ?
• 很多不同用戶的不同小任務(wù)，經(jīng)過層層調(diào)度，最后匯集到同一臺(tái)物理機(jī)上，如何避免單機(jī)上真正運(yùn)行時(shí)，對(duì)硬件資源使用的各種不公平，避免老實(shí)人吃虧。避免重要關(guān)鍵任務(wù)受普通任務(wù)影響，這都需要內(nèi)核層面的隔離保障機(jī)制。同時(shí)還要兼顧隔離性和性能、成本的折中考慮。這都需要“單機(jī)調(diào)度”。

2013年，伏羲在飛天5K項(xiàng)目中對(duì)系統(tǒng)架構(gòu)進(jìn)行了第一次大重構(gòu)，解決了規(guī)模、性能、利用率、容錯(cuò)等線上問題，并取得世界排序大賽Sortbenchmark四項(xiàng)冠軍，這標(biāo)志著Fuxi 1.0的成熟。

2019年，伏羲再次出發(fā)，從技術(shù)上對(duì)系統(tǒng)進(jìn)行了第二次重構(gòu)，發(fā)布Fuxi 2.0版本：阿里自研的新一代高性能、分布式的數(shù)據(jù)、資源、計(jì)算、單機(jī)調(diào)度系統(tǒng)。Fuxi 2.0進(jìn)行了全面的技術(shù)升級(jí)，在全區(qū)域數(shù)據(jù)排布、去中心化調(diào)度、在線離線混合部署、動(dòng)態(tài)計(jì)算等方面全方位滿足新業(yè)務(wù)場景下的調(diào)度需求。

伏羲2.0成果概覽

? 業(yè)內(nèi)首創(chuàng)跨地域多數(shù)據(jù)中心的數(shù)據(jù)調(diào)度方案-Yugong，通過3%的冗余存儲(chǔ)，節(jié)省80%的跨地域網(wǎng)絡(luò)帶寬
? 業(yè)內(nèi)領(lǐng)先的去中心化資源調(diào)度架構(gòu)，單集群支持10萬服務(wù)器*10萬并發(fā)job的高頻調(diào)度
? 動(dòng)態(tài)DAG闖入傳統(tǒng)SQL優(yōu)化盲區(qū)，TPC-DS性能提升27%，conditional join性能提升3X。
? 創(chuàng)新性的數(shù)據(jù)動(dòng)態(tài)shuffle和全局跨級(jí)優(yōu)化，取代業(yè)界磁盤shuffle；線上千萬job，整體性能提升20%，成本下降15%，出錯(cuò)率降低一個(gè)數(shù)量級(jí)
? 在線離線規(guī)模化混合部署，在線集群利用率由10%提升到40%，雙十一大促節(jié)省4200臺(tái)F53資源，且同時(shí)保障在線離線業(yè)務(wù)穩(wěn)定。

2. 數(shù)據(jù)調(diào)度2.0 - 跨地域的數(shù)據(jù)調(diào)度

阿里巴巴在全球都建有數(shù)據(jù)中心，每個(gè)地區(qū)每天會(huì)產(chǎn)生一份當(dāng)?shù)氐慕灰子唵涡畔?#xff0c;存在就近的數(shù)據(jù)中心。北京的數(shù)據(jù)中心，每天會(huì)運(yùn)行一個(gè)定時(shí)任務(wù)來統(tǒng)計(jì)當(dāng)天全球所有的訂單信息，需要從其他數(shù)據(jù)中心讀取這些交易數(shù)據(jù)。當(dāng)數(shù)據(jù)的產(chǎn)生和消費(fèi)不在一個(gè)數(shù)據(jù)中心時(shí)，我們稱之為跨數(shù)據(jù)中心數(shù)據(jù)依賴（下文簡稱跨中心依賴）。

圖. 阿里巴巴全球數(shù)據(jù)中心

MaxCompute上每天運(yùn)行著數(shù)以千萬計(jì)的作業(yè)，處理EB級(jí)別的數(shù)據(jù)。這些計(jì)算和數(shù)據(jù)分布在全球的數(shù)據(jù)中心，復(fù)雜的業(yè)務(wù)依賴關(guān)系產(chǎn)生了大量的跨中心依賴。相比于數(shù)據(jù)中心內(nèi)的網(wǎng)絡(luò)，跨數(shù)據(jù)中心網(wǎng)絡(luò)（尤其是跨域的網(wǎng)絡(luò)）是非常昂貴的，同時(shí)具有帶寬小、延遲高、穩(wěn)定性低的特點(diǎn)。比如網(wǎng)絡(luò)延遲，數(shù)據(jù)中心內(nèi)部網(wǎng)絡(luò)的網(wǎng)絡(luò)延遲一般在100微秒以下，而跨地域的網(wǎng)絡(luò)延遲則高達(dá)數(shù)十毫秒，相差百倍以上。因此，如何高效地將跨中心依賴轉(zhuǎn)化為數(shù)據(jù)中心內(nèi)部的數(shù)據(jù)依賴，減少跨數(shù)據(jù)中心網(wǎng)絡(luò)帶寬消耗，從而降低成本、提高系統(tǒng)效率，對(duì)MaxCompute這樣超大規(guī)模計(jì)算平臺(tái)而言，具有極其重要的意義。

圖. MaxCompute平臺(tái)數(shù)據(jù)及依賴增長趨勢

為了解決這個(gè)問題，我們?cè)跀?shù)據(jù)中心上增加了一層調(diào)度層，用于在數(shù)據(jù)中心之間調(diào)度數(shù)據(jù)和計(jì)算。這層調(diào)度獨(dú)立于數(shù)據(jù)中心內(nèi)部的調(diào)度，目的是實(shí)現(xiàn)跨地域維度上存儲(chǔ)冗余--計(jì)算均衡--長傳帶寬--性能最優(yōu)之間的最佳平衡。這層調(diào)度層包括跨數(shù)據(jù)中心數(shù)據(jù)緩存、業(yè)務(wù)整體排布、作業(yè)粒度調(diào)度。

首先是對(duì)訪問頻次高的數(shù)據(jù)進(jìn)行跨數(shù)據(jù)中心緩存，在緩存空間有限的約束下，選擇合適的數(shù)據(jù)進(jìn)行換入換出。不同于其他緩存系統(tǒng)，MaxCompute的數(shù)據(jù)（分區(qū)）以表的形式組織在一起，每張表每天產(chǎn)生一個(gè)或多個(gè)分區(qū)，作業(yè)訪問數(shù)據(jù)也有一些特殊規(guī)律，比如一般訪問的是連續(xù)分區(qū)、生成時(shí)間越新的分區(qū)訪問概率越大。

其次是業(yè)務(wù)的整體排布策略。數(shù)據(jù)和計(jì)算以業(yè)務(wù)為單位組織在一起（MaxCompute中稱之為project），每個(gè)project被分配在一個(gè)數(shù)據(jù)中心，包括數(shù)據(jù)存儲(chǔ)和計(jì)算作業(yè)。如果將project看做一個(gè)整體，可以根據(jù)作業(yè)對(duì)數(shù)據(jù)的依賴關(guān)系計(jì)算出project之間的相互依賴關(guān)系。如果能將有互相數(shù)據(jù)依賴的project放在一個(gè)數(shù)據(jù)中心，就可以減少跨中心依賴。但project間的依賴往往復(fù)雜且不斷變化，很難有一勞永逸的排布策略，并且project排布需要對(duì)project進(jìn)行整體遷移，周期較長，且需要消耗大量的帶寬。

最后，當(dāng)project之間的互相依賴集中在極少數(shù)幾個(gè)作業(yè)上，并且作業(yè)的輸入數(shù)據(jù)量遠(yuǎn)大于輸出數(shù)據(jù)量時(shí)，比起數(shù)據(jù)緩存和project整體遷移，更好的辦法是將這些作業(yè)調(diào)度到數(shù)據(jù)所在的數(shù)據(jù)中心，再將作業(yè)的輸出遠(yuǎn)程寫回原數(shù)據(jù)中心，即作業(yè)粒度調(diào)度。如何在作業(yè)運(yùn)行之前就預(yù)測到作業(yè)的輸入輸出數(shù)據(jù)量和資源消耗，另一方面當(dāng)作業(yè)調(diào)度到remote數(shù)據(jù)中心后，如何保證作業(yè)運(yùn)行不會(huì)變慢，不影響用戶體驗(yàn)，這都是作業(yè)粒度調(diào)度要解決的問題。

本質(zhì)上，數(shù)據(jù)緩存、業(yè)務(wù)排布、作業(yè)粒度調(diào)度三者都在解同一個(gè)問題，即在跨地域多數(shù)據(jù)中心系統(tǒng)中減少跨中心依賴量、優(yōu)化作業(yè)的data locality、減少網(wǎng)絡(luò)帶寬消耗。

1.2.1 跨數(shù)據(jù)中心數(shù)據(jù)緩存策略

我們首次提出了跨地域、跨數(shù)據(jù)中心數(shù)據(jù)緩存這一概念，通過集群的存儲(chǔ)換集群間帶寬，在有限的冗余存儲(chǔ)下，找到存儲(chǔ)和帶寬最佳的tradeoff。通過深入的分析MaxCompute的作業(yè)、數(shù)據(jù)的特點(diǎn)，我們?cè)O(shè)計(jì)了一種高效的算法，根據(jù)作業(yè)歷史的workload、數(shù)據(jù)的大小和分布，自動(dòng)進(jìn)行緩存的換入換出。

我們研究了多種數(shù)據(jù)緩存算法，并對(duì)其進(jìn)行了對(duì)比試驗(yàn)，下圖展示了不同緩存策略的收益，橫軸是冗余存儲(chǔ)空間，縱軸是帶寬消耗。從圖中可以看出，隨著冗余存儲(chǔ)的增加，帶寬成本不斷下降，但收益比逐漸降低，我們最終采用的k-probe算法在存儲(chǔ)和帶寬間實(shí)現(xiàn)了很好的平衡。

1.2.2 以project為粒度的多集群業(yè)務(wù)排布算法

隨著上層業(yè)務(wù)的不斷發(fā)展，業(yè)務(wù)的資源需求和數(shù)據(jù)需求也在不斷變化。比如一個(gè)集群的跨中心依賴增長迅速，無法完全通過數(shù)據(jù)緩存來轉(zhuǎn)化為本地讀取，這就會(huì)造成大量的跨數(shù)據(jù)中心流量。因此我們需要定期對(duì)業(yè)務(wù)的排布進(jìn)行分析，根據(jù)業(yè)務(wù)對(duì)計(jì)算資源、數(shù)據(jù)資源的需求情況，以及集群、機(jī)房的規(guī)劃，通過業(yè)務(wù)的遷移來降低跨中心依賴以及均衡各集群壓力。

下圖展示了某個(gè)時(shí)刻業(yè)務(wù)遷移的收益分析：左圖橫軸為遷移的project數(shù)量，縱軸為帶寬減少比例，可以看出大約移動(dòng)60個(gè)project就可以減少約30%的帶寬消耗。右圖統(tǒng)計(jì)了不同排布下（遷移0個(gè)、20個(gè)、50個(gè)project）的最優(yōu)帶寬消耗，橫軸為冗余存儲(chǔ)，縱軸為帶寬。

1.2.3 跨數(shù)據(jù)中心計(jì)算調(diào)度機(jī)制

我們打破了計(jì)算資源按照數(shù)據(jù)中心進(jìn)行規(guī)劃的限制，理論上允許作業(yè)跑在任何一個(gè)數(shù)據(jù)中心。我們將調(diào)度粒度拆解到作業(yè)粒度，根據(jù)每個(gè)作業(yè)的數(shù)據(jù)需求、資源需求，為其找到一個(gè)最合適的數(shù)據(jù)中心。在對(duì)作業(yè)進(jìn)行調(diào)度之前需要知道這個(gè)作業(yè)的輸入和輸出，目前我們有兩種方式獲得這一信息，對(duì)于周期性作業(yè)，通過對(duì)作業(yè)歷史運(yùn)行數(shù)據(jù)進(jìn)行分析推測出作業(yè)的輸入輸出；對(duì)于偶發(fā)的作業(yè)，我們發(fā)現(xiàn)其產(chǎn)生較大跨域流量時(shí)，動(dòng)態(tài)的將其調(diào)度到數(shù)據(jù)所在的數(shù)據(jù)中心上運(yùn)行。另外，調(diào)度計(jì)算還要考慮作業(yè)對(duì)計(jì)算資源的需求，防止作業(yè)全部調(diào)度到熱點(diǎn)數(shù)據(jù)所在的數(shù)據(jù)中心，造成任務(wù)堆積。

1.3 線上效果

線上三種策略相輔相成，數(shù)據(jù)緩存主要解決周期類型作業(yè)、熱數(shù)據(jù)的依賴；作業(yè)粒度調(diào)度主要解決臨時(shí)作業(yè)、歷史數(shù)據(jù)的依賴；并周期性地通過業(yè)務(wù)整體排布進(jìn)行全局優(yōu)化，用來降低跨中心依賴。整體來看，通過三種策略的共同作用，降低了約90%的跨地域數(shù)據(jù)依賴，通過約3%的冗余存儲(chǔ)節(jié)省了超過80%的跨數(shù)據(jù)中心帶寬消耗，將跨中心依賴轉(zhuǎn)化為本地讀取的比例提高至90%。下圖以機(jī)房為單位展示了帶寬的收益：

3. 資源調(diào)度2.0 - 去中心化的多調(diào)度器架構(gòu)

2019年雙十一，MaxCompute平臺(tái)產(chǎn)生的數(shù)據(jù)量已接近EB級(jí)別，作業(yè)規(guī)模達(dá)到了千萬，有幾十億的worker跑在幾百萬核的計(jì)算單元上，在超大規(guī)模（單集群超過萬臺(tái)），高并發(fā)的場景下，如何快速地給不同的計(jì)算任務(wù)分配資源，實(shí)現(xiàn)資源的高速流轉(zhuǎn)，需要一個(gè)聰明的“大腦”，而這就是集群的資源管理與調(diào)度系統(tǒng)（簡稱資源調(diào)度系統(tǒng)）。

資源調(diào)度系統(tǒng)負(fù)責(zé)連接成千上萬的計(jì)算節(jié)點(diǎn)，將數(shù)據(jù)中心海量的異構(gòu)資源抽象，并提供給上層的分布式應(yīng)用，像使用一臺(tái)電腦一樣使用集群資源，它的核心能力包括規(guī)模、性能、穩(wěn)定性、調(diào)度效果、多租戶間的公平性等等。一個(gè)成熟的資源調(diào)度系統(tǒng)需要在以下五個(gè)方面進(jìn)行權(quán)衡，做到“既要又要”，非常具有挑戰(zhàn)性。

13年的5K項(xiàng)目初步證明了伏羲規(guī)模化能力，此后資源調(diào)度系統(tǒng)不斷演進(jìn)，并通過MaxCompute平臺(tái)支撐了阿里集團(tuán)的大數(shù)據(jù)計(jì)算資源需求，在核心調(diào)度指標(biāo)上保持著對(duì)開源系統(tǒng)的領(lǐng)先性，比如1）萬臺(tái)規(guī)模集群，調(diào)度延時(shí)控制在了10微秒級(jí)別，worker啟動(dòng)延時(shí)控制在30毫秒；2）支持任意多級(jí)租戶的資源動(dòng)態(tài)調(diào)節(jié)能力（支持十萬級(jí)別的租戶）；3）極致穩(wěn)定，調(diào)度服務(wù)全年99.99%的可靠性，并做到服務(wù)秒級(jí)故障恢復(fù)。

2.1 單調(diào)度器的局限性

2.1.1 線上的規(guī)模與壓力

大數(shù)據(jù)計(jì)算的場景與需求正在快速增長（下圖是過去幾年MaxComputer平臺(tái)計(jì)算和數(shù)據(jù)的增長趨勢）。單集群早已突破萬臺(tái)規(guī)模，急需提供十萬臺(tái)規(guī)模的能力。

圖. MaxCompute 2015 ~ 2018線上作業(yè)情況

但規(guī)模的增長將帶來復(fù)雜度的極速上升，機(jī)器規(guī)模擴(kuò)大一倍，資源請(qǐng)求并發(fā)度也會(huì)翻一番。在保持既有性能、穩(wěn)定性、調(diào)度效果等核心能力不下降的前提下，可以通過對(duì)調(diào)度器持續(xù)性能優(yōu)化來擴(kuò)展集群規(guī)模（這也是伏羲資源調(diào)度1.0方向），但受限于單機(jī)的物理限制，這種優(yōu)化總會(huì)存在天花板，因此需要從架構(gòu)上優(yōu)化來徹底規(guī)模和性能的可擴(kuò)展性問題。

2.1.2 調(diào)度需求的多樣性

伏羲支持了各種各樣的大數(shù)據(jù)計(jì)算引擎，除了離線計(jì)算（SQL、MR)，還包括實(shí)時(shí)計(jì)算、圖計(jì)算，以及近幾年迅速發(fā)展面向人工智能領(lǐng)域的機(jī)器學(xué)習(xí)引擎。

圖. 資源調(diào)度器的架構(gòu)類型

場景的不同對(duì)資源調(diào)度的需求也不相同，比如，SQL類型的作業(yè)通常體積小、運(yùn)行時(shí)間短，對(duì)資源匹配的要求低，但對(duì)調(diào)度延時(shí)要求高，而機(jī)器學(xué)習(xí)的作業(yè)一般體積大、運(yùn)行時(shí)間長，調(diào)度結(jié)果的好壞可能對(duì)運(yùn)行時(shí)間產(chǎn)生直接影響，因此也能容忍通過較長的調(diào)度延時(shí)換取更優(yōu)的調(diào)度結(jié)果。資源調(diào)度需求這種多樣性，決定了單一調(diào)度器很難做到“面面俱到”，需要各個(gè)場景能定制各自的調(diào)度策略，并進(jìn)行獨(dú)立優(yōu)化。

2.1.3 灰度發(fā)布與工程效率

資源調(diào)度系統(tǒng)是分布式系統(tǒng)中最復(fù)雜最重要的的模塊之一，需要有嚴(yán)苛的生產(chǎn)發(fā)布流程來保證其線上穩(wěn)定運(yùn)行。單一的調(diào)度器對(duì)開發(fā)人員要求高，出問題之后影響范圍大，測試發(fā)布周期長，嚴(yán)重影響了調(diào)度策略迭代的效率，在快速改進(jìn)各種場景調(diào)度效果的過程中，這些弊端逐漸顯現(xiàn)，因此急需從架構(gòu)上改進(jìn)，讓資源調(diào)度具備線上的灰度能力，從而幅提升工程效率。

2.2 去中心化的多調(diào)度器架構(gòu)

為了解決上述規(guī)模和擴(kuò)展性問題，更好地滿足多種場景的調(diào)度需求，同時(shí)從架構(gòu)上支持灰度能力，伏羲資源調(diào)度2.0在1.0的基礎(chǔ)上對(duì)調(diào)度架構(gòu)做了大規(guī)模的重構(gòu)，引入了去中心化的多調(diào)度器架構(gòu)。

圖. 資源調(diào)度的架構(gòu)類型

我們將系統(tǒng)中最核心的資源管理和資源調(diào)度邏輯進(jìn)行了拆分解耦，使兩者同時(shí)具備了多partition的可擴(kuò)展能力（如下圖所示），其中：
? 資源調(diào)度器（Scheduler）：負(fù)責(zé)核心的機(jī)器資源和作業(yè)資源需求匹配的調(diào)度邏輯，可以橫向擴(kuò)展。
? 資源管理和仲裁服務(wù)（ResourceManagerService，簡稱RMS）：負(fù)責(zé)機(jī)器資源和狀態(tài)管理，對(duì)各個(gè)Scheduler的調(diào)度結(jié)果進(jìn)行仲裁，可以橫向擴(kuò)展。
? 調(diào)度協(xié)調(diào)服務(wù)（Coordinator）：管理資源調(diào)度系統(tǒng)的配置信息，Meta信息，以及對(duì)機(jī)器資源、Scheduler、RMS的可用性和服務(wù)角色間的可見性做仲裁。不可橫向擴(kuò)展，但有秒級(jí)多機(jī)主備切換能力。
? 調(diào)度信息收集監(jiān)控服務(wù)（FuxiEye）：統(tǒng)計(jì)集群中每臺(tái)機(jī)的運(yùn)行狀態(tài)信息，給Scheduler提供調(diào)度決策支持，可以橫向擴(kuò)展。
? 用戶接口服務(wù)（ApiServer）：為資源調(diào)度系統(tǒng)提供外部調(diào)用的總?cè)肟?#xff0c;會(huì)根據(jù)Coordinator提供的Meta信息將用戶請(qǐng)求路由到資源調(diào)度系統(tǒng)具體的某一個(gè)服務(wù)上，可以橫向擴(kuò)展。

圖. 伏羲多調(diào)度器新架構(gòu)

2.3 上線數(shù)據(jù)

以下是10w規(guī)模集群/10萬作業(yè)并發(fā)場景調(diào)度器核心指標(biāo)（5個(gè)Scheduler、5個(gè)RMS，單RMS負(fù)責(zé)2w臺(tái)機(jī)器，單Scheduler并發(fā)處理2w個(gè)作業(yè)）。通過數(shù)據(jù)可以看到，集群10w臺(tái)機(jī)器的調(diào)度利用率超過了99%，關(guān)鍵調(diào)度指標(biāo)，單Scheduler向RMS commit的slot的平均數(shù)目達(dá)到了1w slot/s。

在保持原有單調(diào)度器各項(xiàng)核心指標(biāo)穩(wěn)定不變的基礎(chǔ)上，去中心化的多調(diào)度器框架實(shí)現(xiàn)了機(jī)器規(guī)模和應(yīng)用并發(fā)度的雙向擴(kuò)展，徹底解決了集群的可擴(kuò)展性問題。

目前資源調(diào)度的新架構(gòu)已全面上線，各項(xiàng)指標(biāo)持續(xù)穩(wěn)定。在多調(diào)度器架構(gòu)基礎(chǔ)上，我們把機(jī)器學(xué)習(xí)場景調(diào)度策略進(jìn)行了分離，通過獨(dú)立的調(diào)度器來進(jìn)行持續(xù)的優(yōu)化。同時(shí)通過測試專用的調(diào)度器，我們也讓資源調(diào)度具備了灰度能力，調(diào)度策略的開發(fā)和上線周期顯著縮短。

4. 計(jì)算調(diào)度2.0 - 從靜態(tài)到動(dòng)態(tài)

分布式作業(yè)的執(zhí)行與單機(jī)作業(yè)的最大區(qū)別，在于數(shù)據(jù)的處理需要拆分到不同的計(jì)算節(jié)點(diǎn)上，“分而治之”的執(zhí)行。這個(gè)“分”,包括數(shù)據(jù)的切分，聚合以及對(duì)應(yīng)的不同邏輯運(yùn)行階段的區(qū)分，也包括在邏輯運(yùn)行階段間數(shù)據(jù)的shuffle傳輸。每個(gè)分布式作業(yè)的中心管理點(diǎn)，也就是application master (AM)。這個(gè)管理節(jié)點(diǎn)也經(jīng)常被稱為DAG (Directional Acyclic Graph， 有向無環(huán)圖) 組件，是因?yàn)槠渥钪匾呢?zé)任，就是負(fù)責(zé)協(xié)調(diào)分布式系統(tǒng)中的作業(yè)執(zhí)行流程，包括計(jì)算節(jié)點(diǎn)的調(diào)度以及數(shù)據(jù)流(shuffle)。

對(duì)于作業(yè)的邏輯階段和各個(gè)計(jì)算節(jié)點(diǎn)的管理, 以及shuffle策略的選擇/執(zhí)行，是一個(gè)分布式作業(yè)能夠正確完成重要前提。這一特點(diǎn)，無論是傳統(tǒng)的MR作業(yè)，分布式SQL作業(yè)，還是分布式的機(jī)器學(xué)習(xí)/深度學(xué)習(xí)作業(yè)，都是一脈相承的，為了幫助更好的理解計(jì)算調(diào)度(DAG和Shuffle)在大數(shù)據(jù)平臺(tái)中的位置，我們可以通過MaxCompute分布式SQL的執(zhí)行過程做為例子來了解：

在這么一個(gè)簡單的例子中，用戶有一張訂單表order_data，存儲(chǔ)了海量的交易信息，用戶想所有查詢花費(fèi)超過1000的交易訂單按照userid聚合后，每個(gè)用戶的花費(fèi)之和是多少。于是提交了如下SQL query:

INSERT OVERWRITE TABLE result SELECT userid, SUM(spend) FROM order_data WHERE spend > 1000 GROUP BY userid;

這個(gè)SQL經(jīng)過編譯優(yōu)化之后生成了優(yōu)化執(zhí)行計(jì)劃，提交到fuxi管理的分布式集群中執(zhí)行。我們可以看到，這個(gè)簡單的SQL經(jīng)過編譯優(yōu)化，被轉(zhuǎn)換成一個(gè)具有M->R兩個(gè)邏輯節(jié)點(diǎn)的DAG圖，也就是傳統(tǒng)上經(jīng)典的MR類型作業(yè)。而這個(gè)圖在提交給fuxi系統(tǒng)后，根據(jù)每個(gè)邏輯節(jié)點(diǎn)需要的并發(fā)度，數(shù)據(jù)傳輸邊上的shuffle方式，調(diào)度時(shí)間等等信息，就被物化成右邊的物理執(zhí)行圖。物理圖上的每個(gè)節(jié)點(diǎn)都代表了一個(gè)具體的執(zhí)行實(shí)例，實(shí)例中包含了具體處理數(shù)據(jù)的算子，特別的作為一個(gè)典型的分布式作業(yè)，其中包含了數(shù)據(jù)交換的算子shuffle——負(fù)責(zé)依賴外部存儲(chǔ)和網(wǎng)絡(luò)交換節(jié)點(diǎn)間的數(shù)據(jù)。一個(gè)完整的計(jì)算調(diào)度，包含了上圖中的DAG的調(diào)度執(zhí)行以及數(shù)據(jù)shuffle的過程。

阿里計(jì)算平臺(tái)的fuxi計(jì)算調(diào)度，經(jīng)過十年的發(fā)展和不斷迭代，成為了作為阿里集團(tuán)內(nèi)部以及阿里云上大數(shù)據(jù)計(jì)算的重要基礎(chǔ)設(shè)施。今天計(jì)算調(diào)度同時(shí)服務(wù)了以MaxCompute SQL和PAI為代表的多種計(jì)算引擎，在近10萬臺(tái)機(jī)器上日均運(yùn)行著千萬界別的分布式DAG作業(yè)，每天處理EB數(shù)量級(jí)的數(shù)據(jù)。一方面隨著業(yè)務(wù)規(guī)模和需要處理的數(shù)據(jù)量的爆發(fā)，這個(gè)系統(tǒng)需要服務(wù)的分布式作業(yè)規(guī)模也在不斷增長；另一方面，業(yè)務(wù)邏輯以及數(shù)據(jù)來源的多樣性，計(jì)算調(diào)度在阿里已經(jīng)很早就跨越了不同規(guī)模上的可用/夠用的前中期階段，2.0上我們開始探索更加前沿的智能化執(zhí)行階段。

在云上和阿里集團(tuán)的大數(shù)據(jù)實(shí)踐中，我們發(fā)現(xiàn)對(duì)于計(jì)算調(diào)度需要同時(shí)具備超大規(guī)模和智能化的需求，以此為基本訴求我們開了Fuxi計(jì)算調(diào)度2.0的研發(fā)。下面就為大家從DAG調(diào)度和數(shù)據(jù)shuffle兩個(gè)方面分別介紹計(jì)算調(diào)度2.0的工作。

4.1 Fuxi DAG 2.0--動(dòng)態(tài)、靈活的分布式計(jì)算生態(tài)

4.1.1 DAG調(diào)度的挑戰(zhàn)

傳統(tǒng)的分布式作業(yè)DAG，一般是在作業(yè)提交前靜態(tài)指定的，這種指定方式，使得作業(yè)的運(yùn)行沒有太多動(dòng)態(tài)調(diào)整的空間。放在DAG的邏輯圖與物理圖的背景中來說，這要求分布式系統(tǒng)在運(yùn)行作業(yè)前，必須事先了解作業(yè)邏輯和處理數(shù)據(jù)各種特性，并能夠準(zhǔn)確回答作業(yè)運(yùn)行過程，各個(gè)節(jié)點(diǎn)和連接邊的物理特性問題，然而在現(xiàn)實(shí)情況中，許多和運(yùn)行過程中數(shù)據(jù)特性相關(guān)的問題，都只有個(gè)在執(zhí)行過程中才能被最準(zhǔn)確的獲得。靜態(tài)的DAG執(zhí)行，可能導(dǎo)致選中的是非最優(yōu)的執(zhí)行計(jì)劃，從而導(dǎo)致各種運(yùn)行時(shí)的效率低下，甚至作業(yè)失敗。這里我們可以用一個(gè)分布式SQL中很常見的例子來說明：

SELECT a.spend, a.userid, b.age FROM (SELECT spend, useridFROM order_dataWHERE spend > 1000) a JOIN (SELECT userid, ageFROM userWHERE age > 60) b ON a.userid = b.userid;

上面是一個(gè)簡單的join的例子，目的是獲取60歲以上用戶花費(fèi)大于1000的詳細(xì)信息，由于年紀(jì)和花費(fèi)在兩張表中，所以此時(shí)需要做一次join。一般來說join有兩種實(shí)現(xiàn)方式：

一是Sorted Merge Join(如下圖左側(cè)的所示)：也就是對(duì)于a和b兩個(gè)子句執(zhí)行后的數(shù)據(jù)按照join key(userid)進(jìn)行分區(qū)，然后在下游節(jié)點(diǎn)按照相同的key進(jìn)行Merge Join操作，實(shí)現(xiàn)Merge Join需要對(duì)兩張表都要做shuffle操作——也就是進(jìn)行一次數(shù)據(jù)狡猾，特別的如果有數(shù)據(jù)傾斜(例如某個(gè)userid對(duì)應(yīng)的交易記錄特別多)，這時(shí)候MergeJoin過程就會(huì)出現(xiàn)長尾，影響執(zhí)行效率；

二是實(shí)現(xiàn)方式是Map join(Hash join)的方式(如下圖右側(cè)所示)：上述sql中如果60歲以上的用戶信息較少，數(shù)據(jù)可以放到一個(gè)計(jì)算節(jié)點(diǎn)的內(nèi)存中，那對(duì)于這個(gè)超小表可以不做shuffle，而是直接將其全量數(shù)據(jù)broadcast到每個(gè)處理大表的分布式計(jì)算節(jié)點(diǎn)上，大表不用進(jìn)行shuffle操作，通過在內(nèi)存中直接建立hash表，完成join操作，由此可見map join優(yōu)化能大量減少 (大表) shuffle同時(shí)避免數(shù)據(jù)傾斜，能夠提升作業(yè)性能。但是如果選擇了map join的優(yōu)化，執(zhí)行過程中發(fā)現(xiàn)小表數(shù)據(jù)量超過了內(nèi)存限制(大于60歲的用戶很多)，這個(gè)時(shí)候query執(zhí)行就會(huì)由于oom而失敗，只能重新執(zhí)行。

但是在實(shí)際執(zhí)行過程中，具體數(shù)據(jù)量的大小，需要在上游節(jié)點(diǎn)完成后才能被感知，因此在提交作業(yè)前很難準(zhǔn)確的判斷是否可以采用Map join優(yōu)化，從上圖可以看出在Map Join和Sorted Merge Join上DAG圖是兩種結(jié)構(gòu)，因此這需要DAG調(diào)度在執(zhí)行過程中具有足夠的動(dòng)態(tài)性，能夠動(dòng)態(tài)的修改DAG圖來達(dá)到執(zhí)行效率的最優(yōu)。我們?cè)诎⒗锛瘓F(tuán)和云上海量業(yè)務(wù)的實(shí)踐中發(fā)現(xiàn)，類似map join優(yōu)化的這樣的例子是很普遍的，從這些例子可以看出，隨著大數(shù)據(jù)平臺(tái)優(yōu)化的深入進(jìn)行，對(duì)于DAG系統(tǒng)的動(dòng)態(tài)性要求越來越高。

由于業(yè)界大部分DAG調(diào)度框架都在邏輯圖和物理圖之間沒有清晰的分層，缺少執(zhí)行過程中的動(dòng)態(tài)性，無法滿足多種計(jì)算模式的需求。例如spark社區(qū)很早提出了運(yùn)行時(shí)調(diào)整Join策略的需求(Join: Determine the join strategy (broadcast join or shuffle join) at runtime)，但是目前仍然沒有解決。

除此上述用戶體感明顯的場景之外，隨著MaxCompute計(jì)算引擎本身更新?lián)Q代和優(yōu)化器能力的增強(qiáng)，以及PAI平臺(tái)的新功能演進(jìn)，上層的計(jì)算引擎自身能力在不斷的增強(qiáng)。對(duì)于DAG組件在作業(yè)管理，DAG執(zhí)行等方面的動(dòng)態(tài)性，靈活性等方面的需求也日益強(qiáng)烈。在這樣的一個(gè)大的背景下，為了支撐計(jì)算平臺(tái)下個(gè)10年的發(fā)展，伏羲團(tuán)隊(duì)啟動(dòng)了DAG 2.0的項(xiàng)目，在更好的支撐上層計(jì)算需求。

4.1.2 DAG2.0 動(dòng)態(tài)靈活統(tǒng)一的執(zhí)行框架

DAG2.0通過邏輯圖和物理圖的清晰分層，可擴(kuò)展的狀態(tài)機(jī)管理，插件式的系統(tǒng)管理，以及基于事件驅(qū)動(dòng)的調(diào)度策略等基座設(shè)計(jì)，實(shí)現(xiàn)了對(duì)計(jì)算平臺(tái)上多種計(jì)算模式的統(tǒng)一管理，并更好的提供了作業(yè)執(zhí)行過程中在不同層面上的動(dòng)態(tài)調(diào)整能力。作業(yè)執(zhí)行的動(dòng)態(tài)性和統(tǒng)一DAG執(zhí)行框架是DAG2.0的兩個(gè)主要特色：

作業(yè)執(zhí)行的動(dòng)態(tài)性

如前所訴，分布式作業(yè)執(zhí)行的許多物理特性相關(guān)的問題，在作業(yè)運(yùn)行前是無法被感知的。例如一個(gè)分布式作業(yè)在運(yùn)行前，能夠獲得的只有原始輸入的一些基本特性(數(shù)據(jù)量等)， 對(duì)于一個(gè)較深的DAG執(zhí)行而言，這也就意味著只有根節(jié)點(diǎn)的物理計(jì)劃(并發(fā)度選擇等) 可能相對(duì)合理，而下游的節(jié)點(diǎn)和邊的物理特性只能通過一些特定的規(guī)則來猜測。這就帶來了執(zhí)行過程中的不確定性，因此，要求一個(gè)好的分布式作業(yè)執(zhí)行系統(tǒng)，需要能夠根據(jù)中間運(yùn)行結(jié)果的特點(diǎn)，來進(jìn)行執(zhí)行過程中的動(dòng)態(tài)調(diào)整。

而DAG/AM作為分布式作業(yè)唯一的中心節(jié)點(diǎn)和調(diào)度管控節(jié)點(diǎn)，是唯一有能力收集并聚合相關(guān)數(shù)據(jù)信息，并基于這些數(shù)據(jù)特性來做作業(yè)執(zhí)行的動(dòng)態(tài)調(diào)整。這包括簡單的物理執(zhí)行圖調(diào)整（比如動(dòng)態(tài)的并發(fā)度調(diào)整），也包括復(fù)雜一點(diǎn)的調(diào)整比如對(duì)shuffle方式和數(shù)據(jù)編排方式重組。除此以外，數(shù)據(jù)的不同特點(diǎn)也會(huì)帶來邏輯執(zhí)行圖調(diào)整的需求：對(duì)于邏輯圖的動(dòng)態(tài)調(diào)整，在分布式作業(yè)處理中是一個(gè)全新的方向，也是我們?cè)贒AG 2.0里面探索的新式解決方案。

還是以map join優(yōu)化作為例子，由于map join與默認(rèn)join方式(sorted merge join)對(duì)應(yīng)的其實(shí)是兩種不同優(yōu)化器執(zhí)行計(jì)劃，在DAG層面，對(duì)應(yīng)的是兩種不同的邏輯圖。DAG2.0的動(dòng)態(tài)邏輯圖能力很好的支持了這種運(yùn)行過程中根據(jù)中間數(shù)據(jù)特性的動(dòng)態(tài)優(yōu)化，而通過與上層引擎優(yōu)化器的深度合作，在2.0上實(shí)現(xiàn)了業(yè)界首創(chuàng)的conditional join方案。如同下圖展示，在對(duì)于join使用的算法無法被事先確定的時(shí)候，分布式調(diào)度執(zhí)行框架可以允許優(yōu)化提交一個(gè)conditional DAG，這樣的DAG同時(shí)包括使用兩種不同join的方式對(duì)應(yīng)的不同執(zhí)行計(jì)劃支路。在實(shí)際執(zhí)行時(shí)，AM根據(jù)上游產(chǎn)出數(shù)據(jù)量，動(dòng)態(tài)選擇一條支路執(zhí)行(plan A or plan B)。這樣子的動(dòng)態(tài)邏輯圖執(zhí)行流程，能夠保證每次作業(yè)運(yùn)行時(shí)，根據(jù)實(shí)際產(chǎn)生的中間數(shù)據(jù)特性，選擇最優(yōu)的執(zhí)行計(jì)劃。在這個(gè)例子中，

• 當(dāng)M1輸出的數(shù)據(jù)量較小時(shí)，允許其輸出被全量載入下游單個(gè)計(jì)算節(jié)點(diǎn)的內(nèi)存，DAG就會(huì)選擇優(yōu)化的map join(plan A)，來避免額外的shuffle和排序。
• 當(dāng)M1輸出的數(shù)據(jù)量大到一定程度，已經(jīng)不屬于map join的適用范圍，DAG就可以自動(dòng)選擇走merge join，來保證作業(yè)的成功執(zhí)行。

除了map join這個(gè)典型場景外，借助DAG2.0的動(dòng)態(tài)調(diào)度能力，MaxCompute在解決其他用戶痛點(diǎn)上也做了很多探索，并取得了不錯(cuò)的效果。例如智能動(dòng)態(tài)并發(fā)度調(diào)整：在執(zhí)行過程中依據(jù)分區(qū)數(shù)據(jù)統(tǒng)計(jì)調(diào)整，動(dòng)態(tài)調(diào)整并發(fā)度；自動(dòng)合并小分區(qū)，避免不必要的資源使用，節(jié)約用戶資源使用；切分大分區(qū)，避免不必要的長尾出現(xiàn)等等。

統(tǒng)一的AM/DAG執(zhí)行框架

除了動(dòng)態(tài)性在SQL執(zhí)行中帶來的重大性能提升外，DAG 2.0抽象分層的點(diǎn)，邊，圖架構(gòu)上，也使其能通過對(duì)點(diǎn)和邊上不同物理特性的描述，對(duì)接不同的計(jì)算模式。業(yè)界各種分布式數(shù)據(jù)處理引擎，包括SPARK, FLINK, HIVE, SCOPE, TENSORFLOW等等，其分布式執(zhí)行框架的本源都可以歸結(jié)于Dryad提出的DAG模型。我們認(rèn)為對(duì)于圖的抽象分層描述，將允許在同一個(gè)DAG系統(tǒng)中，對(duì)于離線/實(shí)時(shí)/流/漸進(jìn)計(jì)算等多種模型都可以有一個(gè)好的描述。

如果我們對(duì)分布式SQL進(jìn)行細(xì)分的話，可以看見業(yè)界對(duì)于不同場景上的優(yōu)化經(jīng)常走在兩個(gè)極端：要么優(yōu)化throughput (大規(guī)模，相對(duì)高延時(shí))，要么優(yōu)化latency(中小數(shù)據(jù)量，迅速完成)。前者以Hive為典型代表，后者則以Spark以及各種分布式MPP解決方案為代表。而在阿里分布式系統(tǒng)的發(fā)展過程中，歷史上同樣出現(xiàn)了兩種對(duì)比較為顯著的執(zhí)行方式：SQL線離線(batch)作業(yè)與準(zhǔn)實(shí)時(shí)(interactive)作業(yè)。這兩種模式的資源管理和作業(yè)執(zhí)行，過去是搭建在兩套完全分開的代碼實(shí)現(xiàn)上的。這除了導(dǎo)致兩套代碼和功能無法復(fù)用以外，兩種計(jì)算模式的非黑即白，使得彼此在資源利用率和執(zhí)行性能之間無法tradeoff。而在DAG 2.0模型上，通過對(duì)點(diǎn)/邊物理特性的映射，實(shí)現(xiàn)了這兩種計(jì)算模式比較自然的融合和統(tǒng)一。離線作業(yè)和準(zhǔn)實(shí)時(shí)作業(yè)在邏輯節(jié)點(diǎn)和邏輯邊上映射不同的物理特性后，都能得到準(zhǔn)確的描述:

• 離線作業(yè)：每個(gè)節(jié)點(diǎn)按需去申請(qǐng)資源，一個(gè)邏輯節(jié)點(diǎn)代表一個(gè)調(diào)度單位；節(jié)點(diǎn)間連接邊上傳輸?shù)臄?shù)據(jù)，通過落盤的方式來保證可靠性；
• 準(zhǔn)實(shí)時(shí)作業(yè)：整個(gè)作業(yè)的所有節(jié)點(diǎn)都統(tǒng)一在一個(gè)調(diào)度單位內(nèi)進(jìn)行g(shù)ang scheduling；節(jié)點(diǎn)間連接邊上通過網(wǎng)絡(luò)/內(nèi)存直連傳輸數(shù)據(jù)，并利用數(shù)據(jù)pipeline來追求最優(yōu)的性能。

在此統(tǒng)一離線作業(yè)與準(zhǔn)實(shí)時(shí)作業(yè)的到一套架構(gòu)的基礎(chǔ)上，這種統(tǒng)一的描述方式，使得探索離線作業(yè)高資源利用率，以及準(zhǔn)實(shí)時(shí)作業(yè)的高性能之間的tradeoff成為可能：當(dāng)調(diào)度單位可以自由調(diào)整，就可以實(shí)現(xiàn)一種全新的混合的計(jì)算模式，我們稱之為Bubble執(zhí)行模式。

這種混合Bubble模式，使得DAG的用戶，也就是上層計(jì)算引擎的開發(fā)者(比如MaxCompute的優(yōu)化器)，能夠結(jié)合執(zhí)行計(jì)劃的特點(diǎn)，以及引擎終端用戶對(duì)資源使用和性能的敏感度，來靈活選擇在執(zhí)行計(jì)劃中切出Bubble子圖。在Bubble內(nèi)部充分利用網(wǎng)絡(luò)直連和計(jì)算節(jié)點(diǎn)預(yù)熱等方式提升性能，沒有切入Bubble的節(jié)點(diǎn)則依然通過傳統(tǒng)離線作業(yè)模式運(yùn)行。在統(tǒng)一的新模型之上，計(jì)算引擎和執(zhí)行框架可以在兩個(gè)極端之間，根據(jù)具體需要，選擇不同的平衡點(diǎn)。

4.1.3 效果

DAG2.0的動(dòng)態(tài)性使得很多執(zhí)行優(yōu)化可以運(yùn)行時(shí)決定，使得實(shí)際執(zhí)行的效果更優(yōu)。例如，在阿里內(nèi)部的作業(yè)中，動(dòng)態(tài)的conditional join相比靜態(tài)的執(zhí)行計(jì)劃，整體獲得了將近3X的性能提升。

混合Bubble執(zhí)行模式平衡了離線作業(yè)高資源利用率以及準(zhǔn)實(shí)時(shí)作業(yè)的高性能，這在1TB TPCH測試集上有顯著的體現(xiàn)，

• Bubble相對(duì)離線作業(yè)：在多使用20%資源的情況下，Bubble模式性能提升將近一倍；
• Bubble相對(duì)準(zhǔn)實(shí)時(shí)模式：在節(jié)省了2.6X資源情況下， Bubble性能僅下降15%;

4.2 Fuxi Shuffle 2.0 - 磁盤內(nèi)存網(wǎng)絡(luò)的最佳使用

4.2.1 背景

大數(shù)據(jù)計(jì)算作業(yè)中，節(jié)點(diǎn)間的數(shù)據(jù)傳遞稱為shuffle, 主流分布式計(jì)算系統(tǒng)都提供了數(shù)據(jù)shuffle服務(wù)的子系統(tǒng)。如前述DAG計(jì)算模型中，task間的上下游數(shù)據(jù)傳輸就是典型的shuffle過程。

在數(shù)據(jù)密集型作業(yè)中，shuffle階段的時(shí)間和資源使用占比非常高，有其他大數(shù)據(jù)公司研究顯示，在大數(shù)據(jù)計(jì)算平臺(tái)上Shuffle階段均是在所有作業(yè)的資源使用中占比超過50%. 根據(jù)統(tǒng)計(jì)在MaxCompute生產(chǎn)中shuffle占作業(yè)運(yùn)行時(shí)間和資源消耗的30-70%，因此優(yōu)化shuffle流程不但可以提升作業(yè)執(zhí)行效率，而且可以整體上降低資源使用，節(jié)約成本，提升MaxCompute在云計(jì)算市場的競爭優(yōu)勢。

從shuffle介質(zhì)來看，最廣泛使用的shuffle方式是基于磁盤文件的shuffle. 這種模式這種方式簡單，直接，通常只依賴于底層的分布式文件系統(tǒng)，適用于所有類型作業(yè)。而在典型的常駐內(nèi)存的實(shí)時(shí)/準(zhǔn)實(shí)時(shí)計(jì)算中，通常使用網(wǎng)絡(luò)直連shuffle的方式追求極致性能。Fuxi Shuffle在1.0版本中將這兩種shuffle模式進(jìn)行了極致優(yōu)化，保障了日常和高峰時(shí)期作業(yè)的高效穩(wěn)定運(yùn)行。

挑戰(zhàn)

我們先以使用最廣泛的，基于磁盤文件系統(tǒng)的離線作業(yè)shuffle為例。

通常每個(gè)mapper生成一個(gè)磁盤文件，包含了這個(gè)mapper寫給下游所有reducer的數(shù)據(jù)。而一個(gè)reducer要從所有mapper所寫的文件中，讀取到屬于自己的那一小塊。右側(cè)則是一個(gè)系統(tǒng)中典型規(guī)模的MR作業(yè)，當(dāng)每個(gè)mapper處理256MB數(shù)據(jù)，而下游reducer有10000個(gè)時(shí)，平均每個(gè)reducer讀取來自每個(gè)mapper的數(shù)據(jù)量就是25.6KB, 在機(jī)械硬盤HDD為介質(zhì)的存儲(chǔ)系統(tǒng)中，屬于典型的讀碎片現(xiàn)象，因?yàn)榧僭O(shè)我們的磁盤iops能達(dá)到1000, 對(duì)應(yīng)的throughput也只有25MB/s, 嚴(yán)重影響性能和磁盤壓力。

【基于文件系統(tǒng)shuffle的示意圖 / 一個(gè)20000*10000的MR作業(yè)的碎片讀】

分布式作業(yè)中并發(fā)度的提升往往是加速作業(yè)運(yùn)行的最重要手段之一。但處理同樣的數(shù)據(jù)量，并發(fā)度越高意味著上述碎片讀現(xiàn)象越嚴(yán)重。通常情況下選擇忍受一定的碎片IO現(xiàn)象而在集群規(guī)模允許的情況下提升并發(fā)度，還是更有利于作業(yè)的性能。所以碎片IO現(xiàn)象在線上普遍存在，磁盤也處于較高的壓力水位。

一個(gè)線上的例子是，某些主流集群單次讀請(qǐng)求size為50-100KB, Disk util指標(biāo)長期維持在90%的警戒線上。這些限制了對(duì)作業(yè)規(guī)模的進(jìn)一步追求。

我們不禁考慮，作業(yè)并發(fā)度和磁盤效率真的不能兼得嗎？

4.2.2 Fuxi的答案：Fuxi Shuffle 2.0

引入Shuffle Service - 高效管理shuffle資源

為了針對(duì)性地解決上述碎片讀問題及其引發(fā)的一連串負(fù)面效應(yīng)，我們?nèi)麓蛟炝嘶趕huffle service的shuffle模式。Shuffle service的最基本工作方式是，在集群每臺(tái)機(jī)器部署一個(gè)shuffle
agent節(jié)點(diǎn)，用來歸集寫給同一reducer的shuffle數(shù)據(jù)。如下圖

可以看到，mapper生成shuffle數(shù)據(jù)的過程變?yōu)閙apper將shuffle數(shù)據(jù)通過網(wǎng)絡(luò)傳輸給每個(gè)reducer對(duì)應(yīng)的shuffle agent, 而shuffle agent歸集一個(gè)reducer來自所有mapper的數(shù)據(jù)，并追加到shuffle磁盤文件中，兩個(gè)過程是流水線并行化起來的。

Shuffle agent的歸集功能將reducer的input數(shù)據(jù)從碎片變?yōu)榱诉B續(xù)數(shù)據(jù)文件，對(duì)HDD介質(zhì)相當(dāng)友好。由此，整個(gè)shuffle過程中對(duì)磁盤的讀寫均為連續(xù)訪問。從標(biāo)準(zhǔn)的TPCH等測試中可以看到不同場景下性能可取得百分之幾十到幾倍的提升，且大幅降低磁盤壓力、提升CPU等資源利用率。

Shuffle Service的容錯(cuò)機(jī)制

Shuffle service的歸集思想在公司內(nèi)外都有不同的工作展現(xiàn)類似的思想，但都限于“跑分”和小范圍使用。因?yàn)檫@種模式對(duì)于各環(huán)節(jié)的錯(cuò)誤天生處理困難。

以shuffle agent文件丟失/損壞是大數(shù)據(jù)作業(yè)的常見問題為例，傳統(tǒng)的文件系統(tǒng)shuffle可以直接定位到出錯(cuò)的數(shù)據(jù)文件來自哪個(gè)mapper，只要重跑這個(gè)mapper即可恢復(fù)。但在前述shuffle service流程中，由于shuffle agent輸出的shuffle這個(gè)文件包含了來自所有mapper的shuffle數(shù)據(jù)，損壞文件的重新生成需要以重跑所有mapper為代價(jià)。如果這種機(jī)制應(yīng)用于所有線上作業(yè)，顯然是不可接受的。

我們?cè)O(shè)計(jì)了數(shù)據(jù)雙副本機(jī)制解決了這個(gè)問題，使得大多數(shù)通常情況下reducer可以讀取到高效的agent生成的數(shù)據(jù)，而當(dāng)少數(shù)agent數(shù)據(jù)丟失的情況，可以讀取備份數(shù)據(jù)，備份數(shù)據(jù)的重新生成只依賴特定的上游mapper.

具體來說，mapper產(chǎn)生的每份shuffle數(shù)據(jù)除了發(fā)送給對(duì)于shuffle agent外，也會(huì)按照與傳統(tǒng)文件系統(tǒng)shuffle數(shù)據(jù)類似的格式，在本地寫一個(gè)備份。按前面所述，這份數(shù)據(jù)寫的代價(jià)較小但讀取的性能不佳，但由于僅在shuffle agent那個(gè)副本出錯(cuò)時(shí)才會(huì)讀到備份數(shù)據(jù)，所以對(duì)作業(yè)整體性能影響很小，也不會(huì)引起集群級(jí)別的磁盤壓力升高。

有效的容錯(cuò)機(jī)制使得shuffle service相對(duì)于文件系統(tǒng)shuffle，在提供更好的作業(yè)性能的同時(shí)，因shuffle數(shù)據(jù)出錯(cuò)的task重試比例降低了一個(gè)數(shù)量級(jí)，給線上全面投入使用打好了穩(wěn)定性基礎(chǔ)。

線上生產(chǎn)環(huán)境的極致性能穩(wěn)定性

在前述基礎(chǔ)功能之上，Fuxi線上的shuffle系統(tǒng)應(yīng)用了更多功能和優(yōu)化，在性能、成本、穩(wěn)定性等方便取得了進(jìn)一步的提升。舉例如下。

1. 流控和負(fù)載均衡
前面的數(shù)據(jù)歸集模型中，shuffle agent作為新角色銜接了mapper的數(shù)據(jù)發(fā)送與數(shù)據(jù)落盤。分布式集群中磁盤、網(wǎng)絡(luò)等問題可能影響這條鏈路上的數(shù)據(jù)傳輸，節(jié)點(diǎn)本身的壓力也可能影響shuffle agent的工作狀態(tài)。當(dāng)因集群熱點(diǎn)等原因使得shuffle agent負(fù)載過重時(shí)，我們提供了必要的流控措施緩解網(wǎng)絡(luò)和磁盤的壓力；和模型中一個(gè)reducer有一個(gè)shuffle agent收集數(shù)據(jù)不同，我們使用了多個(gè)shuffle agent承擔(dān)同樣的工作，當(dāng)發(fā)生數(shù)據(jù)傾斜時(shí)，這個(gè)方式可以有效地將壓力分散到多個(gè)節(jié)點(diǎn)上。從線上表現(xiàn)看，這些措施消除了絕大多數(shù)的shuffle期間擁塞流控和集群負(fù)載不均現(xiàn)象。


2. 故障shuffle
agent的切換
各種軟硬件故障導(dǎo)致shuffle agent對(duì)某個(gè)reducer的數(shù)據(jù)工作不正常時(shí)，后續(xù)數(shù)據(jù)可以實(shí)時(shí)切換到其他正常shuffle agent. 這樣，就會(huì)有更多的數(shù)據(jù)可以從shuffle agent側(cè)讀到，而減少低效的備份副本訪問。


3. Shuffle agent數(shù)據(jù)的回追
很多時(shí)候發(fā)生shuffle
agent切換時(shí)（如機(jī)器下線），原shuffle agent生成的數(shù)據(jù)可能已經(jīng)丟失或訪問不到。在后續(xù)數(shù)據(jù)發(fā)送到新的shuffle agent同時(shí)，Fuxi還會(huì)將丟失的部分?jǐn)?shù)據(jù)從備份副本中l(wèi)oad起來并同樣發(fā)送給新的shuffle agent, 使得后續(xù)reducer所有的數(shù)據(jù)都可以讀取自shuffle agent側(cè)，極大地提升了容錯(cuò)情況下的作業(yè)性能。


4. 新shuffle模式的探索
前述數(shù)據(jù)歸集模型及全面擴(kuò)展優(yōu)化，在線上集群中單位資源處理的數(shù)據(jù)量提升了約20%, 而因出錯(cuò)重試的發(fā)生頻率降至原來文件系統(tǒng)shuffle的5%左右。但這就是最高效的shuffle方式了嗎？

我們?cè)谏a(chǎn)環(huán)境對(duì)部分作業(yè)應(yīng)用了一種新的shuffle模型，這種模型中mapper的發(fā)送端和reducer的接收端都通過一個(gè)agent節(jié)點(diǎn)來中轉(zhuǎn)shuffle流量。線上已經(jīng)有部分作業(yè)使用此種方式并在性能上得到了進(jìn)一步的提升。

內(nèi)存數(shù)據(jù)shuffle

離線大數(shù)據(jù)作業(yè)可能承擔(dān)了主要的計(jì)算數(shù)據(jù)量，但流行的大數(shù)據(jù)計(jì)算系統(tǒng)中有非常多的場景是通過實(shí)時(shí)/準(zhǔn)實(shí)時(shí)方式運(yùn)行的，作業(yè)全程的數(shù)據(jù)流動(dòng)發(fā)生在網(wǎng)絡(luò)和內(nèi)存，從而在有限的作業(yè)規(guī)模下取得極致的運(yùn)行性能，如大家熟悉的Spark, Flink等系統(tǒng)。

Fuxi DAG也提供了實(shí)時(shí)/準(zhǔn)實(shí)時(shí)作業(yè)運(yùn)行環(huán)境，傳統(tǒng)的shuffle方式是通過網(wǎng)絡(luò)直連，也能收到明顯優(yōu)于離線shuffle的性能。這種方式下，要求作業(yè)中所有節(jié)點(diǎn)都要調(diào)度起來才能開始運(yùn)行，限制了作業(yè)的規(guī)模。而實(shí)際上多數(shù)場景計(jì)算邏輯生成shuffle數(shù)據(jù)的速度不足以填滿shuffle帶寬，運(yùn)行中的計(jì)算節(jié)點(diǎn)等待數(shù)據(jù)的現(xiàn)象明顯，性能提升付出了資源浪費(fèi)的代價(jià)。

我們將shuffle service應(yīng)用到內(nèi)存存儲(chǔ)中，以替換network傳輸?shù)膕huffle方式。一方面，這種模式解耦了上下游調(diào)度，整個(gè)作業(yè)不再需要全部節(jié)點(diǎn)同時(shí)拉起；另一方面通過精確預(yù)測數(shù)據(jù)的讀寫速度并適時(shí)調(diào)度下游節(jié)點(diǎn)，可以取得與network傳輸shuffle相當(dāng)?shù)淖鳂I(yè)性能，而資源消耗降低50%以上。這種shuffle方式還使得DAG系統(tǒng)中多種運(yùn)行時(shí)調(diào)整DAG的能力可以應(yīng)用到實(shí)時(shí)/準(zhǔn)實(shí)時(shí)作業(yè)中。

4.2.3 收益

Fuxi Shuffle 2.0全面上線生產(chǎn)集群，處理同樣數(shù)據(jù)量的作業(yè)資源比原來節(jié)省15%，僅shuffle方式的變化就使得磁盤壓力降低23%，作業(yè)運(yùn)行中發(fā)生錯(cuò)誤重試的比例降至原來的5%。

【線上典型集群的性能與穩(wěn)定性提升示意圖（不同組數(shù)據(jù)表示不同集群）】

對(duì)使用內(nèi)存shuffle的準(zhǔn)實(shí)時(shí)作業(yè)，我們?cè)赥PCH等標(biāo)準(zhǔn)測試集中與網(wǎng)絡(luò)shuffle性能相當(dāng)，資源使用只有原來的30%左右，且支持了更大的作業(yè)規(guī)模，和DAG 2.0系統(tǒng)更多的動(dòng)態(tài)調(diào)度功能應(yīng)用至準(zhǔn)實(shí)時(shí)作業(yè)。

5. 單機(jī)調(diào)度

大量分布式作業(yè)匯集到一臺(tái)機(jī)器上，如何將單機(jī)有限的各種資源合理分配給每個(gè)作業(yè)使用，從而達(dá)到作業(yè)運(yùn)行質(zhì)量、資源利用率、作業(yè)穩(wěn)定性的多重保障，是單機(jī)調(diào)度要解決的任務(wù)。

典型的互聯(lián)網(wǎng)公司業(yè)務(wù)一般區(qū)分為離線業(yè)務(wù)與在線業(yè)務(wù)兩種類型。在阿里巴巴，我們也同樣有在線業(yè)務(wù)如淘寶、天貓、釘釘、Blink等，這類業(yè)務(wù)的特點(diǎn)是對(duì)響應(yīng)延遲特別敏感，一旦服務(wù)抖動(dòng)將會(huì)出現(xiàn)添加購物車失敗、下單失敗、瀏覽卡頓、釘釘消息發(fā)送失敗等各種異常情況，嚴(yán)重影響用戶體驗(yàn)，同時(shí)為了應(yīng)對(duì)在618、雙11等各種大促的情況，需要提前準(zhǔn)備大量的機(jī)器。由于以上種種原因，日常狀態(tài)這些機(jī)器的資源利用率不足10%，產(chǎn)生資源浪費(fèi)的情況。與此同時(shí)，阿里的離線業(yè)務(wù)又是另外一幅風(fēng)景，MaxCompute計(jì)算平臺(tái)承擔(dān)了阿里所有大數(shù)據(jù)離線計(jì)算業(yè)務(wù)類型，各個(gè)集群資源利用率常態(tài)超負(fù)載運(yùn)行，數(shù)據(jù)量和計(jì)算量每年都在保持高速增長。

一方面是在線業(yè)務(wù)資源利用率不足，另一方面是離線計(jì)算長期超負(fù)載運(yùn)行，那么能否將在線業(yè)務(wù)與離線計(jì)算進(jìn)行混合部署，提升資源利用率同時(shí)大幅降低成本，實(shí)現(xiàn)共贏。

5.1 三大挑戰(zhàn)

如何保障在線服務(wù)質(zhì)量
在線集群的平均CPU利用率只有10%左右，混部的目標(biāo)就是將剩余的資源提供給MaxCompute進(jìn)行離線計(jì)算使用，從而達(dá)到節(jié)約成本的目的。那么，如何能夠保障資源利用率提升的同時(shí)又能夠保護(hù)在線服務(wù)不受影響呢？

如何保障離線穩(wěn)定
當(dāng)資源發(fā)生沖突時(shí)，第一反應(yīng)往往是保護(hù)在線，犧牲離線。畢竟登不上淘寶天貓下不了單可是大故障。可是，離線如果無限制的犧牲下去，服務(wù)質(zhì)量將會(huì)出現(xiàn)大幅度下降。試想，我在dataworks上跑個(gè)SQL，之前一分鐘就出結(jié)果，現(xiàn)在十幾分鐘甚至一個(gè)小時(shí)都跑不出來，大數(shù)據(jù)分析的同學(xué)估計(jì)也受不了了。

如何衡量資源質(zhì)量
電商業(yè)務(wù)通過富容器的方式集成多種容器粒度的分析手段，但是前文描述過離線作業(yè)的特點(diǎn)，如何能夠精準(zhǔn)的對(duì)離線作業(yè)資源使用進(jìn)行資源畫像分析，如果能夠評(píng)估資源受干擾的程度，混部集群的穩(wěn)定性等問題，是對(duì)我們的又一個(gè)必須要解決的挑戰(zhàn)

5.2 資源隔離分級(jí)管理

單機(jī)的物理資源總是有限的，按照資源特性可以大體劃分為可伸縮資源與不可伸縮資源兩大類。CPU、Net、IO等屬于可伸縮資源，Memory屬于不可伸縮資源，不同類型的資源有不同層次的資源隔離方案。另一方面，通用集群中作業(yè)類型種類繁多，不同作業(yè)類型對(duì)資源的訴求是不同的。這里包括在線、離線兩個(gè)大類的資源訴求，同時(shí)也包含了各自內(nèi)部不同層次的優(yōu)先級(jí)二次劃分需求，十分復(fù)雜。

基于此，Fuxi2.0提出了一套基于資源優(yōu)先級(jí)的資源劃分邏輯，在資源利用率、多層次資源保障復(fù)雜需求尋找到了解決方案。

下面我們將針對(duì)CPU分級(jí)管理進(jìn)行深入描述，其他維度資源管理策略我們將在今后的文章中進(jìn)行深入介紹。

CPU分級(jí)管理

通過精細(xì)的組合多種內(nèi)核策略，將CPU區(qū)分為高、中、低三類優(yōu)先級(jí)

隔離策略如下圖所示

基于不同類型的資源對(duì)應(yīng)不同的優(yōu)先級(jí)作業(yè)

5.3 資源畫像

Fuxi作為資源調(diào)度模塊，對(duì)資源使用情況的精準(zhǔn)畫像是衡量資源分配，調(diào)查/分析/解決解決資源問題的關(guān)鍵。針對(duì)在線作業(yè)的資源情況，集團(tuán)和業(yè)界都有較多的解決方案。這類通用的資源采集角色存在以下無法解決的問題無法應(yīng)用于離線作業(yè)資源畫像的數(shù)據(jù)采集階段

1. 采集時(shí)間精度過低。大部分信息是分鐘級(jí)別，而MaxCompute作業(yè)大部分運(yùn)行時(shí)間在秒級(jí)。
2. 無法定位MaxCompute信息。MaxCompute是基于Cgroup資源隔離，因此以上工具無法針對(duì)作業(yè)進(jìn)行針對(duì)性采集
3. 采集指標(biāo)不足。有大量新內(nèi)核新增的微觀指標(biāo)需要進(jìn)行收集，過去是不支持的

為此，我們提出了FuxiSensor的資源畫像方案，架構(gòu)如上圖所示，同時(shí)利用SLS進(jìn)行數(shù)據(jù)的收集和分析。在集群、Job作業(yè)、機(jī)器、worker等不同層次和粒度實(shí)現(xiàn)了資源信息的畫像，實(shí)現(xiàn)了秒級(jí)的數(shù)據(jù)采集精度。在混部及MaxCompute的實(shí)踐中，成為資源問題監(jiān)控、報(bào)警、穩(wěn)定性數(shù)據(jù)分析、作業(yè)異常診斷、資源監(jiān)控狀況的統(tǒng)一入口，成為混部成功的關(guān)鍵指標(biāo)。

5.4 線上效果

日常資源利用率由10%提升到40%以上

在線抖動(dòng)小于5%

5.5 單機(jī)調(diào)度小結(jié)

為了解決三大挑戰(zhàn)，通過完善的各維度優(yōu)先級(jí)隔離策略，將在線提升到高優(yōu)先級(jí)資源維度，我們保障了在線的服務(wù)質(zhì)量穩(wěn)定；通過離線內(nèi)部優(yōu)先級(jí)區(qū)分及各種管理策略，實(shí)現(xiàn)了離線質(zhì)量的穩(wěn)定性保障；通過細(xì)粒度資源畫像信息，實(shí)現(xiàn)了資源使用的評(píng)估與分析，最終實(shí)現(xiàn)了混部在阿里的大規(guī)模推廣與應(yīng)用，從而大量提升了集群資源利用率，為離線計(jì)算節(jié)省了大量成本。

6. 展望

從2009到2019年歷經(jīng)十年的錘煉，伏羲系統(tǒng)仍然在不斷的演化，滿足不斷涌現(xiàn)的業(yè)務(wù)新需求，引領(lǐng)分布式調(diào)度技術(shù)的發(fā)展。接下來，我們會(huì)從以下幾個(gè)方面繼續(xù)創(chuàng)新：

• 資源調(diào)度FuxiMaster將基于機(jī)器學(xué)習(xí)，實(shí)現(xiàn)智能化調(diào)度策略和動(dòng)態(tài)精細(xì)的資源管理模式，進(jìn)一步提高集群資源利用率，提供更強(qiáng)大靈活的分布式集群資源管理服務(wù)。
• 新一代DAG2.0繼續(xù)利用動(dòng)態(tài)性精耕細(xì)作，優(yōu)化各種不同類型的作業(yè)；與SQL深入合作，解決線上痛點(diǎn)，推動(dòng)SQL引擎深度優(yōu)化，提升性能的同時(shí)也讓SQL作業(yè)運(yùn)行更加智能化；探索機(jī)器學(xué)習(xí)場景的DAG調(diào)度，改善訓(xùn)練作業(yè)的效率，提升GPU使用率。
• 數(shù)據(jù)Shuffle2.0則一方面優(yōu)化shuffle流程，追求性能、成本、穩(wěn)定性的極致，另一方面與DAG 2.0深入結(jié)合，提升更多場景；同時(shí)探索新的軟硬件架構(gòu)帶來的新的想象空間。
• 智能化的精細(xì)單機(jī)資源管控，基于資源畫像信息通過對(duì)歷史數(shù)據(jù)分析產(chǎn)生未來趨勢預(yù)測，通過多種資源管控手段進(jìn)行精準(zhǔn)的資源控制，實(shí)現(xiàn)資源利用率和不同層次服務(wù)質(zhì)量的完美均衡。

最后，我們熱忱歡迎集團(tuán)各個(gè)團(tuán)隊(duì)一起交流探討，共同打造世界一流的分布式調(diào)度系統(tǒng)！

MaxCompute產(chǎn)品官網(wǎng)?https://www.aliyun.com/product/odps
更多阿里巴巴大數(shù)據(jù)計(jì)算技術(shù)交流，歡迎掃碼加入“MaxCompute開發(fā)者社區(qū)”釘釘群。

原文鏈接
本文為云棲社區(qū)原創(chuàng)內(nèi)容，未經(jīng)允許不得轉(zhuǎn)載。

總結(jié)

以上是生活随笔為你收集整理的面向大数据与云计算调度挑战的阿里经济体核心调度系统的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。