面向大數據與雲計算調度挑戰的阿里經濟體核心調度系統

編者按 伏羲（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, userid FROM order_data WHERE spend > 1000 ) a JOIN ( SELECT userid, age FROM user WHERE 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等，這類業務的特色是對響應延遲特別敏感，一旦服務抖動將會出現添加購物車失敗、下單失敗、瀏覽卡頓、釘釘消息發送失敗等各類異常狀況，嚴重影響用戶體驗，同時爲了應對在61八、雙11等各類大促的狀況，須要提早準備大量的機器。因爲以上種種緣由，平常狀態這些機器的資源利用率不足10%，產生資源浪費的狀況。與此同時，阿里的離線業務又是另一幅風景，MaxCompute計算平臺承擔了阿里全部大數據離線計算業務類型，各個集羣資源利用率常態超負載運行，數據量和計算量每一年都在保持高速增加。 一方面是在線業務資源利用率不足，另外一方面是離線計算長期超負載運行，那麼可否將在線業務與離線計算進行混合部署，提高資源利用率同時大幅下降成本，實現雙贏。 5.1 三大挑戰

1. 如何保障在線服務質量 在線集羣的平均CPU利用率只有10%左右，混部的目標就是將剩餘的資源提供給MaxCompute進行離線計算使用，從而達到節約成本的目的。那麼，如何可以保障資源利用率提高的同時又可以保護在線服務不受影響呢？
2. 如何保障離線穩定 當資源發生衝突時，第一反應每每是保護在線，犧牲離線。畢竟登不上淘寶天貓下不了單但是大故障。但是，離線若是無限制的犧牲下去，服務質量將會出現大幅度降低。試想，我在dataworks上跑個SQL，以前一分鐘就出結果，如今十幾分鍾甚至一個小時都跑不出來，大數據分析的同窗估計也受不了了。
3. 如何衡量資源質量 電商業務經過富容器的方式集成多種容器粒度的分析手段，可是前文描述過離線做業的特色，如何可以精準的對離線做業資源使用進行資源畫像分析，若是可以評估資源受干擾的程度，混部集羣的穩定性等問題，是對咱們的又一個必需要解決的挑戰

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開發者社區」釘釘羣。