一、增量更新和處理架構

1、設計增量更新架構的背景

當前數據業務場景日趨復雜， 對于時效性要求低的單一全量數據處理場景，MaxCompute可以較好地滿足需求。時效性要求很高的秒級實時數據處理或者流處理，需要使用實時系統、流系統來滿足需求。

但對于大部份業務場景，通常并不要求秒級數據更新可見，更多的是分鐘級或者小時級的增量數據處理場景，同時也會有海量數據的批處理場景。

對于此類業務場景，使用單一引擎或聯邦多引擎都會存在一些劣勢。如圖所示，如果使用單一的 MaxCompute 離線批量處理鏈路，分鐘級的數據和全量數據做處理和存儲，會存在冗余的計算和存儲成本，時效性也不能較好地得到滿足。但如果單純使用實時系統，資源消耗成本比較高，性價比較低，在處理大規模批處理時穩定性也不足。

因此，一般綜合解決方案會采用Lambda架構去支持大數據的復雜業務，全量批處理使用 MaxCompute 鏈路，時效性要求高的使用增量處理實時鏈路。但該架構也存在大家所熟知的問題，如圖中所示。

2、MaxCompute近實時增量更新和處理一體化業務架構實踐

如圖所示，對于各種數據源，我們提供豐富的數據源接入工具，來支持近實時的增量導入和離線批量數據的導入，內部使用統一的數據存儲和優化服務管理數據，統一的計算引擎支持近實時增量處理鏈路和大規模離線批處理的整體鏈路，統一的元數據服務支撐事務和文件的元數據管理。該一體化整體架構優勢顯著，可解決Lambda的一系列問題，如節省了冗余的數據存儲成本以及不同系統間的數據遷移成本，消除了多套系統處理差異導致的數據不一致問題，相對于單獨使用實時/流處理的方式，性價比更高；并且既可以滿足增量處理鏈路的時效性，也能滿足批處理的高效性。

此外，該套架構還提供了upsert、timetravel等一系列實用的功能，擴展整體的業務場景，節省用戶的資源成本，并提升用戶體驗。

3、MaxCompute近實時增量更新和處理整體技術架構

如圖所示為MaxCompute近實時增量更新和處理的整體技術架構，主要分為五個模塊進行改造，包括數據接入、計算引擎、數據優化服務，元數據管理，數據文件組織層；其它部分可直接復用 MaxCompute 已有的技術架構和實現。

• 數據接入層，主要支持各種數據源全量和近實時增量導入功能。結合多種數據導入產品定制開發支持了豐富的數據導入方式，并且改造完善了MaxCompute高效的數據通道服務（內部稱為Tunnel），支持全量和近實時增量的高效導入。
• 計算引擎層主要包含MC自研的SQL引擎以及其它第三方引擎，負責 Timetravel 和增量場景下的SQL全鏈路處理和優化。
• 數據優化服務主要由 MaxCompute內部的Storage Service ，負責智能自動管理增量數據文件，其中包括小文件合并 Clustering，數據文件的Compaction，數據排序等優化服務。
• 元數據服務，主要負責增量場景下并發事務的沖突管理、數據版本管理，Timetravel 管理、數據文件元數據管理等。
• 數據文件組織主要包含File Format以及Table Format，主要包含對全量和增量數據文件格式的管理以及讀寫相關的模塊。

4、Transactional Table 2.0（事務表2.0）

為了支持增量更新，我們設計了一種新的表類型-Transactional Table 2.0 (簡稱TT2)。對于建表操作只需在普通表基礎上額外設置主鍵primary key (PK)，以及表屬性 Transactional 為 true 即可，無需其他額外配置。

其中，PK用于支持 upsert功能， PK值相同的多行記錄最終會 merge合并成一行，以滿足主鍵唯一性的約束；Transactional表屬性表示支持 ACID屬性及事務機制。

如圖所示，TT2支持多種數據文件格式，主要支持Base File和Delta File。Delta File表示每次事務寫入的增量數據文件，會保持每一行數據的中間狀態，用于滿足近實時的增量查詢需求；Base File是由Delta File進行Compaction合并生成，不會保留中間狀態，采用了列式壓縮存儲的格式，用戶支持高效的全量數據查詢場景。

為了進一步優化讀寫效率，TT2支持按照Bucket Index 對數據進行分桶存儲，bucket 數量可通過配置表屬性write.bucket.num指定；分桶后，大部分數據操作如數據寫入、重排、優化等，可以按照bucket粒度進行并發處理，如果對bucket數據列查詢過濾，可進行Bucket級別裁剪優化，提升查詢的效率。

5、近實時增量寫入

下面將會介紹如何將數據寫入TT2表，主要分為批量寫入和近實時寫入；這里先描述如何設計高并發的近實時增量寫入場景。

我們定制開發了 Flink Connector，以及Dataworks 數據集成及其他工具，都可以實現數據的增量寫入，這些工具內部會調用MaxCompute的Tunnel通道服務提供的客戶端SDK，數據便可以以分鐘級并發寫入存儲。

數據寫入接口，目前僅支持upsert和delete兩種格式，未來會進行擴展；upsert 包含 insert/update 兩種隱含語義，如數據行不存在就代表 insert，如已存在就代表 update。commit 接口代表原子提交這段時間寫入的數據，如返回成功，則寫入數據查詢可見；如返回失敗，則數據不可見，數據需要重寫或重試。該操作滿足Read Commit隔離級別。

6、批量寫入

下面將介紹批量寫入。批量導入主要通過 SQL 進行操作。為了方便用戶操作，我們實現了整套的DML語法。SQL 引擎內核模塊包括 Compiler、Optimizer、Runtime 等都做了大量改造開發來支持新架構的功能，如針對 pk 列的去重操作，runtime 構造 upsert 格式數據的寫入，以及并發寫入等等。此處還涉及到DML操作過程中，Meta服務需要完整的事務機制來保證讀寫隔離、事務沖突檢測等操作。

7、數據組織優化服務

由于TT2事務表需要支持分鐘級近實時增量數據導入，有可能會產生大量小文件，導致存儲訪問壓力大、數據讀寫 IO 效率低、分析效率低等問題。因此，我們開發了Clustering服務來解決小文件合并的問題，該服務由之前提到的Storage Service來執行。

由左圖所示可以了解Clustering 服務的整體操作過程。在t1到t9的時間段，可見產生了大量的delta文件，Clustering會周期性地分析數據文件情況，如果滿足觸發條件，就會以Bucket為粒度，并發地執行合并操作，為了滿足不同業務場景的需求，合并的策略比較豐富，比如根據文件大小、數量、時序相關的多個維度，并按照不同層次進行合并，此外，對于超過一定大小的文件，做了一些優化，不會對其進行合并。

TT2還會寫入upsert和delete格式的數據，可能會造成中間狀態的冗余記錄比較多，計算成本高且處理效率低下，因此我們設計了Compaction操作，對所有記錄進行merge合并，消除上述中間狀態。Compaction操作由Storage Service負責執行，即支持手動觸發，也可以按照時間頻率自動觸發。

結合上圖可大概了解 Compaction 服務的整體操作流程。t1 到 t3 時間段，一些 delta files 寫入進來，觸發 compaction 操作，同樣會以 bucket 粒度并發執行，把所有的 delta files 進行 merge，然后生成新的 base file。之后 t4 和 t6 時間段，又寫入了一批新的 delta files，再觸發 compaction 操作，會把當前存在的 base file 和新增的 delta files 一起做 merge 操作，重新生成一個新的 base file。該過程會迭代進行，因此base文件可以實現加速全量快照查詢的目的。

8、Timetravel和增量查詢

此處Timetravel 查詢，主要用來查詢歷史版本的數據，主要用于有數據歷史狀態回溯需求的業務場景，或數據出錯時恢復歷史狀態數據進行數據校驗等。

通過一個簡單的case進行講解，例如上面創建了一張表，包含一些pk 列和val 列。左邊圖展示了數據變化過程，在 t2 和 t4 時刻分別執行了compaction操作，生成了兩個base文件: b1和b2。

在t1時刻，只需讀取 delta file (d1) 進行輸出；如果用戶查詢 t2 時刻，當時通過Compaction生成了b1這個base文件，只需讀取 base文件并輸出對應記錄即可。base文件會對d1和d2兩個文件合并，生成了三條記錄，消除了2a這個中間記錄。如查詢 t3 時刻，就會包含 base file ( b1) 加上 delta file (d3) 進行Merge合并輸出，后續時刻的查詢過程同上，不再贅述。

因此可以看出，Timetravel會找到要查詢的歷史版本前最新的base文件，以及后續的delta文件，一起進行Merge輸出。對于base文件主要用于提高查詢效率，用戶可以根據自己的業務場景選擇合適的頻率進行Compaction操作。由于Compaction操作本身也會占用一定的存儲和計算，因此不能盲目頻繁地執行。

下面的表格是一個增量查詢的場景，主要用于業務的近實時增量處理鏈路。查詢的時間范圍是一個左開右閉的區間，即 begin 是一個開區間，必須大于它，end 是一個閉區間。

如 begin 是 t1-1，end 是 t1，只讀取 t1 時間段對應的 delta file (d1) 進行輸出，如果 end 是 t2，會讀取兩個 delta files (d1 和 d2)；如果 begin 是 t1，end 是 t5，即查詢的時間范圍為 [t2, t5]，會查詢所有的delta文件，即d2,d3,d4,d5，進行合并輸出。這便是增量查詢和Timetravel查詢的區別。

此外，增量查詢對一些專門的場景進行優化，例如Clustering合并小文件，從語義上對已有數據記錄進行合并，因此增量查詢時不會作為新增的數據查詢出來。

9、特點總結

作為一個新設計的架構，MaxCompute 會盡量去覆蓋HUDI / Iceberg + Spark/Presto整體數據湖解決方案的業務場景，有助于有類似業務需求的用戶進行數據和業務鏈路遷移。此外，MaxCompute 離線近實時增量處理一體化架構還具備一些獨特的亮點：

• 統一的存儲、元數據、計算引擎一體化設計，做了非常深度和高效的集成，具備存儲成本低，數據文件管理高效，查詢效率高。
• 全套統一的 SQL 語法支持，非常便于用戶使用。
• 深度定制優化的數據導入工具，支持一些復雜的業務場景。
• 無縫銜接 MaxCompute 現有的業務場景，可以減少遷移、存儲、計算成本。
• 完全自動化管理數據文件，保證更好的讀寫穩定性和性能，自動優化存儲效率和成本。
• 基于 MaxCompute 平臺完全托管，用戶可以開箱即用。

二、智能物化視圖

1、智能物化視圖演進

使用過SQL的人基本都對物化視圖有大概了解，其實就是將邏輯視圖的結果物化下來，本質上就是存儲數據的物理表。其作用主要是把耗時操作的計算結果保存下來，避免重復計算，從而達到整體的查詢加速的目的。MaxCompute的物化視圖也經歷了一系列的演進過程。一開始我們就支持了比較豐富的SQL語法功能，比如聚簇，分區等。

對于分區物化視圖，類似于分區表，數據是通過分區的粒度進行存儲和管理的。在實際場景中，物化視圖的分區和源表的分區不一定保持一致，例如源表增加新的分區，物化視圖可能還沒來得及更新，或者只更新部分分區的場景。如果用戶要查詢指定的分區，但物化視圖只存了部分歷史分區數據，MaxCompute支持了分區穿透的功能來優化此場景的查詢。對于物化視圖存在的分區，可以從物化視圖中查詢，對于物化視圖不存在的分區，直接從源表中穿透讀取。這樣就可以利用物化視圖的結果，還能保證結果和源表一致。

此外最普遍的場景就是計算邏輯和物化視圖表達式計算邏輯相似，語義的輸出結果是物化視圖的子集。為了充分利用好物化視圖的結果，支持在物化視圖的數據集上，對數據進一步加工，獲得用戶查詢的結果。例如查找值大于10的數據 ，在改寫后便可以直接從物化視圖中直接增加一個過濾條件＞10，便可以搜索出大于10的結果，避免了查詢源表全量數據的過程，查詢改寫功能可有效提升查詢性能，降低資源消耗。對于圖中展示的查詢改寫例子比較簡單，MaxCompute已支持非常豐富的復雜操作，比如aggregate、join等，只要表達式等效，或查詢的結果集是物化視圖的子集，能夠轉換成對應的表達式，都可以進行改寫。

由于源表和物化視圖的數據存儲在不同地方。當源表發生更新，但物化視圖沒有更新時，SQL查詢無法利用物化視圖的結果。需要整體回退，查詢源表。為了更好地提高查詢效率，我們在語法上支持定時觸發操作，在一定的時間范圍內保證物化視圖和源表數據基本保持一致。

2、物化視圖智能推薦機制

為了使用物化視圖，用戶需要非常了解物化視圖的概念，運行原理，以及業務情況，才能達到較好的使用效果。但很多場景中，公司業務較為復雜，個人無法從全局了解公司的業務情況，因此無法從查詢最優的角度來創建高效的物化視圖。此外，用戶對于創建物化視圖前后的資源消耗情況，也難以評估。

為了加大物化視圖的使用場景和推廣，降低整體物化視圖的使用門檻，MaxCompute引擎支持自動化地分析用戶業務歷史作業的運行情況，根據合理的策略篩選出效果比較好的物化視圖，上圖為簡單的智能物化視圖機制的運行原理。首先，引擎會對所有作業進行分析，抽取出所有符合要求的子表達式， 實際策略上會盡可能選擇包含aggregate和join的子表達式做物化視圖，最終查詢優化的整體效果會更好。

其次，會對所有符合要求的子表達式進行格式統一的歸一化處理。例如將所有算子的順序進行排布整理等，隨后會對歸一化符合要求的子表達式進行合并，生成一些新的公共表達式，從而擴展應用場景。

最后，對所有篩選生成的表達式的執行效果進行評估，給出哪些表達式適合作為物化視圖的候選。此處需要獲取物化視圖計算時需要的CPU、內存、存儲等信息，從而做出相對準確的對比評估。

最后，會根據公共表達式的使用頻率和執行占用的資源效果，整體評估物化視圖優化應用的效果，按順序展現給所有用戶候選的公共表達式列表，因此即使是小白用戶，也可以無腦的選擇推薦排名靠前的物化視圖進行使用和驗證，可大大減少資源消耗，同時可以提高用戶的業務性能。

該功能在MaxCompute公共云已經上線，效果非常好，預計可以節省14%的CU資源。

三、Adaptive執行優化

1、SQL引擎多層次Adaptive執行

對比Spark到3.0版本才支持Adaptive執行框架，MaxCompute的SQL引擎一開始的定位就是多層次和多維度的動態Adaptive執行計算優化。

以圖中所述的執行聚合聚合操作的SQL為例，SQL Optimizer模塊會根據Compiler解析的SQL語法樹，根據靜態的Table或者分區級別的Stats信息結合RBO/CBO/HBO計算出執行代價較低的執行Plan，提交給Job Master執行，調度Runtime Worker進行數據計算處理。Runtime內部也會根據上游Worker的輸入數據Stats進行Plan優化調整，此外，運行中的算子會根據實時流入的數據特征，動態切換最合適的算法進行計算。

同時，Job Master也會不斷收集operator、work級別統計的數據Stats，回傳給Job Master做一些匯總和分析，進一步做Stage級別的動態優化調整，比如并發度的調整等。

此外，在運行時，Job Master還會把Stage級別的數據Stats回傳給Optimizer，它會根據這些實時Stats對還未執行的Plan重新進行優化，然后把新的Job Plan再次提交給Job Master繼續執行。

由上述流程可知，MaxCompute的SQL引擎可以自適應地根據多維度的Stats執行多層次的Adaptive優化，這可以充分發揮和協調各個模塊的能力。后面將會簡單介紹下對每個層次的優化實踐。

2、SQL Plan DAG動態調整

假如用戶執行SQL: select * from t1 join t2 where t1.a=t2.b，上圖展示一種Plan級別DAG的動態調整示例。對于Join的分布式實現，主要分為Shuffle Join和Map Join兩種實現，Shuffle Join如左側的Plan A所示，左右兩張表都要進行一次Shuffle操作，主要用于左右表都很大的場景。對于右側Plan B，會把右表的所有記錄挪到左表的所有map實例中，避免了左表的Shuffle操作，適用于左右表一個很大一個很小的場景。

但優化器在執行的過程中，無法感知右表的大小，所以無法事先決定采用哪種join實現。針對這種場景，我們同時生成plan A和plan B兩種計劃，并把它們同時傳給Job Master，Job Master會先執行右表，獲取到右表的數據總size后，再決定采用plan A還是plan B，在Plan B生效的場景下，相對Plan A通常可極大節省大表的shuffle開銷，提升幾倍的性能。

3、Stage動態調整

下面介紹Stage級別的動態調整的兩個場景。一個典型的場景就是Stage并發度的調整。當上游Stage完成之后，會按照預先設置的并發度計算出下游實例應該處理的數據量，如圖所示，便可以動態調整并發數。size較小的實例可以進行合并，size較大的實例可以進行拆分，均勻分布每個新實例的處理量，從而避免長尾和資源碰撞，使整體資源使用價值最大化。

另一個是Shuffle Join Worker動態調整的場景。在運行的過程中，如果發現有些表的數據實例發生嚴重傾斜，大概率會出現長尾問題，引擎會動態將其拆分為n個實例， 比如圖中左表數據量為60這個實例，會被拆分成3個數據量20的實例并發執行，另外一個實例會把數據broadcast分發到左表的三個實例中并發進行Join操作，從而避免長尾，縮短整體運行時間。

4、Worker內DAG動態調整

下面將介紹Worker內DAG執行的動態調整，如圖所示是一個Shuffle Join的實例。在開始運行時，可以根據從上游Worker獲取到左右表實例的size來決定走Hash Join還是Merge Join，如果符合Hash Join的DAG執行，就可以避免大表排序時發生大量的Spill落盤的操作，節省大量的IO資源，從而可以提升運行速度。

5、Operator動態執行

最后，worker內部的具體某個算子其實也可以動態執行。運行時會根據實時數據特征來Adaptive選擇不同的算法進行執行。對于Partial Hash Aggregator，可以根據實時的聚合效果決定是否持續進行聚合；對于排序可以先拿一些數據樣本做一下預排序，根據排序效果決定采用哪種排序處理后續的數據；壓縮方面，也可以根據壓縮效果決定是否壓縮等。

四、問答部分

物化視圖和物理表有什么區別？

物化視圖本質上可以理解為也是一張物理表，只不過多了一個源表的關聯信息，源表更新時，物化視圖需要同步更新。另外，物化視圖支持智能推薦，也可用于預計算的cache，用戶無需感知物化視圖存在，在查詢SQL時，如果對部分計算做了cache，可以直接從cache中讀取數據，來避免重復耗時的計算，在具體存儲上二者并沒有太大的區別。

物化視圖有沒有設置過期時間的考慮？

物化視圖會有一個生命周期，超過生命周期，物化視圖也會被刪除。

Hash Join和Merge Join有什么優劣，實際場景應該如何選擇？

這其實是分布式中的一個典型場景，對于Hash Join在具體實現上其實分兩種，一種是我們說的Map Join，Map Join會把小表全部廣播到大表側的每一個實例上，這樣大表側就無需做數據分布，可以直接從源表中讀出一部分數據，跟broadcast過來的小表做一個Hash join進行輸出即可，這樣可以避免大表側的shuffle數據重排操作。

Hash Join還有另外一種場景，也就是Shuffle Join，就是大表和小表同時做shuffle。在每個具體實例上，數據可以選擇走Hash Join，還是走Merge Join，二者是存在算法上的不同。Hash Join是選擇一個小表構建Hash表，大表會直接通過lookup進行輸出，不涉及任何排序操作，只要內存中能放下小表即可，效率比較高。對于Merge Join，左右兩張表都比較大的場景，無法從內存中放下一個Hash表，可以對左右表的數據進行排序，排序完的數據通過有序的join就無需通過Hash方式，而是可以在內存中通過流式的方式去判斷兩個group是否為同一個key即可。

此外，還有一種場景，就是左右兩邊的數據本來就是有序的，比如一些Cluster表的數據。這樣可以直接應用Merge Join，效果也會更高。所以本質上一是跟左右表的大小相關，另外也跟算法的效率相關。