編者按

伏羲（Fuxi）是十年前最初創立飛天平臺時的三大服務之一（分布式存儲 Pangu，分布式計算 MaxCompute，分布式調度 Fuxi），當時的設計初衷是為了解決大規模分布式資源的調度問題（本質上是多目標的最優匹配問題）。

隨阿里經濟體和阿里云豐富的業務需求（尤其是雙十一）和磨練，伏羲的內涵不斷擴大，從單一的資源調度器（對標開源系統的YARN）擴展成大數據的核心調度服務，覆蓋數據調度（Data Placement）、資源調度（Resouce Management）、計算調度（Application Manager）、和本地微（自治）調度（即正文中的單機調度）等多個領域，并在每一個細分領域致力于打造超越業界主流的差異化能力。

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

1. 引言

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

同樣的10年，伴隨著阿里年年雙11，MaxCompute同樣支撐了阿里內部大數據的蓬勃發展，從原來的幾百臺，到現在的10萬臺物理機規模。

雙線需求，殊途同歸，海量資源池，如何自動匹配到大量不同需求的異地客戶計算需求上，需要調度系統的工作。本文主要介紹阿里大數據的調度系統FUXI往2.0的演進。先給大家介紹幾個概念：
?

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

2013年，伏羲在飛天5K項目中對系統架構進行了第一次大重構，解決了規模、性能、利用率、容錯等線上問題，并取得世界排序大賽Sortbenchmark四項冠軍，這標志著Fuxi 1.0的成熟。

2019年，伏羲再次出發，從技術上對系統進行了第二次重構，發布Fuxi 2.0版本：阿里自研的新一代高性能、分布式的數據、資源、計算、單機調度系統。Fuxi 2.0進行了全面的技術升級，在全區域數據排布、去中心化調度、在線離線混合部署、動態計算等方面全方位滿足新業務場景下的調度需求。

伏羲2.0成果概覽

? 業內首創跨地域多數據中心的數據調度方案-Yugong，通過3%的冗余存儲，節省80%的跨地域網絡帶寬
? 業內領先的去中心化資源調度架構，單集群支持10萬服務器*10萬并發job的高頻調度
? 動態DAG闖入傳統SQL優化盲區，TPC-DS性能提升27%，conditional join性能提升3X。
? 創新性的數據動態shuffle和全局跨級優化，取代業界磁盤shuffle；線上千萬job，整體性能提升20%，成本下降15%，出錯率降低一個數量級
? 在線離線規模化混合部署，在線集群利用率由10%提升到40%，雙十一大促節省4200臺F53資源，且同時保障在線離線業務穩定。

2. 數據調度2.0 - 跨地域的數據調度

阿里巴巴在全球都建有數據中心，每個地區每天會產生一份當地的交易訂單信息，存在就近的數據中心。北京的數據中心，每天會運行一個定時任務來統計當天全球所有的訂單信息，需要從其他數據中心讀取這些交易數據。當數據的產生和消費不在一個數據中心時，我們稱之為跨數據中心數據依賴（下文簡稱跨中心依賴）。

圖. 阿里巴巴全球數據中心

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

圖. MaxCompute平臺數據及依賴增長趨勢

為了解決這個問題，我們在數據中心上增加了一層調度層，用于在數據中心之間調度數據和計算。這層調度獨立于數據中心內部的調度，目的是實現跨地域維度上存儲冗余--計算均衡--長傳帶寬--性能最優之間的最佳平衡。這層調度層包括跨數據中心數據緩存、業務整體排布、作業粒度調度。

首先是對訪問頻次高的數據進行跨數據中心緩存，在緩存空間有限的約束下，選擇合適的數據進行換入換出。不同于其他緩存系統，MaxCompute的數據（分區）以表的形式組織在一起，每張表每天產生一個或多個分區，作業訪問數據也有一些特殊規律，比如一般訪問的是連續分區、生成時間越新的分區訪問概率越大。

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

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

本質上，數據緩存、業務排布、作業粒度調度三者都在解同一個問題，即在跨地域多數據中心系統中減少跨中心依賴量、優化作業的data locality、減少網絡帶寬消耗。

1.2.1 跨數據中心數據緩存策略

我們首次提出了跨地域、跨數據中心數據緩存這一概念，通過集群的存儲換集群間帶寬，在有限的冗余存儲下，找到存儲和帶寬最佳的tradeoff。通過深入的分析MaxCompute的作業、數據的特點，我們設計了一種高效的算法，根據作業歷史的workload、數據的大小和分布，自動進行緩存的換入換出。

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

1.2.2 以project為粒度的多集群業務排布算法

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

下圖展示了某個時刻業務遷移的收益分析：左圖橫軸為遷移的project數量，縱軸為帶寬減少比例，可以看出大約移動60個project就可以減少約30%的帶寬消耗。右圖統計了不同排布下（遷移0個、20個、50個project）的最優帶寬消耗，橫軸為冗余存儲，縱軸為帶寬。

1.2.3 跨數據中心計算調度機制

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

1.3 線上效果

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

3. 資源調度2.0 - 去中心化的多調度器架構

2019年雙十一，MaxCompute平臺產生的數據量已接近EB級別，作業規模達到了千萬，有幾十億的worker跑在幾百萬核的計算單元上，在超大規模（單集群超過萬臺），高并發的場景下，如何快速地給不同的計算任務分配資源，實現資源的高速流轉，需要一個聰明的“大腦”，而這就是集群的資源管理與調度系統（簡稱資源調度系統）。

資源調度系統負責連接成千上萬的計算節點，將數據中心海量的異構資源抽象，并提供給上層的分布式應用，像使用一臺電腦一樣使用集群資源，它的核心能力包括規模、性能、穩定性、調度效果、多租戶間的公平性等等。一個成熟的資源調度系統需要在以下五個方面進行權衡，做到“既要又要”，非常具有挑戰性。

13年的5K項目初步證明了伏羲規模化能力，此后資源調度系統不斷演進，并通過MaxCompute平臺支撐了阿里集團的大數據計算資源需求，在核心調度指標上保持著對開源系統的領先性，比如1）萬臺規模集群，調度延時控制在了10微秒級別，worker啟動延時控制在30毫秒；2）支持任意多級租戶的資源動態調節能力（支持十萬級別的租戶）；3）極致穩定，調度服務全年99.99%的可靠性，并做到服務秒級故障恢復。

2.1 單調度器的局限性

2.1.1 線上的規模與壓力

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

圖. MaxCompute 2015 ~ 2018線上作業情況

但規模的增長將帶來復雜度的極速上升，機器規模擴大一倍，資源請求并發度也會翻一番。在保持既有性能、穩定性、調度效果等核心能力不下降的前提下，可以通過對調度器持續性能優化來擴展集群規模（這也是伏羲資源調度1.0方向），但受限于單機的物理限制，這種優化總會存在天花板，因此需要從架構上優化來徹底規模和性能的可擴展性問題。

2.1.2 調度需求的多樣性

伏羲支持了各種各樣的大數據計算引擎，除了離線計算（SQL、MR)，還包括實時計算、圖計算，以及近幾年迅速發展面向人工智能領域的機器學習引擎。

圖. 資源調度器的架構類型

場景的不同對資源調度的需求也不相同，比如，SQL類型的作業通常體積小、運行時間短，對資源匹配的要求低，但對調度延時要求高，而機器學習的作業一般體積大、運行時間長，調度結果的好壞可能對運行時間產生直接影響，因此也能容忍通過較長的調度延時換取更優的調度結果。資源調度需求這種多樣性，決定了單一調度器很難做到“面面俱到”，需要各個場景能定制各自的調度策略，并進行獨立優化。

2.1.3 灰度發布與工程效率

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

2.2 去中心化的多調度器架構

為了解決上述規模和擴展性問題，更好地滿足多種場景的調度需求，同時從架構上支持灰度能力，伏羲資源調度2.0在1.0的基礎上對調度架構做了大規模的重構，引入了去中心化的多調度器架構。

圖. 資源調度的架構類型

我們將系統中最核心的資源管理和資源調度邏輯進行了拆分解耦，使兩者同時具備了多partition的可擴展能力（如下圖所示），其中：
? 資源調度器（Scheduler）：負責核心的機器資源和作業資源需求匹配的調度邏輯，可以橫向擴展。
? 資源管理和仲裁服務（ResourceManagerService，簡稱RMS）：負責機器資源和狀態管理，對各個Scheduler的調度結果進行仲裁，可以橫向擴展。
? 調度協調服務（Coordinator）：管理資源調度系統的配置信息，Meta信息，以及對機器資源、Scheduler、RMS的可用性和服務角色間的可見性做仲裁。不可橫向擴展，但有秒級多機主備切換能力。
? 調度信息收集監控服務（FuxiEye）：統計集群中每臺機的運行狀態信息，給Scheduler提供調度決策支持，可以橫向擴展。
? 用戶接口服務（ApiServer）：為資源調度系統提供外部調用的總入口，會根據Coordinator提供的Meta信息將用戶請求路由到資源調度系統具體的某一個服務上，可以橫向擴展。

圖. 伏羲多調度器新架構

2.3 上線數據

以下是10w規模集群/10萬作業并發場景調度器核心指標（5個Scheduler、5個RMS，單RMS負責2w臺機器，單Scheduler并發處理2w個作業）。通過數據可以看到，集群10w臺機器的調度利用率超過了99%，關鍵調度指標，單Scheduler向RMS commit的slot的平均數目達到了1w slot/s。

在保持原有單調度器各項核心指標穩定不變的基礎上，去中心化的多調度器框架實現了機器規模和應用并發度的雙向擴展，徹底解決了集群的可擴展性問題。

目前資源調度的新架構已全面上線，各項指標持續穩定。在多調度器架構基礎上，我們把機器學習場景調度策略進行了分離，通過獨立的調度器來進行持續的優化。同時通過測試專用的調度器，我們也讓資源調度具備了灰度能力，調度策略的開發和上線周期顯著縮短。

4. 計算調度2.0 - 從靜態到動態

分布式作業的執行與單機作業的最大區別，在于數據的處理需要拆分到不同的計算節點上，“分而治之”的執行。這個“分”,包括數據的切分，聚合以及對應的不同邏輯運行階段的區分，也包括在邏輯運行階段間數據的shuffle傳輸。每個分布式作業的中心管理點，也就是application master (AM)。這個管理節點也經常被稱為DAG (Directional Acyclic Graph， 有向無環圖) 組件，是因為其最重要的責任，就是負責協調分布式系統中的作業執行流程，包括計算節點的調度以及數據流(shuffle)。

對于作業的邏輯階段和各個計算節點的管理, 以及shuffle策略的選擇/執行，是一個分布式作業能夠正確完成重要前提。這一特點，無論是傳統的MR作業，分布式SQL作業，還是分布式的機器學習/深度學習作業，都是一脈相承的，為了幫助更好的理解計算調度(DAG和Shuffle)在大數據平臺中的位置，我們可以通過MaxCompute分布式SQL的執行過程做為例子來了解：

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

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

這個SQL經過編譯優化之后生成了優化執行計劃，提交到fuxi管理的分布式集群中執行。我們可以看到，這個簡單的SQL經過編譯優化，被轉換成一個具有M->R兩個邏輯節點的DAG圖，也就是傳統上經典的MR類型作業。而這個圖在提交給fuxi系統后，根據每個邏輯節點需要的并發度，數據傳輸邊上的shuffle方式，調度時間等等信息，就被物化成右邊的物理執行圖。物理圖上的每個節點都代表了一個具體的執行實例，實例中包含了具體處理數據的算子，特別的作為一個典型的分布式作業，其中包含了數據交換的算子shuffle——負責依賴外部存儲和網絡交換節點間的數據。一個完整的計算調度，包含了上圖中的DAG的調度執行以及數據shuffle的過程。

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

在云上和阿里集團的大數據實踐中，我們發現對于計算調度需要同時具備超大規模和智能化的需求，以此為基本訴求我們開了Fuxi計算調度2.0的研發。下面就為大家從DAG調度和數據shuffle兩個方面分別介紹計算調度2.0的工作。

4.1 Fuxi DAG 2.0--動態、靈活的分布式計算生態

4.1.1 DAG調度的挑戰

傳統的分布式作業DAG，一般是在作業提交前靜態指定的，這種指定方式，使得作業的運行沒有太多動態調整的空間。放在DAG的邏輯圖與物理圖的背景中來說，這要求分布式系統在運行作業前，必須事先了解作業邏輯和處理數據各種特性，并能夠準確回答作業運行過程，各個節點和連接邊的物理特性問題，然而在現實情況中，許多和運行過程中數據特性相關的問題，都只有個在執行過程中才能被最準確的獲得。靜態的DAG執行，可能導致選中的是非最優的執行計劃，從而導致各種運行時的效率低下，甚至作業失敗。這里我們可以用一個分布式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;

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

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

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

但是在實際執行過程中，具體數據量的大小，需要在上游節點完成后才能被感知，因此在提交作業前很難準確的判斷是否可以采用Map join優化，從上圖可以看出在Map Join和Sorted Merge Join上DAG圖是兩種結構，因此這需要DAG調度在執行過程中具有足夠的動態性，能夠動態的修改DAG圖來達到執行效率的最優。我們在阿里集團和云上海量業務的實踐中發現，類似map join優化的這樣的例子是很普遍的，從這些例子可以看出，隨著大數據平臺優化的深入進行，對于DAG系統的動態性要求越來越高。

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

除此上述用戶體感明顯的場景之外，隨著MaxCompute計算引擎本身更新換代和優化器能力的增強，以及PAI平臺的新功能演進，上層的計算引擎自身能力在不斷的增強。對于DAG組件在作業管理，DAG執行等方面的動態性，靈活性等方面的需求也日益強烈。在這樣的一個大的背景下，為了支撐計算平臺下個10年的發展，伏羲團隊啟動了DAG 2.0的項目，在更好的支撐上層計算需求。

4.1.2 DAG2.0 動態靈活統一的執行框架

DAG2.0通過邏輯圖和物理圖的清晰分層，可擴展的狀態機管理，插件式的系統管理，以及基于事件驅動的調度策略等基座設計，實現了對計算平臺上多種計算模式的統一管理，并更好的提供了作業執行過程中在不同層面上的動態調整能力。作業執行的動態性和統一DAG執行框架是DAG2.0的兩個主要特色：

作業執行的動態性

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

而DAG/AM作為分布式作業唯一的中心節點和調度管控節點，是唯一有能力收集并聚合相關數據信息，并基于這些數據特性來做作業執行的動態調整。這包括簡單的物理執行圖調整（比如動態的并發度調整），也包括復雜一點的調整比如對shuffle方式和數據編排方式重組。除此以外，數據的不同特點也會帶來邏輯執行圖調整的需求：對于邏輯圖的動態調整，在分布式作業處理中是一個全新的方向，也是我們在DAG 2.0里面探索的新式解決方案。

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

• 當M1輸出的數據量較小時，允許其輸出被全量載入下游單個計算節點的內存，DAG就會選擇優化的map join(plan A)，來避免額外的shuffle和排序。
• 當M1輸出的數據量大到一定程度，已經不屬于map join的適用范圍，DAG就可以自動選擇走merge join，來保證作業的成功執行。

除了map join這個典型場景外，借助DAG2.0的動態調度能力，MaxCompute在解決其他用戶痛點上也做了很多探索，并取得了不錯的效果。例如智能動態并發度調整：在執行過程中依據分區數據統計調整，動態調整并發度；自動合并小分區，避免不必要的資源使用，節約用戶資源使用；切分大分區，避免不必要的長尾出現等等。

統一的AM/DAG執行框架

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

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

• 離線作業：每個節點按需去申請資源，一個邏輯節點代表一個調度單位；節點間連接邊上傳輸的數據，通過落盤的方式來保證可靠性；
• 準實時作業：整個作業的所有節點都統一在一個調度單位內進行gang scheduling；節點間連接邊上通過網絡/內存直連傳輸數據，并利用數據pipeline來追求最優的性能。

在此統一離線作業與準實時作業的到一套架構的基礎上，這種統一的描述方式，使得探索離線作業高資源利用率，以及準實時作業的高性能之間的tradeoff成為可能：當調度單位可以自由調整，就可以實現一種全新的混合的計算模式，我們稱之為Bubble執行模式。

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

4.1.3 效果

DAG2.0的動態性使得很多執行優化可以運行時決定，使得實際執行的效果更優。例如，在阿里內部的作業中，動態的conditional join相比靜態的執行計劃，整體獲得了將近3X的性能提升。

混合Bubble執行模式平衡了離線作業高資源利用率以及準實時作業的高性能，這在1TB TPCH測試集上有顯著的體現，

• Bubble相對離線作業：在多使用20%資源的情況下，Bubble模式性能提升將近一倍；
• Bubble相對準實時模式：在節省了2.6X資源情況下， Bubble性能僅下降15%;

4.2 Fuxi Shuffle 2.0 - 磁盤內存網絡的最佳使用

4.2.1 背景

大數據計算作業中，節點間的數據傳遞稱為shuffle, 主流分布式計算系統都提供了數據shuffle服務的子系統。如前述DAG計算模型中，task間的上下游數據傳輸就是典型的shuffle過程。

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

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

挑戰

我們先以使用最廣泛的，基于磁盤文件系統的離線作業shuffle為例。

通常每個mapper生成一個磁盤文件，包含了這個mapper寫給下游所有reducer的數據。而一個reducer要從所有mapper所寫的文件中，讀取到屬于自己的那一小塊。右側則是一個系統中典型規模的MR作業，當每個mapper處理256MB數據，而下游reducer有10000個時，平均每個reducer讀取來自每個mapper的數據量就是25.6KB, 在機械硬盤HDD為介質的存儲系統中，屬于典型的讀碎片現象，因為假設我們的磁盤iops能達到1000, 對應的throughput也只有25MB/s, 嚴重影響性能和磁盤壓力。

【基于文件系統shuffle的示意圖 / 一個20000*10000的MR作業的碎片讀】

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

一個線上的例子是，某些主流集群單次讀請求size為50-100KB, Disk util指標長期維持在90%的警戒線上。這些限制了對作業規模的進一步追求。

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

4.2.2 Fuxi的答案：Fuxi Shuffle 2.0

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

為了針對性地解決上述碎片讀問題及其引發的一連串負面效應，我們全新打造了基于shuffle service的shuffle模式。Shuffle service的最基本工作方式是，在集群每臺機器部署一個shuffle
agent節點，用來歸集寫給同一reducer的shuffle數據。如下圖

可以看到，mapper生成shuffle數據的過程變為mapper將shuffle數據通過網絡傳輸給每個reducer對應的shuffle agent, 而shuffle agent歸集一個reducer來自所有mapper的數據，并追加到shuffle磁盤文件中，兩個過程是流水線并行化起來的。

Shuffle agent的歸集功能將reducer的input數據從碎片變為了連續數據文件，對HDD介質相當友好。由此，整個shuffle過程中對磁盤的讀寫均為連續訪問。從標準的TPCH等測試中可以看到不同場景下性能可取得百分之幾十到幾倍的提升，且大幅降低磁盤壓力、提升CPU等資源利用率。

Shuffle Service的容錯機制

Shuffle service的歸集思想在公司內外都有不同的工作展現類似的思想，但都限于“跑分”和小范圍使用。因為這種模式對于各環節的錯誤天生處理困難。

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

我們設計了數據雙副本機制解決了這個問題，使得大多數通常情況下reducer可以讀取到高效的agent生成的數據，而當少數agent數據丟失的情況，可以讀取備份數據，備份數據的重新生成只依賴特定的上游mapper.

具體來說，mapper產生的每份shuffle數據除了發送給對于shuffle agent外，也會按照與傳統文件系統shuffle數據類似的格式，在本地寫一個備份。按前面所述，這份數據寫的代價較小但讀取的性能不佳，但由于僅在shuffle agent那個副本出錯時才會讀到備份數據，所以對作業整體性能影響很小，也不會引起集群級別的磁盤壓力升高。

有效的容錯機制使得shuffle service相對于文件系統shuffle，在提供更好的作業性能的同時，因shuffle數據出錯的task重試比例降低了一個數量級，給線上全面投入使用打好了穩定性基礎。

線上生產環境的極致性能穩定性

在前述基礎功能之上，Fuxi線上的shuffle系統應用了更多功能和優化，在性能、成本、穩定性等方便取得了進一步的提升。舉例如下。

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


2. 故障shuffle
agent的切換
各種軟硬件故障導致shuffle agent對某個reducer的數據工作不正常時，后續數據可以實時切換到其他正常shuffle agent. 這樣，就會有更多的數據可以從shuffle agent側讀到，而減少低效的備份副本訪問。


3. Shuffle agent數據的回追
很多時候發生shuffle
agent切換時（如機器下線），原shuffle agent生成的數據可能已經丟失或訪問不到。在后續數據發送到新的shuffle agent同時，Fuxi還會將丟失的部分數據從備份副本中load起來并同樣發送給新的shuffle agent, 使得后續reducer所有的數據都可以讀取自shuffle agent側，極大地提升了容錯情況下的作業性能。


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

我們在生產環境對部分作業應用了一種新的shuffle模型，這種模型中mapper的發送端和reducer的接收端都通過一個agent節點來中轉shuffle流量。線上已經有部分作業使用此種方式并在性能上得到了進一步的提升。

內存數據shuffle

離線大數據作業可能承擔了主要的計算數據量，但流行的大數據計算系統中有非常多的場景是通過實時/準實時方式運行的，作業全程的數據流動發生在網絡和內存，從而在有限的作業規模下取得極致的運行性能，如大家熟悉的Spark, Flink等系統。

Fuxi DAG也提供了實時/準實時作業運行環境，傳統的shuffle方式是通過網絡直連，也能收到明顯優于離線shuffle的性能。這種方式下，要求作業中所有節點都要調度起來才能開始運行，限制了作業的規模。而實際上多數場景計算邏輯生成shuffle數據的速度不足以填滿shuffle帶寬，運行中的計算節點等待數據的現象明顯，性能提升付出了資源浪費的代價。

我們將shuffle service應用到內存存儲中，以替換network傳輸的shuffle方式。一方面，這種模式解耦了上下游調度，整個作業不再需要全部節點同時拉起；另一方面通過精確預測數據的讀寫速度并適時調度下游節點，可以取得與network傳輸shuffle相當的作業性能，而資源消耗降低50%以上。這種shuffle方式還使得DAG系統中多種運行時調整DAG的能力可以應用到實時/準實時作業中。

4.2.3 收益

Fuxi Shuffle 2.0全面上線生產集群，處理同樣數據量的作業資源比原來節省15%，僅shuffle方式的變化就使得磁盤壓力降低23%，作業運行中發生錯誤重試的比例降至原來的5%。

【線上典型集群的性能與穩定性提升示意圖（不同組數據表示不同集群）】

對使用內存shuffle的準實時作業，我們在TPCH等標準測試集中與網絡shuffle性能相當，資源使用只有原來的30%左右，且支持了更大的作業規模，和DAG 2.0系統更多的動態調度功能應用至準實時作業。

5. 單機調度

大量分布式作業匯集到一臺機器上，如何將單機有限的各種資源合理分配給每個作業使用，從而達到作業運行質量、資源利用率、作業穩定性的多重保障，是單機調度要解決的任務。

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

一方面是在線業務資源利用率不足，另一方面是離線計算長期超負載運行，那么能否將在線業務與離線計算進行混合部署，提升資源利用率同時大幅降低成本，實現共贏。

5.1 三大挑戰

如何保障在線服務質量
在線集群的平均CPU利用率只有10%左右，混部的目標就是將剩余的資源提供給MaxCompute進行離線計算使用，從而達到節約成本的目的。那么，如何能夠保障資源利用率提升的同時又能夠保護在線服務不受影響呢？

如何保障離線穩定
當資源發生沖突時，第一反應往往是保護在線，犧牲離線。畢竟登不上淘寶天貓下不了單可是大故障。可是，離線如果無限制的犧牲下去，服務質量將會出現大幅度下降。試想，我在dataworks上跑個SQL，之前一分鐘就出結果，現在十幾分鐘甚至一個小時都跑不出來，大數據分析的同學估計也受不了了。

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

5.2 資源隔離分級管理

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

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

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

CPU分級管理

通過精細的組合多種內核策略，將CPU區分為高、中、低三類優先級

隔離策略如下圖所示

基于不同類型的資源對應不同的優先級作業

5.3 資源畫像

Fuxi作為資源調度模塊，對資源使用情況的精準畫像是衡量資源分配，調查/分析/解決解決資源問題的關鍵。針對在線作業的資源情況，集團和業界都有較多的解決方案。這類通用的資源采集角色存在以下無法解決的問題無法應用于離線作業資源畫像的數據采集階段

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

為此，我們提出了FuxiSensor的資源畫像方案，架構如上圖所示，同時利用SLS進行數據的收集和分析。在集群、Job作業、機器、worker等不同層次和粒度實現了資源信息的畫像，實現了秒級的數據采集精度。在混部及MaxCompute的實踐中，成為資源問題監控、報警、穩定性數據分析、作業異常診斷、資源監控狀況的統一入口，成為混部成功的關鍵指標。

5.4 線上效果

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

在線抖動小于5%

5.5 單機調度小結

為了解決三大挑戰，通過完善的各維度優先級隔離策略，將在線提升到高優先級資源維度，我們保障了在線的服務質量穩定；通過離線內部優先級區分及各種管理策略，實現了離線質量的穩定性保障；通過細粒度資源畫像信息，實現了資源使用的評估與分析，最終實現了混部在阿里的大規模推廣與應用，從而大量提升了集群資源利用率，為離線計算節省了大量成本。

6. 展望

從2009到2019年歷經十年的錘煉，伏羲系統仍然在不斷的演化，滿足不斷涌現的業務新需求，引領分布式調度技術的發展。接下來，我們會從以下幾個方面繼續創新：

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

最后，我們熱忱歡迎集團各個團隊一起交流探討，共同打造世界一流的分布式調度系統！

MaxCompute產品官網?https://www.aliyun.com/product/odps
更多阿里巴巴大數據計算技術交流，歡迎掃碼加入“MaxCompute開發者社區”釘釘群。

原文鏈接
本文為云棲社區原創內容，未經允許不得轉載。

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。