<b></b>
<b>摘要：</b>回顧大資料技術領域大事件，最早可追溯到06年hadoop的正式啟動，而環顧四下，圍繞着資料庫及資料處理引擎，業内充斥着各種各樣的大資料技術。這是個技術人的好時代，僅資料庫領域熱門db就有300+，圍繞着hadoop生态圈的大資料處理技術更是繁花似錦。在雲栖社群2017線上技術峰會大資料技術峰會上，阿裡雲大資料計算平台架構師林偉做了題為《maxcompute的大腦：基于代價的優化器》的分享，為大家分享阿裡巴巴大資料計算服務的大腦——基于代價的優化器的設計和架構。
<b>maxcompute簡介</b>
大資料計算服務(maxcompute)是一種快速、完全托管的pb/eb級資料倉庫解決方案，maxcompute具備萬台伺服器擴充能力和跨地域容災能力，是阿裡巴巴内部核心大資料平台，承擔了集團内部絕大多數的計算任務，支撐每日百萬級作業規模。maxcompute向使用者提供了完善的資料導入方案以及多種經典的分布式計算模型，能夠更快速的解決使用者海量資料計算問題，有效降低企業成本，并保障資料安全。
<b>maxcompute架構</b>
maxcompute基本的體系結構如上圖所示，最底層就是在實體機器之上打造的提供統一存儲的盤古分布式檔案存儲系統；在盤古之上一層就是伏羲分布式排程系統，這一層将包括cpu、記憶體、網絡以及磁盤等在内的所有計算資源管理起來；再上一層就是統一的執行引擎也就是maxcompute執行引擎；而在執行引擎之上會打造各種各樣的運算模式，比如流計算、圖計算、離線處理、記憶體計算以及機器學習等等；在這之上還會有一層相關的程式設計語言，也就是maxcompute語言；在語言上面希望為各應用方能夠提供一個很好的平台，讓資料工程師能夠通過平台開發相關的應用，并使得應用能夠快速地在分布式場景裡面得到部署運作。
<b>maxcompute的研發思路</b>
<b>maxcompute的研發思路主要分為以下四個方面：</b>
<b>高性能、低成本和大規模</b>。希望打造的maxcompute平台能夠提運算的高性能，盡可能降低使用者的使用成本，并且在規模上面能夠達到萬台機器以及多叢集的規模。
<b>穩定性，服務化。</b>希望maxcompute平台能夠提供穩定性和服務化的方式，使得使用者不用過多地考慮分布式應用的難度，而隻需要注重于使用者需要進行什麼樣的計算，讓系統本身服務于使用者，并能夠提供穩定性，服務化的接口。
<b>易用性，服務于資料開發者。</b>希望maxcompute平台是易用的，并且能夠很友善地服務于資料開發工程師，不需要資料工程師對于分布式的場景進行很深的了解，而隻要關注于需要用這些資料進行什麼樣的運算就可以，接下來就是由maxcompute平台幫助資料開發工程師高效并且低成本地執行自己的想法。
<b>多功能。</b>希望maxcompute能夠具有更多的功能，不僅僅是支援流計算、圖計算、批處理和機器學習等，而希望更多種類的計算能夠在maxcompute平台上得到更好的支援。
<b>maxcompute的大腦——優化器</b>
基于以上的研發思路，maxcompute平台需要擁有一個更加強大的大腦，這個大腦需要更加了解使用者的資料，更加了解使用者的計算，并且更加了解使用者本身，maxcompute的大腦需要能夠幫助使用者更加高效地優化運算，通過系統層面去了解使用者到底需要進行什麼樣的運算，進而達到之前提到的各種目的，使得使用者能夠從分布式場景中脫離出來，不必去考慮如何才能使得運算高效地執行，而将這部分工作交給maxcompute的大腦，讓它來為使用者提供更智能的平台，這也就是maxcompute所能夠為使用者帶來的價值。
那麼maxcompute的大腦究竟是什麼呢？其實就是優化器。優化器能夠将所有資訊串聯在一起，通過了解系統中資料的相關性以及使用者的企圖，并通過機器的能力去充分地分析各種各樣的環境，在分布式場景中以最高效的方式實作對于使用者運算的執行。在本次分享中以離線計算作為主要例子來對于maxcompute的大腦——優化器進行介紹。
首先對于離線計算的概念進行簡單介紹，maxcompute離線計算架構設計如上圖所示。在計算層面往往會存在一個類似進階語言的腳本語言，maxcompute提供的是類sql的腳本語言，将腳本語言通過frontend送出進來，之後經過處理轉化成為邏輯執行計劃，邏輯執行計劃在optimizer（優化器）的指導下翻譯成更加高效的實體執行計劃，并通過與runtime的連接配接之後由伏羲分布式排程系統将實體執行計劃分解到運算節點上進行運算。
上述過程的核心就在于如何充分地了解使用者的核心計劃并通過優化得到高效的實體執行計劃，這樣的過程就叫做優化器optimizer。目前開源社群内的hive以及spark的一些優化器基本上都是基于規則的優化器，其實對于優化器而言，單機系統上就存在這樣的分類，分成了基于規則的優化器和基于代價的優化器。
在單機場景裡面，oracle 6-9i中使用的是基于規則的優化器，在oracle 8開始有了基于代價的優化器，而oracle 10g則完全取代了之前基于規則的優化器；而在大資料場景裡面，像hive最開始隻有基于規則的優化器，而新版的hive也開始引入了基于代價的優化器，但是hive中還并不是正真意義上的代價優化器。而maxcompute則使用了完全的基于代價的優化器。那麼這兩種優化器有什麼差別呢？其實基于規則的優化器理論上會根據邏輯模式的識别進行規則的轉換，也就是識别出一個模式就可能觸發一個規則将執行計劃從a改成b，但是這種方式對資料不敏感，并且優化是局部貪婪的，就像爬山的人隻看眼前10米的範圍内哪裡是向上的，而不考慮應該先向下走才能走到更高的山頂，是以基于規則的優化器容易陷入局部優但是全局差的場景，容易受應用規則的順序而生産迥異的執行計劃，是以往往結果并不是最優的。而基于代價的優化器是通過volcano火山模型，嘗試各種可能等價的執行計劃，然後根據資料的統計資訊，計算這些等價執行計劃的“代價”，最後從中選用代價cost最低的執行計劃，這樣可以達到全局的最優性。
這裡分享一個具體的例子幫助大家了解為什麼基于規則的優化器無法實作全局的最優化。上圖中的這段腳本的意思就是先在a、b和c上面做完join，join出來的結果在某一列上面進行group by操作并計算出平均值。可以将上述的查詢過程畫成樹形的邏輯執行計劃，在資料庫領域往往是bottom-up的，也就是對于邏輯計劃樹而言，葉子節點是輸入，最終的目标輸出則是根節點，是以最終的資料流向是從下向上的。可以看到在這個邏輯計劃裡面，首先是對于a、b、c三個表進行join，假設size(b)<size(c)<size(a)，也就是b、c這兩張表的大小是比a小的，這樣就可以獲得另一種執行的方案就是先将b和c進行join之後再與a進行join，這之後再進行計算平均值，這樣的做法b和c join的中間結果從機率上面就會比較小，再與a join之後最終産生的結果規模也就會比較小，但是後面還需要對于b的c2列計算平均值，是以如果先做完b和c的join，而在第二次join時需要按照join的條件進行partition分區，需要按照a表的c1列和b表的c1列進行分區，但是後面需要在b表的c2列上計算平均值，這時候就會引入一個改變。因為在做完join之後，其實partition分區是在a表或者b表的c1上面的，但是要做的group by卻是在b表的c2上面的，是以需要引入exchange，這樣就會引入額外的data shuffling。而如果a、b、c三張表的大小差别并不大，其實就可以先讓a和b進行join之後再與c進行join，這樣第二次join正好是在b的c2列上進行的，是以在接下來進行計算平均值的時候就不需要引入額外的data shuffling，雖然進行join的時候代價比原來高，但是因為省去了一次data shuffling，是以從全局上來看則是更為優化的，這個例子就說明了基于規則的優化器可以得到基于局部優化，但是可能無法得到全局的優化。
基于代價的優化器則采用了不同的方案，它首先通過火山模型将查詢展開成為多個等價的可執行計劃。例子中可以先讓a和b join之後再join c或者先讓b和c join之後再join a，在這兩個計劃中，因為下面的計劃中多了一個exchange，而對于基于代價的優化器而言會在最後面有一個cost代價模型，通過計算發現第一個計劃在cost上面更優，是以就會選擇最優的計劃進行執行。在基于代價的優化器中做了很多分布式場景之下特有的cost模型，并且考慮到了non-sql，因為很多場景是與網際網路有關的應用，使用者需要很多non-sql的支援，是以可以通過使用者自定義函數幫助使用者實作一些non-sql與關系資料結合的查詢優化，最後還有一些多種分布式場景的優化，這也是基于代價的優化器差別于單機優化器所做的一些工作。
接下來分享一下volcano火山模型的相關，其實volcano模型是代價模型的一個引擎，這個引擎其實在單機系統上面就已經提出來了。volcano模型裡面也會有一些規則，但是與基于規則的優化器中的規則不同，這裡面的規則更像是一些轉化函數。volcano模型首先會對于邏輯執行計劃進行分組，之後在組上面要完成一件工作，就會先在組裡面探索局部的表達式，然後根據一些規則應用一些變換，這些變換原則上都是代數等價的，在每次進行等價變化的時候其實并不是取代原來的邏輯執行計劃樹，而是在原本的基礎之上分裂出新樹。是以最後将會出現很多個等價的執行計劃樹，最終可以通過基于代價的優化器去選取最好的執行計劃。volcano模型的原則是首先希望每個規則更加局部性，也就是局部性和正交的規則越好，就越能夠使得對于空間探索得更加全面。舉個例子，如果在平面上定義了前後左右四個方向，那麼就可以通過這四個方向搜尋整個二維平面的任何一個點，同樣的優化問題就是在空間裡選取最好的計劃，那麼就希望在每一次變化時候的探索規則都能夠正交，這樣就可以用更少的規則去探索整個空間，這樣如何去探索空間和選取探索最優路徑就可以交給引擎了。
前面分享的比較抽象，這裡進一步進行舉例說明，希望能夠加深大家對于優化過程的了解。假設有一個非常複雜的邏輯執行計劃樹，這就是真正需要做的使用者的任務，現在将其中一小部分提取出來，在進行計劃的優化時首先會分析有沒有已有的規則可以與模式比對，假設圖中的兩個節點正好能與模式比對，一個是filter一個是project，理論上filter想要推到葉節點，也就是越早進行filter越好，現在就有一個模式：如果filter出現在project之上，也就是需要先做filter之後進行project，這樣就可以轉換成另一種計劃，将這兩個節點變成新的節點，也就是可以将filter和project換順序，這樣就是應用規則的過程。同樣的還有另外一個節點，比如是aggregate操作能夠與其他的模式比對，之後就可以尋找對應的規則，并轉化出等價的節點操作，這樣就可以通過複用一棵樹節點的模式來維護多棵樹，在這裡例子中可以看到使用了兩個規則，看上去節點上是隻是一個存儲，但是實際上卻描述了四棵等價樹。之後會對于這四棵等價樹花費的代價進行計算，最後選取花費代價最低的樹作為執行計劃。整體的基于代價的優化過程就是這樣，但是可以看到當邏輯計劃樹規模很大并且規則變化有很多種的情況下，整個的探索空間還是非常龐大的，是以需要在很多因素上對于優化過程進行考慮。
接下來為大家介紹一下優化引擎的大緻算法，下圖是簡化後的優化引擎算法，而在真正實施時還有很多需要考慮的因素下圖中并沒有表示出來。
首先會将一個邏輯執行計劃中的所有邏輯節點都注冊進去，注冊進去的同時就會将規則與已有的邏輯模式進行比對，然後将比對成功的規則推到規則隊列裡面，然後循環地彈出規則隊列中的規則，并真正地應用這個規則。當然應用規則存在兩種條件，一種就是應用之後能夠産生等價樹，也就是能夠在樹的局部分裂出另外一種樹形狀态，而在分裂出來的樹上面也可能與其他的模式比對，如果局部範圍内的全部規則都已經比對完成，就可以開始計算花費的代價。而當通過計算代價得出最佳方案之後，就可以放棄在該局部進行繼續優化，如果認為目前的計劃仍然不是最優的，就可以将該cost記錄下來，繼續優化樹的其他部分，直到最終找到最佳計劃。
<b>分布式查詢中的優化問題執行個體</b>
在這裡給大家列舉一些在分布式系統中有别與單機系統中分布式查詢中的優化問題的執行個體。
例1其實很簡單，就是對于兩個表進行join操作，t1已經按照a，b進行了分區；t2已經按照a進行了分區，join的條件就是t1.a=t2.a。一種方法因為t1是按照a和b分區好的，join條件在a上面，是以需要對于t1按照a重新進行分區之後再與t2進行join。但是如果t1表非常大，遠遠大于t2表的規模，這時候就不想将t1按照重新進行分區，反而可以采用另一種方案，就是将t2作為一個整體，将t2的所有資料廣播給t1每一個資料，因為join條件是在a上面做内連接配接，是以可以做這樣的選擇，這樣就可以避免将很大的資料進行reshuffle。在這個場景中，如何去感覺join的條件是關鍵。上圖例子中的兩個計劃并不存在絕對的最優，而是需要根據的資料的大小、t2資料量以及t1資料分片的分布情況來決定哪一種方案才是最優解，對于這個問題在sifmod12上面有很多的論文進行了讨論，在此就不再展開詳細的叙述。
再分享分布式優化問題的裡另外一個例子，如圖所示，t1和t2還是在a上面進行join，join完成之後會有一個條件限制t1.a>20，完成之後會進行project，并将完成的結果當做新的一列b，最終希望所有的結果是order by b的。t1和t2都是range partition好了，這裡不是hash partition，而且因為已經進行了global sort，是以這裡在做join的時候就可以利用到兩個表之間的range partition boundary，而不需要重新reshuffle資料，比如目前已經知道大于20會在哪些分區裡面出現，可以根據選擇的boundary去讀取相應的資料之後進行作為，可以盡量避免資料shuffling，在做完join之後，還會有一個使用者定義方法，将這個方法出來的結果按照order by b的規則進行排序，假設這個foo()方法是單調遞增的函數，這樣就可以利用上面的條件也就是已經按照範圍分區好了，經過join和foo()還能儲存b的順序，就不用引入一個exchange，可以直接order by b操作。這樣就是分布式中的一個查詢優化，也就是如果能夠了解資料裡面的分片，能夠知道資料的分布式情況還能了解使用者的自定義函數方法，以及這些方法通過什麼樣的途徑與優化器進行互動，就可以對于分布式查詢進行優化。這其實是通過了使用者的annotation就可以知道使用者的方法具有什麼樣的特性，能夠保持什麼樣的資料屬性。
<b>使用者自定義函數udf</b>
在分布式系統特别是non-sql中需要大量的使用者定義函數來進行擴充，因為很多查詢過程不是像join和aggregate這麼簡單的，而需要對于很多比較獨特的功能進行模組化，是以需要使用者自定義的函數實作。一旦有了使用者自定義的函數，優化器往往難以了解udf，那麼優化的範圍将會極大地受到限制，如上圖中的中間黃色的節點包含了使用者自定義的函數，但是可能系統并不知道這個函數所做的事情，那麼在優化的時候就可能分成三個更小的可優化片段，在在三個小片段中進行進一步優化。如果優化器能夠了解使用者自定義的函數在做什麼事情，那麼就可以讓優化器穿透udf達到更大範圍的優化。那麼udf有什麼特性能夠幫助優化器穿透它呢？其實可以分析udf是不是row-wise操作的，考慮它是不是一行一行處理，不存在跨行的，考慮udf是不是單調函數，是不是在處理時有些列是不變的，也就是可以穿透的，它是不是可以保持資料分片或者保持排序，以及在cost上面的一些資訊，它的selectivity高還是低，以及data distribution of output是多還是少等等都能優化器更好地優化，為優化器打開更大的優化空間，實作更加靈活的優化，幫助cost模型選出更優的方案。這也是阿裡巴巴目前在maxcompute優化器上正在做的一些工作。
<b>優化規則</b>
maxcompute基于代價的優化器做了大量的優化，實作如下圖所示的各種優化，這裡就不展叙述開了。可以從下圖中看到在查詢中有很多優化可以去做，這些所有的優化在整個系統引擎上面都是一個個算子，這些算子也在變化圖，産生了很多個等價的樹，由優化的引擎通過cost模型去選擇最佳的方案。
<b>cost模型</b>
什麼是cost模型呢？其實cost模型最需要關注的就是本身的代價模型。每個cost模型都需要關注于局部，比如input是什麼樣的cost，經過join之後又會得到什麼樣的cost，而不需要關注于全局，全局方案的cost則是由引擎通過累計得到的整體cost。好的cost模型力求能夠反映客觀的實體實作，cost模型不需要得到和真實一模一樣，cost模型的最終目的是希望差別方案的優劣，隻需要能夠選出較優的計劃，并不需要cost的絕對值具有什麼樣的特性。現在傳統的資料庫的cost模型還是很早以前的模型，就算硬體結構已經發生了變化，隻要還是馮諾依曼體系結構，架構沒有發生改變，cost模型就可以用于選擇最優的方案。
其實優化器還有很多其他方面的因素可以考慮，比如在規則方面，需要根據規則進行等價的變換，最後根據cost模型選取最優的方案。随着邏輯計劃規模的變大，如果枚舉所有可能的方案就會極大地耗費時間，特别是在maxcompute上希望邏輯執行計劃越大越好，因為這樣就能給優化引擎更大的空間，但是這就帶來當枚舉所有的計劃時，有些枚舉的計劃其實是不必要的，可能已經處于在一個不優化的情況下了。是以如何去做有效的剪枝，如何去避免不必要的探索空間，也是實作一個好的優化器所需要考慮的。另外對探索空間的選擇，可以将時間用在最有可能是最優化的計劃的空間上面，這可能是一個比較好的選擇，因為不能希望通過np-hard的時間去選擇最優的計劃，而應該希望在有限的時間内選取比較好的執行計劃，是以在優化領域中其實不一定需要尋找最佳的方案，而是要避免最差的方案，因為在優化上面總會存在時間限制。
<b>為什麼基于代價的優化器對于maxcompute平台越來越重要了呢？</b>
這是因為阿裡巴巴希望能從hive的一條條查詢語句中走出來，提供更加複雜的存儲過程。在上圖中有一個展示，可以通過變量指派以及預處理if-else編寫出更加複雜的查詢過程和存儲過程，而基于規則的優化器會因為貪婪算法而越走越偏，最終很可能得不到全局最優方案，而邏輯計劃的複雜化使得可以優化的空間變大了，但是同時也使得對于優化器的要求變得更高，是以需要更好的基于代價的優化器幫助選擇比較好的執行計劃。而在分布式以及non-sql等新型的場景下，使用基于代價的優化器有别于傳統單機優化器的方式，是以需要有對于資料、運算和使用者更加深刻的了解來使得基于代價的優化器更加智能。
<b>了解資料</b>
那麼展開來看，什麼叫做了解資料呢。在資料格式方面，了解資料需要對于更多的資料索引以及異構的資料類型進行了解，對于結構化的資料、非結構化的資料以及半結構化的資料都進行了解，而在大資料的場景裡面資料是有一些power-law屬性的，有百萬稀疏列的表格，需要在這樣的場景下實作一個更好的優化；了解資料也需要了解豐富的資料分片方式，這是在分布式場景中才有的，資料分片可以是range/hash/directhash的，而存儲可以是columnstorage/columngrouping的，還需要用hierarchy partition來進行分級分區；還會需要了解完善的資料統計資訊和運作時資料，需要了解histogram、distinct value以及data volume等等。
<b>了解運算</b>
從了解運算方面，需要更加了解使用者自定義的函數，能夠與優化器進行互動，更夠讓使用者通過annotation的方式顯示在運算中資料的屬性上具有的特性，使得可以進行全局範圍的優化。在運作時也會進行更多的優化，比如會在中間運作到一定階段時需要判斷資料量的大小，根據資料量的大小進行并行化的選擇，并根據資料的位置選擇網絡拓撲上的優化政策。還可以做實時性，規模性，性能，成本，可靠性之間的平衡，并且使用網絡shuffling做記憶體計算以及流計算等。
<b>了解使用者</b>
從了解使用者的角度，需要了解在優化器上的使用者場景，了解多租戶場景下使用者對規模，性能，延時以及成本不同需求等，并在這樣的場景下讓優化器選取最佳的方案；在生态上面，優化器是核心的優化引擎，希望能夠在語言上面更多地開放，希望能與更多的語言和生态進行對接，也希望能夠提供強大的ide能來為開發者提供完整的開發體驗；最後希望能夠在統一的平台上提供多種運算的模式，使得優化器真正能夠成為運算的大腦。