[[326482]]

作為阿里巴巴核心大數(shù)據(jù)底座——伏羲調(diào)度和分布式執(zhí)行系統(tǒng)，支撐著阿里集團內(nèi)部以及阿里云上大數(shù)據(jù)平臺絕大部分的大數(shù)據(jù)計算需求，在其上運行的 MaxCompute(ODPS) 以及 PAI 等多種計算引擎，每天為用戶進行海量的數(shù)據(jù)運算。為了支撐計算平臺下個 10 年的發(fā)展，伏羲團隊啟動了 DAG 2.0 項目，從代碼和功能方面實現(xiàn)完全的升級換代，支持更多 DAG 執(zhí)行過程中的動態(tài)性及計算模式。本文將分享 DAG 2.0 核心架構(gòu)及整體設計，以及與上層各個計算引擎的對接，較長，同學們可收藏后再看。

前言

在"阿里體量"的大數(shù)據(jù)生態(tài)中，伏羲系統(tǒng)管理著彈內(nèi)外多個物理集群，超十萬臺物理機， 以及數(shù)百萬的 CPU/GPU cores。每天運行在伏羲分布式平臺上的作業(yè)數(shù)已經(jīng)超過千萬, 是業(yè)界少有的，單天處理 EB 級別數(shù)據(jù)分布式平臺。其中單個作業(yè)規(guī)模已經(jīng)高達數(shù)十萬計算節(jié)點，管理著數(shù)百億的邊連接。在過去的十年中，阿里集團以及阿里云上這樣的作業(yè)數(shù)目和規(guī)模，錘煉了伏羲分布式平臺;與此同時，今天平臺上作業(yè)的日益多樣化，以及向前再發(fā)展的需求，對于伏羲系統(tǒng)架構(gòu)的進一步演化，也都帶來了巨大挑戰(zhàn)與機遇。

一 背景

1 伏羲 DAG/AM 組件

從較高的層面來看整個分布式系統(tǒng)的體系架構(gòu)，物理集群之上運行的分布式系統(tǒng)，大概可以分成資源管理，作業(yè)分布式調(diào)度執(zhí)行，與多個計算節(jié)點的運行這三個層次，如同下圖所示。通常所說的 DAG 組件，指的是每個分布式作業(yè)的中心管理點，也就是 application master (AM)。AM 之所以經(jīng)常被稱為 DAG (Directional Acyclic Graph， 有向無環(huán)圖) 組件，是因為 AM 最重要的責任，就是負責協(xié)調(diào)分布式作業(yè)的執(zhí)行。而現(xiàn)代的分布式系統(tǒng)中的作業(yè)執(zhí)行流程，通常可以通過 DAG 上面的調(diào)度以及數(shù)據(jù)流來描述[1]。相對于傳統(tǒng)的 Map-Reduce[2] 執(zhí)行模式， DAG 的模型能對分布式作業(yè)做更精準的描述，也是當今各種主流大數(shù)據(jù)系統(tǒng) (Hadoop 2.0+, SPARK, FLINK, TENSORFLOW 等) 的設計架構(gòu)基礎，區(qū)別只在于 DAG 的語義是透露給終端用戶，還是計算引擎開發(fā)者。

與此同時，從整個分布式系統(tǒng) stack 來看， AM 肩負著除了運行 DAG 以外更多的責任。作為作業(yè)的中心管控節(jié)點，向下其負責與 Resource Manager 之間的交互，為分布式作業(yè)申請計算資源;向上其負責與計算引擎進行交互，并將收集的信息反饋到 DAG 的執(zhí)行過程中。作為唯一有能力對每一個分布式作業(yè)的執(zhí)行大局有最精準的了解的組件，在全局上對 DAG 的運行做準確的管控和調(diào)整，也是 AM 的重要職責。從上圖描述的分布式系統(tǒng) stack 圖中，我們也可以很直觀的看出，AM 是系統(tǒng)中唯一需要和幾乎所有分布式組件交互的組件，在作業(yè)的運行中起了重要的承上啟下的作用。這一組件之前在伏羲系統(tǒng)中被稱為 JobMaster (JM), 在本文中我們統(tǒng)一用 DAG 或者 AM 來指代。

2 邏輯圖與物理圖

分布式作業(yè)的 DAG，有兩種層面上的表述：邏輯圖與物理圖。簡單地來說 (over-simplified)，終端用戶平時理解的 DAG 拓撲，大多數(shù)情況下描述的是邏輯圖范疇：比如大家平時看到的 logview 圖，雖然里面包含了一些物理信息(每個邏輯節(jié)點的并發(fā)度)，但整體上可以認為描述的就是作業(yè)執(zhí)行流程的邏輯圖。

準確一點說：

• 邏輯圖描述了用戶想要實現(xiàn)的數(shù)據(jù)處理流程，從數(shù)據(jù)庫/SQL 的角度(其他類型引擎也都有類似之處，比如 TENSORFLOW) 來看，可以大體認為 DAG 的邏輯圖，是對優(yōu)化器執(zhí)行計劃的一個延續(xù)。
• 物理圖更多描述了執(zhí)行計劃映射到物理分布式集群的具體描述，體現(xiàn)的是執(zhí)行計劃被物化到分布式系統(tǒng)上，具備的一些特性：比如并發(fā)度，數(shù)據(jù)傳輸方式等等。

而每個邏輯圖的"物理化"，可以有很多等效方式。選擇合適的方式來將邏輯圖變成物理化執(zhí)行，并進行靈活的調(diào)整，是 DAG 組件的重要職責之一。從上圖的邏輯圖到物理圖的映射可以看到，一個圖的物理化過程，實際上就是在回答一系列圖節(jié)點以及各個連接邊物理特性的問題，一旦這些問題得到確認，就能得到在分布式系統(tǒng)上實際執(zhí)行物理圖。

3 為什么需要 DAG 2.0 架構(gòu)升級?

作為從阿里云飛天系統(tǒng)創(chuàng)建伊始就開始研發(fā)的伏羲分布式作業(yè)執(zhí)行框架，DAG 1.0 在過去十年中支撐了阿里集團的大數(shù)據(jù)業(yè)務，在系統(tǒng)規(guī)模以及可靠性等方面都走在了業(yè)界領先。另外一方面，作為一個開發(fā)了十年的系統(tǒng)，雖然在這個期間不斷的演進，DAG 1.0 在基本架構(gòu)上秉承了比較明顯的 Map-Reduce 執(zhí)行框架的一些特點，邏輯圖和物理圖之間沒有清晰的分層，這導致在這個基本架構(gòu)上要繼續(xù)向前走，支持更多 DAG 執(zhí)行過程中的動態(tài)性，以及同時支持多種計算模式等方面，都比較困難。事實上今天在 MaxCompute SQL 線上，離線作業(yè)模式以及準實時作業(yè)模式 (smode) 兩種執(zhí)行模式，使用了兩套完全分開的分布式執(zhí)行框架，這也導致對于優(yōu)化性能和優(yōu)化系統(tǒng)資源使用之間的取舍，很多情況下只能走兩個極端，而無法比較好的 tradeoff。

除此之外，隨著 MaxCompute 以及 PAI 引擎的更新?lián)Q代以及新功能演進，上層的分布式計算自身能力在不斷的增強。對于 AM 組件在作業(yè)管理，DAG 執(zhí)行等方面的動態(tài)性，靈活性等方面的需求也日益強烈。在這樣的一個大的背景下，為了支撐計算平臺下個 10 年的發(fā)展，伏羲團隊啟動了 DAG 2.0 的項目，將從代碼和功能方面，完整替代 1.0 的 JobMaster 組件，實現(xiàn)完全的升級換代。在更好的支撐上層計算需求的同時，也同時對接伏羲團隊在 shuffle 服務 (shuffle service) 上的升級，以及 fuxi master (Resource Manager) 的功能升級。與此同時，站在提供企業(yè)化服務的角度來看，一個好的分布式執(zhí)行框架，除了支持阿里內(nèi)部極致的大規(guī)模大吞吐作業(yè)之外，我們需要支持計算平臺的向外走，支持云上各種規(guī)模和計算模式的需求。除了繼續(xù)錘煉超大規(guī)模的系統(tǒng)擴展能力以外，我們需要降低大數(shù)據(jù)系統(tǒng)使用的門檻，通過系統(tǒng)本身的智能動態(tài)化能力，來提供自適應(各種數(shù)據(jù)規(guī)模以及處理模式)的大數(shù)據(jù)企業(yè)界服務，是 DAG 2.0 在設計架構(gòu)中考慮的另一重要維度。

二 DAG 2.0 架構(gòu)以及整體設計

DAG 2.0 項目，在調(diào)研了業(yè)界各個分布式系統(tǒng)(包括SPARK/FLINK/Dryad/Tez/Tensorlow)DAG 組件之后，參考了 Dryad/Tez 的框架。新一代的架構(gòu)上，通過邏輯圖和物理圖的清晰分層，可擴展的狀態(tài)機管理，插件式的系統(tǒng)管理，以及基于事件驅(qū)動的調(diào)度策略等基座設計，實現(xiàn)了對計算平臺上多種計算模式的統(tǒng)一管理，并更好的提供了作業(yè)執(zhí)行過程中在不同層面上的動態(tài)調(diào)整能力。

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

傳統(tǒng)的分布式作業(yè)執(zhí)行流程，作業(yè)的執(zhí)行計劃是在提交之前確定的。以 SQL 執(zhí)行為例，一個 SQL 語句，在經(jīng)過編譯器和優(yōu)化器后產(chǎn)生執(zhí)行圖，并被轉(zhuǎn)換成分布式系統(tǒng)(伏羲)的執(zhí)行計劃。

這個作業(yè)流程在大數(shù)據(jù)系統(tǒng)中是比較標準的操作。然而在具體實現(xiàn)中，如果在 DAG 的執(zhí)行缺乏自適應動態(tài)調(diào)整能力的話，整個執(zhí)行計劃都需要事先確定，會使得作業(yè)的運行沒有太多動態(tài)調(diào)整的空間。放在 DAG 的邏輯圖與物理圖的背景中來說，這要求框架在運行作業(yè)前，必須事先了解作業(yè)邏輯和處理數(shù)據(jù)各種特性，并能夠準確回答作業(yè)運行過程，各個節(jié)點和連接邊的物理特性問題，來實現(xiàn)邏輯圖往物理圖的轉(zhuǎn)換。

然而在現(xiàn)實情況中，許多物理特性相關的問題，在作業(yè)運行前是無法被感知的。以數(shù)據(jù)特性為例，一個分布式作業(yè)在運行前，能夠獲得的只有原始輸入的一些特性(數(shù)據(jù)量等)， 對于一個較深的 DAG 執(zhí)行而言，這也就意味著只有根節(jié)點的物理計劃(并發(fā)度選擇等) 是相對合理的，而下游的節(jié)點和邊的物理特性只能通過一些特定的規(guī)則來猜測。雖然在輸入數(shù)據(jù)有豐富的 statistics 的前提下，優(yōu)化器有可能可以將這些 statistics，與執(zhí)行 plan 中的各個 operator 特性結(jié)合起來，進行一些適度的演算：從而推斷在整個執(zhí)行流程中，每一步產(chǎn)生的中間數(shù)據(jù)可能符合什么樣的特性。但這種推斷在實現(xiàn)上，尤其在面對阿里大體量的實際生產(chǎn)環(huán)境中，面臨著巨大的挑戰(zhàn)，例如：

實際輸入數(shù)據(jù)的 statistics 的缺失

即便是 SQL 作業(yè)處理的結(jié)構(gòu)化數(shù)據(jù)，也無法保證其源表數(shù)據(jù)特性擁有很好的統(tǒng)計。事實上今天因為數(shù)據(jù)落盤方式多樣化，以及精細化統(tǒng)計方式的缺失，大部分的源表數(shù)據(jù)都是沒有完整的 statistics 的。此外對于集群內(nèi)部和外部需要處理的非結(jié)構(gòu)化數(shù)據(jù)，數(shù)據(jù)的特性的統(tǒng)計更加困難。

分布式作業(yè)中存在的大量用戶邏輯黑盒

作為一個通用的大數(shù)據(jù)處理系統(tǒng)，不可避免的需要支持用戶邏輯在系統(tǒng)中的運行。比如 SQL 中常用的 UDF/UDTF/UDJ/Extractor/Outputer 等等，這些使用 Java/Python 實現(xiàn)的用戶邏輯，計算引擎和分布式系統(tǒng)并無法理解，在整個作業(yè)流程中是類似黑盒的存在。以 MaxCompute 為例，線上有超過 20% 的 SQL 作業(yè)，尤其是重點基線作業(yè)，都包含用戶代碼。這些大量用戶代碼的存在，也造成了優(yōu)化器在很多情況下無法對中間產(chǎn)出數(shù)據(jù)的特性進行預判。

優(yōu)化器預判錯誤代價昂貴

在優(yōu)化器選擇執(zhí)行計劃時，會有一些優(yōu)化方法，在數(shù)據(jù)符合一定特殊特性的時候，被合理選中能帶來性能優(yōu)化。但是一旦選擇的前提假設錯誤(比如數(shù)據(jù)特性不符合預期)，會適得其反，甚至帶來嚴重的性能回退或作業(yè)失敗。在這種前提下，依據(jù)靜態(tài)的信息實現(xiàn)進行過多的預測經(jīng)常得不到理想的結(jié)果。

這種種原因造成的作業(yè)運行過程中的非確定性，要求一個好的分布式作業(yè)執(zhí)行系統(tǒng)，需要能夠根據(jù)中間運行結(jié)果的特點，來進行執(zhí)行過程中的動態(tài)調(diào)整。因為只有在中間數(shù)據(jù)已經(jīng)在執(zhí)行過程中產(chǎn)生后，其數(shù)據(jù)特性才能被最準確的獲得，動態(tài)性的缺失，可能帶來一系列的線上問題，比如：

• 物理資源的浪費：比如計算節(jié)點事先選擇的資源類型的不合理，或者大量的計算被消耗用于處理后繼會被丟棄的無效數(shù)據(jù)。
• 作業(yè)的嚴重長尾：比如中間數(shù)據(jù)分布傾斜或不合理編排，導致一個 stage 上計算節(jié)點需要處理的數(shù)據(jù)量極端化。
• 作業(yè)的不穩(wěn)定：比如由于優(yōu)化器靜態(tài)計劃的錯判，導致不合理的執(zhí)行計劃無法完成。

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

點，邊，圖的清晰物理邏輯分層，和基于事件的數(shù)據(jù)收集和調(diào)度管理，以及插件式的功能實現(xiàn)，方便了 DAG 2.0 在運行期間的數(shù)據(jù)收集，以及使用這些數(shù)據(jù)來系統(tǒng)性地回答，邏輯圖向物理圖轉(zhuǎn)化過程中需要確定的問題。從而在必要的時候?qū)崿F(xiàn)物理圖和邏輯圖的雙重動態(tài)性，對執(zhí)行計劃進行合理的調(diào)整。在下文中提到幾個落地場景中，我們會進一步舉例說明基于 2.0 的這種強動態(tài)性能力，實現(xiàn)更加自適應，更加高效的分布式作業(yè)的執(zhí)行。

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

DAG 2.0 抽象分層的點，邊，圖架構(gòu)上，也使其能通過對點和邊上不同物理特性的描述，對接不同的計算模式。業(yè)界各種分布式數(shù)據(jù)處理引擎，包括 SPARK, FLINK, HIVE, SCOPE, TENSORFLOW 等等，其分布式執(zhí)行框架的本源都可以歸結(jié)于 Dryad[1] 提出的 DAG 模型。我們認為對于圖的抽象分層描述，將允許在同一個 DAG 系統(tǒng)中，對于離線/實時/流/漸進計算等多種模型都可以有一個好的描述。在 DAG 2.0 初步落地的過程中，首要目標是在同一套代碼和架構(gòu)系統(tǒng)上，統(tǒng)一當前伏羲平臺上運行的幾種計算模式，包括 MaxCompute 的離線作業(yè)，準實時作業(yè)，以及 PAI 平臺上的 Tensorflow 作業(yè)和其他的非 SQL 類作業(yè)。對更多新穎計算模式的探索，也會有計劃的分步驟進行。

1)統(tǒng)一的離線作業(yè)與準實時作業(yè)執(zhí)行框架

首先我們來看平臺上作業(yè)數(shù)占到絕大多數(shù)的 SQL 線離線作業(yè) (batch job) 與準實時作業(yè) (smode)。前面提到過，由于種種歷史原因，之前 MaxCompompute SQL 線的這兩種模式的資源管理和作業(yè)執(zhí)行，是搭建在兩套完全分開的代碼實現(xiàn)上的。這除了導致兩套代碼和功能無法復用以外，兩種計算模式的非黑即白，使得彼此在資源利用率和執(zhí)行性能之間無法 tradeoff。而在 2.0 的 DAG 模型上，我們實現(xiàn)了這兩種計算模式比較自然的融合和統(tǒng)一，如下圖所示：

在通過對邏輯節(jié)點和邏輯邊上映射不同的物理特性，離線作業(yè)和準實時作業(yè)都能得到準確的描述:

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

今天在線上，離線模式因為其 on-demand 的資源申請以及中間數(shù)據(jù)落盤等特點，作業(yè)在資源利用率，規(guī)模性和穩(wěn)定性方面都有明顯的優(yōu)勢。而準實時模式則通過常駐的計算資源池以及 gang scheduling 這種 greedy 資源申請，降低了作業(yè)運行過程中的 overhead，并使得數(shù)據(jù)的 pipelined 傳輸處理成為可能，達到加速作業(yè)運行的效果，但其資源使用的特點，也使其無法在廣泛范圍內(nèi)來支持大規(guī)模作業(yè)。DAG 2.0 的升級，不僅在同一套架構(gòu)上統(tǒng)一了這兩種計算模式，更重要的是這種統(tǒng)一的描述方式，使得探索離線作業(yè)高資源利用率，以及準實時作業(yè)的高性能之間的 tradeoff 成為可能：當調(diào)度單位可以自由調(diào)整，就可以實現(xiàn)一種全新的混合的計算模式，我們稱之為 Bubble 執(zhí)行模式。

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

Greedy Bubble

在可用的資源(集群規(guī)模，quota 等)受限，一個大規(guī)模作業(yè)無法實現(xiàn) gang scheduling 時，如果用戶對資源利用率不敏感，唯一的目標是盡快跑完一個大規(guī)模作業(yè)。這種情況下，可以實現(xiàn)基于可用計算節(jié)點數(shù)目，實施 greedy 的 bubble 切割的策略, 盡量切出大的 bubble。

Efficient Bubble

在作業(yè)的運行過程中，節(jié)點間的運算可能存在天然的 barrier (比如 sort 運算， 建 hash 表等等)。如果把兩個通過 barrier 邊連接的節(jié)點切到一個 bubble 中，雖然作業(yè) e2e 性能上還是會有調(diào)度 overhead 降低等帶來的提升，但是因為數(shù)據(jù)無法完全 pipeline 起來，資源的利用率達不到最高。那么在對資源的利用率較為敏感時，可以避免 bubble 內(nèi)部出現(xiàn) barrier 邊。這同樣是計算引擎可以根據(jù)執(zhí)行計劃做出決定的。

這里只列舉了兩個簡單的策略，其中還有更多可以細化以及針對性優(yōu)化的地方。在不同的場景上，通過 DAG 層面提供的這種靈活按照 bubble 執(zhí)行計算的能力，允許上層計算可以在不同場景上挑選合適的策略，更好的支持各種不同計算的需求。

2)支持新型計算模式的描述

1.0 的執(zhí)行框架的底層設計受 Map-Reduce 模式的影響較深，節(jié)點之間的邊連接，同時混合了調(diào)度順序，運行順序，以及數(shù)據(jù)流動的多種語義。通過一條邊連接的兩個節(jié)點，下游節(jié)點必須在上游節(jié)點運行結(jié)束，退出，并產(chǎn)生數(shù)據(jù)后才能被調(diào)度。這種描述對于新型的一些計算模式并不適用。比如對于 Parameter Server 計算模式，Parameter Server(PS) 與 Worker 在運行過程中有如下特點：

• PS 作為 parameter 的 serving entity, 可以獨立運行。
• Worker 作為 parameter 的 consumer 和 updater, 需要 PS 在運行后才能有效的運行，并且在運行過程中需要和 PS 持續(xù)的進行數(shù)據(jù)交互。

這種運行模式下，PS 和 worker 之間天然存在著調(diào)度上的前后依賴關系。但是因為 PS 與 worker 必須同時運行，不存在 PS 先退出 worker 才調(diào)度的邏輯。所以在 1.0 框架上， PS 與 worker 只能作為兩個孤立無聯(lián)系的 stage 來分開調(diào)度和運行。此外所有 PS 與 worker 之間，也只能完全通過計算節(jié)點間直連通訊，以及在外部 entity (比如 zookeeper 或 nuwa) 協(xié)助來進行溝通與協(xié)調(diào)。這導致 AM/DAG 作為中心管理節(jié)點作用的缺失，作業(yè)的管理基本被下放計算引擎上，由計算節(jié)點之間自行試圖協(xié)調(diào)來完成。這種無中心化的管理，對稍微復雜的情況下 (failover 等) 無法很好的處理。

在 DAG 2.0 的框架上，為了更準確的描述節(jié)點之間的調(diào)度和運行關系，引入并且實現(xiàn)了 concurrent edge 的概念：通過 concurrent edge 連接的上下游節(jié)點，在調(diào)度上存在先后，但是可以同時運行。而調(diào)度的時機也可以靈活配置：可以上下游同步調(diào)度，也可以在上游運行到一定程度后，通過事件來觸發(fā)下游的調(diào)度。在這種靈活的描述能力上， PS 作業(yè)可以通過如下這種 DAG 來描述，這不僅使得作業(yè)節(jié)點間的關系描述更加準確，而且使得 AM 能夠理解作業(yè)的拓撲，進行更加有效的作業(yè)管理，包括在不同計算節(jié)點發(fā)生 failover 時不同的處理策略等。

此外，DAG 2.0 新的描述模型，也允許 PAI 平臺上的 Tensorflow/PS 作業(yè)實現(xiàn)更多的動態(tài)優(yōu)化，并進行新的創(chuàng)新性工作。在上圖的 dynamic PS DAG 中，就引進了一個額外的 control 節(jié)點，這一節(jié)點可以在作業(yè)運行過程中(包括 PS workload 運行之前和之后)，對作業(yè)的資源申請，并發(fā)度等進行動態(tài)的調(diào)整，確保作業(yè)的優(yōu)化執(zhí)行。

事實上 concurrent edge 這個概念，描述的是上下游節(jié)點運行/調(diào)度時機的物理特性，也是我們在清晰的邏輯物理分層的架構(gòu)上實現(xiàn)的一個重要擴展。不僅對于 PS 作業(yè)模式，在之前描述過的對于通過 bubble 來統(tǒng)一離線與準實時作業(yè)計算模式，這個概念也有重要的作用。

三 DAG 2.0 與上層計算引擎的集成

DAG 2.0 作為計算平臺的分布式運行基座，它的升級換代，為上層的各種計算引擎提供了更多靈活高效的執(zhí)行能力，而這些能力的落地，需要通過與具體計算場景的緊密結(jié)合來實現(xiàn)。接下來通過 2.0 與上層各個計算引擎(包括 MaxCompute 以及 PAI 平臺等)的一些對接場景，具體舉例說明 2.0 新的調(diào)度執(zhí)行框架，如何賦能平臺上層的計算與應用。

1 運行過程中的 DAG 動態(tài)調(diào)整

作為計算平臺上的作業(yè)大戶，MaxCompute 平臺上多種多樣的計算場景，尤其是離線作業(yè)中的各種復雜邏輯，為動態(tài)圖能力的落地提供了豐富多樣的場景，這里從動態(tài)物理圖和邏輯圖幾個方面討論幾個例子。

1)動態(tài)并發(fā)度調(diào)整

基于作業(yè)運行期間中間數(shù)據(jù)大小進行動態(tài)并發(fā)度調(diào)整，是 DAG 動態(tài)調(diào)整中最基本的能力。以傳統(tǒng) MR 作業(yè)為例，對于一個靜態(tài) MR 作業(yè)而言，能根據(jù)讀取數(shù)據(jù)量來比較準確判斷 Mapper 的并發(fā)，但是對于 Reducer 的并發(fā)只能簡單推測，比如下圖中對于處理 1TB 的 MR 作業(yè)而言，提交作業(yè)時，只能根據(jù) Mapper 1000 并發(fā)，來猜測給出 500 的 Reducer 并發(fā)度，而如果數(shù)據(jù)在 Mapper 經(jīng)過大量過濾導致最終只產(chǎn)出 10MB 中間數(shù)據(jù)時，500 并發(fā)度 Redcuer 顯然是非常浪費的，動態(tài)的 DAG 必須能夠根據(jù)實際的 Mapper 產(chǎn)出來進行 Reducer 并發(fā)調(diào)整(500 -> 1)。

而實際實現(xiàn)中，最簡單的動態(tài)調(diào)整，會直接按照并發(fā)度調(diào)整比例來聚合上游輸出的 partition 數(shù)據(jù)，如下圖這個并發(fā)度從 10 調(diào)整到 5 的例子所示，在調(diào)整的過程中，可能產(chǎn)生不必要的數(shù)據(jù)傾斜。

DAG 2.0 基于中間數(shù)據(jù)的動態(tài)并發(fā)調(diào)整實現(xiàn)，充分考慮了數(shù)據(jù) partition 可能存在傾斜的情況，對動態(tài)調(diào)整的策略進行了優(yōu)化，使得動態(tài)調(diào)整的策略后數(shù)據(jù)的分布更加均勻，可以有效避免由于動態(tài)調(diào)整可能引入的數(shù)據(jù)傾斜。

這種最常見下游并發(fā)調(diào)整方式是 DAG 2.0 動態(tài)物理圖能力的一個直觀展示。在 2.0 中項目中，結(jié)合計算引擎的數(shù)據(jù)處理的特點，還探索了基于源數(shù)據(jù)的動態(tài)并發(fā)調(diào)整。例如對于最常見的兩個原表數(shù)據(jù)的 join (M1 join M2 at J)， 如果用節(jié)點大小來表示其處理數(shù)據(jù)的的多少，那對于下圖這么一個例子，M1 處理的是中等的一個數(shù)據(jù)表(假設 M1 需要并發(fā)度為 10)，M2 處理的是較大的數(shù)據(jù)表(并發(fā)度為1000)，naïve 的執(zhí)行方式會將按照 10 + 1000 的并發(fā)度調(diào)度，同時因為 M2 輸出需要全量 shuffle 到 J， J 需要的并發(fā)度也會較大 (~1000)。

而實際上，對于這種計算 pattern 而言，M2 需要讀取(并進行處理)的，應該只有能和 M1 的輸出 join 得上的數(shù)據(jù)，也就是說在考慮了整體執(zhí)行 cost 后，在這種 M1 期望的輸出數(shù)據(jù)要比 M2 小的多的情況下，可以先行調(diào)度 M1 完成計算，將 M1 輸出數(shù)據(jù)的 statistics 在 AM/DAG 端進行聚合，然后只挑選出 M2 的有效數(shù)據(jù)進行處理。這里 "M2 的有效數(shù)據(jù)"的選擇本質(zhì)上是一個 predicate push down 的過程，可以由計算引擎的優(yōu)化器和運行時聯(lián)合進行判斷。也就是說，這種情況下 M2 的并發(fā)度調(diào)整，是和上層計算緊密結(jié)合的。

一個最直觀的例子是，如果 M2 是一個 1000 個分區(qū)的分區(qū)表，并且分區(qū)的 key 和 join 的 key 相同，那么可以只讀取 M2 能和 M1 輸出 join 上的有效數(shù)據(jù)的 partition 進行讀取處理。假如 M1 的輸出只包含了 M2 原表數(shù)據(jù)的 3 個 partition keys, 那么在 M2 就只需要調(diào)度 3 個計算節(jié)點來處理這 3 個分區(qū)的數(shù)據(jù)。也就是說 M2 的并發(fā)度從默認的 1000，可以降低到 3，這在保證同樣的邏輯計算等效性與正確性的前提下，能大大降低計算資源的消耗，并數(shù)倍加速作業(yè)的運行。這里的優(yōu)化來自幾個方面：

• M2 的并發(fā)度 (1000 -> 3) 以及處理的數(shù)據(jù)量大大降低
• M2 需要 shuffle 到 J 的數(shù)據(jù)量以及 shuffle 需要的計算量大大降低
• J 需要處理的數(shù)據(jù)量以及其并發(fā)度能大大降低

從上圖這個例子中我們也可以看到，為了保證 M1 -> M2 的調(diào)度順序上，DAG 中在 M1 和 M2 間引入了一條依賴邊，而這條邊上是沒有數(shù)據(jù)流動的，是一條只表示執(zhí)行先后的依賴邊。這與傳統(tǒng) MR/DAG 執(zhí)行框架里，邊的連接與數(shù)據(jù)流動緊綁定的假設也有不同，是在 DAG 2.0 中對于邊概念的一個拓展之一。

DAG 執(zhí)行引擎作為底層分布式調(diào)度執(zhí)行框架，其直接的對接"用戶"是上層計算引擎的開發(fā)團隊，其升級對于終端用戶除了性能上的提升，直接的體感可能會少一點。這里我們舉一個終端用戶體感較強的具體例子，來展示 DAG 更加動態(tài)的執(zhí)行能力，能夠給終端用戶帶來的直接好處。就是在 DAG 動態(tài)能力的基礎上，實現(xiàn)的 LIMIT 的優(yōu)化。

對于 SQL 用戶來說，對數(shù)據(jù)進行一些基本的 at hoc 操作，了解數(shù)據(jù)表的特性，一個非常常見的操作是 LIMIT，比如：

SELECT * FROM tpch_lineitem WHERE l_orderkey > 0 LIMIT 5; 

在分布式執(zhí)行框架上，這個操作對應的執(zhí)行計劃，是通過將源表做切分后，然后調(diào)度起所需數(shù)目的 mapper 去讀取全部數(shù)據(jù)，再將 mapper 的輸出匯總到 reducer 后去做最后的 LIMIT 截斷操作。假設源表 (這里的 tpch_lineitem) 是一個很大的表，需要 1000 個 mapper 才能讀取，那么在整個分布式執(zhí)行過程中，涉及的調(diào)度代價就是要調(diào)度 1000 mapper + 1 reducer。這個過程中會有一些上層計算引擎可以優(yōu)化的地方，比如每個 mapper 可以最多輸出 LIMIT 需要的 record 數(shù)目(這里的 LIMIT 5)提前退出，而不必處理完所有分配給它的數(shù)據(jù)分片等等。但是在一個靜態(tài)的執(zhí)行框架上，為了獲取這樣簡單的信息，整體 1001 個計算節(jié)點的調(diào)度無法避免。這給這種 ad hoc query 執(zhí)行，帶來了巨大的 overhead, 在集群資源緊張的時候尤其明顯。

DAG 2.0 上， 針對這種 LIMIT 的場景，依托新執(zhí)行框架的動態(tài)能力，實現(xiàn)了一些優(yōu)化，這主要包括幾方面：

• 上游 Exponential start：對于這種大概率下上游 mapper 計算節(jié)點不需要全部運行的情況，DAG 框架將對 mapper 進行指數(shù)型的分批調(diào)度，也就是調(diào)度按照 1， 10 ... FULL 的分批執(zhí)行
• 下游的 Early scheduling：上游產(chǎn)生的 record 數(shù)目作為執(zhí)行過程中的統(tǒng)計數(shù)據(jù)上報給 AM， AM 在判斷上游已經(jīng)產(chǎn)生足夠的 record 條數(shù)后，則提前調(diào)度下游 reducer 來消費上游的數(shù)據(jù)。
• 上游的 Early termination：下游 reducer 在判斷最終輸出的 LIMIT 條數(shù)已經(jīng)滿足條件后，直接退出。這時候 AM 可以觸發(fā)上游 mapper 整個邏輯節(jié)點的提前退出(在這種情況下，大部分 mapper 可能都還沒有調(diào)度起來)，整個作業(yè)也能提前完成。

這種計算引擎和 DAG 在執(zhí)行過程中的靈活動態(tài)交互，能夠帶來大量的資源節(jié)省，以及加速作業(yè)的執(zhí)行。在線下測試和實際上線效果上，基本上絕大多數(shù)作業(yè)在 mapper 執(zhí)行完 1 個計算節(jié)點后就能提前退出，而無需全量調(diào)起 (1000 vs 1)。

下圖是在線下測試中，當 mapper 并發(fā)為 4000 時，上述 query 優(yōu)化前后的區(qū)別：

可以看到，執(zhí)行時間優(yōu)化后增速了 5X+， 計算資源的消耗更是減小了數(shù)百倍。

這個線下測試結(jié)果作為比較典型的例子，稍微有些理想化。為了評估真實的效果，在 DAG 2.0 上線后，選取了 LIMIT 優(yōu)化生效的線上作業(yè)，統(tǒng)計了一星期結(jié)果如下：這個優(yōu)化平均為每個作業(yè)節(jié)省了 (254.5 cores x min CPU + 207.3 GB x min) 的計算資源，同時每個作業(yè)上，平均能節(jié)省 4349 個(無效)計算節(jié)點的調(diào)度。

LIMIT 執(zhí)行上的改進，作為一個針對特殊場景上實現(xiàn)的優(yōu)化，涉及了整個 DAG 執(zhí)行不同策略的調(diào)整，這種細化的改進能力，能更直觀的體現(xiàn) DAG 2.0 架構(gòu)升級諸多好處：靈活的架構(gòu)使得 DAG 的執(zhí)行中擁有了更多的動態(tài)調(diào)整能力，也能和計算引擎在一起進行更多有針對性的優(yōu)化。

不同情況下的動態(tài)并發(fā)度調(diào)整，以及具體調(diào)度執(zhí)行策略的動態(tài)調(diào)整，只是圖的物理特性動態(tài)調(diào)整的幾個例子。事實上對于物理特性運行時的調(diào)整，在 2.0 的框架之上有各種各樣的應用，比如通過動態(tài)數(shù)據(jù)編排/shuffle 來解決各種運行期間的skew問題等，這里不再做進一步的展開。接下來我們再來看看 DAG 2.0 上對于邏輯圖的動態(tài)調(diào)整做的一些探索。

2)動態(tài)邏輯圖的調(diào)整

分布式 SQL 中，map join 是一個比較常見的優(yōu)化，其實現(xiàn)原理是在 join 的兩個表中，如果有一個超小的表(可以 fit 到單個計算節(jié)點的內(nèi)存中)，那對于這個超小表可以不做 shuffle，而是直接將其全量數(shù)據(jù) broadcast 到每個處理大表的分布式計算節(jié)點上。通過在內(nèi)存中直接建立 hash 表，完成 join 操作。map join 優(yōu)化能大量減少 (大表) shuffle 和排序，非常明顯的提升作業(yè)運行性能。但是其局限性也同樣顯著：如果"超小表"實際不小，無法 fit 進單機內(nèi)存，那么在試圖建立內(nèi)存中的 hash 表時就會因為 OOM 而導致整個分布式作業(yè)的失敗，而需要重跑。所以雖然 map join 在正確使用時，可以帶來較大的性能提升，但實際上優(yōu)化器在產(chǎn)生 map join 的 plan 時需要偏保守，很多情況下需要用戶顯式的提供 map join hint 來產(chǎn)生這種優(yōu)化。此外不管是用戶還是優(yōu)化器的選擇，對于非源表的輸入都無法做很好的判斷，因為中間數(shù)據(jù)的大小往往需要在作業(yè)運行過程中才能準確得知。

而 map join 與默認 join 方式 (sorted merge join) 對應的其實是兩種不同優(yōu)化器執(zhí)行計劃，在 DAG 層面，其對應的是兩種不同的邏輯圖。要支持這種運行過程中根據(jù)中間數(shù)據(jù)特性的動態(tài)優(yōu)化，就需要 DAG 框架具備動態(tài)邏輯圖的執(zhí)行能力，這也是在 DAG 2.0 上開發(fā)的 conditional join 功能。

如同下圖展示，在對于 join 使用的算法無法被事先確定的時候，允許優(yōu)化器提供一個 conditional DAG，這樣的 DAG 同時包括使用兩種不同 join 的方式對應的不同執(zhí)行計劃支路。在實際執(zhí)行時，AM 根據(jù)上游產(chǎn)出數(shù)據(jù)量，動態(tài)選擇一條支路執(zhí)行 (plan A or plan B)。這樣子的動態(tài)邏輯圖執(zhí)行流程，能夠保證每次作業(yè)運行時都能根據(jù)實際作業(yè)數(shù)據(jù)特性，選擇最優(yōu)的執(zhí)行計劃。

conditional join 是動態(tài)邏輯圖的第一個落地場景，在線上選擇一批適用性作業(yè)，動態(tài)的 conditional join 相比靜態(tài)的執(zhí)行計劃，整體獲得了將近 3X 的性能提升。

2 混合 Bubble 模式

Bubble 模式是我們在 DAG 2.0 架構(gòu)上探索的一種全新的作業(yè)運行方式，通過對于 bubble 大小以及位置的調(diào)整，可以獲取性能和資源利用率的不同 tradeoff 點。這里通過一些更加直觀的例子，來幫助大家理解 Bubble 執(zhí)行在分布式作業(yè)中的實際應用。

在上圖的 TPCH Q21 上。比如在 Q21 上，我們看到了通過將作業(yè)被切分為三個 "bubble"，數(shù)據(jù)能夠有效的在節(jié)點之間 pipeline 起來，并且通過熱點節(jié)點實現(xiàn)調(diào)度的加速。最終消耗的資源數(shù) (cpu * time) 是準實時作業(yè)的 35%, 而性能則與一體化調(diào)度的準實時作業(yè)非常相近 (96%), 比離線作業(yè)性能提升 70% 左右。

在標準 TPCH 1TB 全量測試中，混合 bubble 模式體現(xiàn)出了相比離線和準實時的一體化模式 (gang scheduling) 更好的資源/性能 tradeoff。選用 Greedy Bubble(size = 500) 的策略，bubble 相比離線作業(yè)性能提升了 2X (資源消耗僅增加 17%，具體數(shù)值略)。同時與一體化調(diào)度的準實時作業(yè)比較，bubble 執(zhí)行在只消耗了 40% 不到的資源 (cpu * time) 的前提下，其性能達到了準實時作業(yè)的 85% (具體數(shù)值略)?？梢钥吹?，這種新型的 bubble 執(zhí)行模式，允許我們在實際應用中獲取很好的性能與資源的平衡，達到系統(tǒng)資源有效的利用。Bubble 執(zhí)行模式目前正在阿里集團內(nèi)部全量上線中，我們在實際線上的作業(yè)也看到了與 TPCH 測試非常相似的效果。

如同之前所述，混合 bubble 模式支持了不同切分策略，這里提供的只是一種切分策略上的效果。在與上層計算引擎 (e.g., MaxCompute 優(yōu)化器) 緊密結(jié)合時，這種 DAG 分布式調(diào)度 bubble 執(zhí)行的能力，能夠允許我們根據(jù)可用資源和作業(yè)計算特點，來尋找性能與資源利用率的最佳平衡點。

四 資源的動態(tài)配置和動態(tài)管理

傳統(tǒng)分布式作業(yè)對于每個計算節(jié)點需要的資源類型 (CPU/GPU/Memory) 和大小都是預先確定下來的。然而在作業(yè)運行之前，對計算節(jié)點資源類型和大小的合理選擇，是比較困難的。即便對于計算引擎的開發(fā)者，也需要通過一些比較復雜的規(guī)則，才能預估出大概合理的配置。而對于需要將這些配置透明給終端用戶的計算模式，終端用戶要做出選擇就更加困難。

在這里以 PAI 的 Tensorflow (TF) 作業(yè)為例，描述 DAG 2.0 的資源動態(tài)配置能力，怎樣幫助平臺的 TF 作業(yè)選擇合理的 GPU 類型資源以及提高 GPU 資源的利用率。相比 CPU 而言，GPU 作為一種較新的計算資源，硬件的更新?lián)Q代較快，同時普通終端用戶對于其計算特點也相對不了解。因此終端用戶在指定 GPU 資源類型時，經(jīng)常存在著不合理的情況。與此同時，GPU 在線上又是相對稀缺資源。今天在線上，GPU 申請量經(jīng)常超過集群 GPU 總數(shù)，導致用戶需要花很長時間排隊等待資源。而另外一方面，集群中 GPU 的實際利用率卻偏低，平均只有 20% 左右。這種申請和實際使用之間存在的 Gap，往往是由于用戶作業(yè)配置中，事先指定的 GPU 資源配置不合理造成。

在 DAG 2.0 的框架上，PAI TF GPU 作業(yè) (見 session 2.2.2 的 dynamic PS DAG) 引入了一個額外的"計算控制節(jié)點"，可以通過運行 PAI 平臺的資源預測算法，來判斷當前作業(yè)實際需要的 GPU 資源類型，并在必要的時候，通過向 AM 發(fā)送動態(tài)事件，來請求修改下游 worker 實際申請的 GPU 類型。這其中資源預測算法，可以是根據(jù)算法的類型，數(shù)據(jù)的特點，以及歷史作業(yè)信息來做 HBO (history based optimization)，也可以通過 dry-run 的方法來進行試運行，以此確定合理的資源類型。

具體實現(xiàn)上，這個場景中 control stage 與 Worker 之間通過 concurrent edge 連接，這條邊上的調(diào)度觸發(fā)條件是在 control stage 已經(jīng)做出資源選擇決定之后，通過其發(fā)出的事件來觸發(fā)。這樣的作業(yè)運行期間的動態(tài)資源配置，在線上功能測試中，帶來了 40% 以上的集群 GPU 利用率提升。

作為物理特性一個重要的維度，對計算節(jié)點的資源特性在運行時的動態(tài)調(diào)整能力，在 PAI 以及 MaxCompute 上都能找到廣泛的應用。以 MaxCompute SQL 為例，對于下游節(jié)點的 CPU/Memory 的大小，可以根據(jù)上游數(shù)據(jù)的特點進行有效的預判;同時對于系統(tǒng)中發(fā)生的 OOM，可以嘗試自動調(diào)高 OOM 后重試的計算節(jié)點的內(nèi)存申請，避免作業(yè)的失敗，等等。這些都是在 DAG 2.0 上新的架構(gòu)上實現(xiàn)的一些新功能，在這里不做具體的展開。

五 工程化與上線

作為分布式系統(tǒng)的底座，DAG 本身的動態(tài)能力以及靈活度，在與上層計算引擎結(jié)合時，能夠支持上層計算實現(xiàn)更加高效準確的執(zhí)行計劃，在特定場景上實現(xiàn)數(shù)倍的性能提升以及對資源利用率的提高。在上文中，也集中介紹了整個 DAG 2.0 項目工作中，開發(fā)實現(xiàn)的一些新功能與新的計算模式。除了對接計算引擎來實現(xiàn)更高效的執(zhí)行計劃，調(diào)度本身的敏捷性，是 AM/DAG 執(zhí)行性能的基本素質(zhì)。DAG 2.0 的調(diào)度決策均基于事件驅(qū)動框架以及靈活的狀態(tài)機設計來實現(xiàn)，在這里也交出 DAG 2.0 在基本工程素養(yǎng)和性能方面的成績單：

這里選用了最簡單的 Map-Reduce (MR) 作業(yè)為例，對于這種作業(yè)，調(diào)度執(zhí)行上并無太多可以取巧的地方，考驗的是調(diào)度系統(tǒng)本身的敏捷度和整個處理流程中的全面去阻塞能力。這個例子也凸顯了 DAG 2.0 的調(diào)度性能優(yōu)勢，尤其作業(yè)規(guī)模越大，優(yōu)勢越發(fā)明顯。此外，對于更接近線上的 work-load 的場景，在 TPCDS 標準 benchmark 中，當執(zhí)行計劃和運行邏輯完全相同時，2.0 (未打開動態(tài)執(zhí)行等功能)的高性能調(diào)度也給作業(yè)帶來了顯著的提升。

最后，對于一個從頭到尾完整替代原有系統(tǒng)的新一代的全新框架，怎樣無縫對接線上場景，實現(xiàn)大規(guī)模的上線，是一個同樣重要(甚至更重要)的話題，也是對一個實際生產(chǎn)系統(tǒng)進行升級，與小范圍的新系統(tǒng) POC 之間最大的區(qū)別。今天的伏羲調(diào)度系統(tǒng)，每天支撐著阿里集團內(nèi)外大數(shù)據(jù)計算平臺千萬的分布式作業(yè)。DAG/AM 這一核心分布式調(diào)度執(zhí)行組件的更新?lián)Q代，要完整替換線上已經(jīng)支撐了大數(shù)據(jù)業(yè)務 10 年的分布式生產(chǎn)系統(tǒng)，而不造成現(xiàn)有場景的失敗，這需要的不僅僅是架構(gòu)和設計上的先進性。如何在"飛行中換引擎", 保質(zhì)保量的實現(xiàn)系統(tǒng)升級，其挑戰(zhàn)完全不亞于新的系統(tǒng)架構(gòu)本身的設計。要實現(xiàn)這樣的升級，擁有一個穩(wěn)固的工程基座，以及測試/發(fā)布框架，都是不可或缺的。沒有這樣子的底座，上層的動態(tài)功能與新計算模式，都無從談起。

目前 DAG 2.0 目前已全面覆蓋了阿里集團 MaxCompute 所有線上的 SQL 離線作業(yè)和所有準實時作業(yè)，以及 PAI 平臺的所有 Tensorflow 作業(yè)(CPU 和 GPU)+ PyTorch 作業(yè)。每天支撐數(shù)千萬分布式作業(yè)的運行，并經(jīng)受了 19 年雙11 /雙12 的考驗。在面對兩次大促創(chuàng)歷史記錄的數(shù)據(jù)洪峰(相比 18 年增長 50%+)壓力下，保障了集團重點基線在大促當天準時產(chǎn)出。與此同時，更多種類型的作業(yè)(例如跨集群復制作業(yè)等等)正在遷移到 DAG 2.0 的新架構(gòu)，并且依托新的架構(gòu)升級計算作業(yè)本身的能力。DAG 2.0 的框架基座的上線，為各條計算線上依托其實現(xiàn)新功能打下了堅實基礎。

六 展望

伏羲 DAG 2.0 核心架構(gòu)的升級，旨在夯實阿里計算平臺長期發(fā)展的基礎，并支持上層計算引擎與分布式調(diào)度方面結(jié)合，實現(xiàn)各種創(chuàng)新和創(chuàng)建新計算生態(tài)。架構(gòu)的升級本身是向前邁出的重要一步，但也只是第一步。要支撐企業(yè)級的，各種規(guī)模，各種模式的全頻譜計算平臺，需要將新架構(gòu)的能力和上層計算引擎，以及伏羲系統(tǒng)其他組件進行深度整合。依托阿里的應用場景，DAG 2.0 除了在作業(yè)規(guī)模等方面繼續(xù)在業(yè)界保持領先之外，架構(gòu)和功能上也有許多創(chuàng)新， 比如前面我們已經(jīng)介紹過的：

• 在業(yè)界首次在分布式執(zhí)行框架上，實現(xiàn)了執(zhí)行過程中邏輯圖和物理圖的雙重動態(tài)可調(diào);
• 通過 Bubble 機制實現(xiàn)了混合的計算模式，探索資源利用率和作業(yè)性能間的最佳平衡。

除此之外，2.0 更加清晰的系統(tǒng)封層架構(gòu)帶來的一個重要改變就是能有利于新功能更快速開發(fā)，提速平臺和引擎向前創(chuàng)新。由于篇幅有限，本文只能由點及面地介紹一部分新功能與新計算模式，還有許許多多已經(jīng)實現(xiàn)，或正在開發(fā)中的功能，在業(yè)界都是全新的探索，暫時不做進一步展開，比如：

• 準實時作業(yè)體系架構(gòu)的整體升級: 資源管理與多作業(yè)管理的解耦，支持準實時作業(yè)場景上的動態(tài)圖功能。
• 常駐的單 container 多 slot 執(zhí)行的 cache-aware 查詢加速服務 (MaxCompute 短查詢)
• 基于狀態(tài)機的作業(yè)節(jié)點管理以及失敗下的智能重跑機制。
• 動態(tài)可定義的 shuffle 方式：通過 recursive shuffle 等方式動態(tài)解決線上大規(guī)模作業(yè)中的 in-cast 問題。
• 基于 adaptive 的中間數(shù)據(jù)動態(tài)切分與聚合，解決實際分布式作業(yè)中各種數(shù)據(jù)傾斜問題
• 支持 PAI TF GPU 作業(yè)的多執(zhí)行計劃選項。
• 通過 DAG 執(zhí)行過程中與優(yōu)化器的交互，實現(xiàn)漸進式的交互式動態(tài)優(yōu)化。
• 支持 Imperative 語言特性，通過 DAG 的動態(tài)自增長等能力，對接 IF/ELSE/LOOP 等語義。

核心調(diào)度底座能力的提升，能夠為上層的各種分布式計算引擎提供真正企業(yè)級的服務能力，提供必須的彈藥。而這些計算調(diào)度能力提升帶來的紅利，最終會通過 MaxCompute 和 PAI 等引擎，透傳到終端的阿里云計算服務的各個企業(yè)。在過去的十年，阿里業(yè)務由內(nèi)向外的驅(qū)動，鍛造了業(yè)界規(guī)模最大的云上分布式平臺。而通過更好服務集團內(nèi)部以及云上的企業(yè)用戶，我們希望能夠提升平臺的企業(yè)級服務能力，可以完成由內(nèi)向外，到由外至內(nèi)的整個正向循環(huán)過程，推動計算系統(tǒng)螺旋式上升的不斷創(chuàng)新，并通過性能/規(guī)模，以及智能化自適應能力兩個維度方面的推進，降低分布式計算服務的使用門檻，真正實現(xiàn)大數(shù)據(jù)的普惠。

Reference

[1]Dryad: Distributed Data-parallel Programs from Sequential Building Blocks

Michael Isard, Mihai Budiu, Yuan Yu, Andrew Birrell, Dennis Fetterly, Proceedings of the 2007 Eurosys Conference | March 2007

[2]MapReduce: Simplified data processing on large clusters

Jeff Dean and Sanjay Ghemawat, Proceedings of the 6th Symposium on Operating

Systems Design and Implementation (OSDI)| December 2004

[3]Apache Tez: A Unifying Framework for Modeling and Building Data Processing Applications

Bikas Sahah, Hitesh Shahh etc., Proceedings of the 2015 ACM SIGMOD International Conference on Management of Data| June 2015