一、湖倉(cāng)系統(tǒng) 阿里云EMR湖倉(cāng)系統(tǒng)

相較于傳統(tǒng)的數(shù)倉(cāng)、數(shù)據(jù)湖來(lái)講，湖倉(cāng)系統(tǒng)是一種新的數(shù)據(jù)管理系統(tǒng)。上圖展示了阿里云EMR湖倉(cāng)系統(tǒng)的整體架構(gòu)，它是圍繞著Delta Lake、Iceberg、Hudi等開(kāi)源數(shù)據(jù)湖格式構(gòu)建的，它同時(shí)具備數(shù)倉(cāng)的高性能和數(shù)據(jù)湖的低成本、開(kāi)放性。這些數(shù)據(jù)湖格式基于開(kāi)源的Parquet和ORC構(gòu)建，能夠在AWS S3、阿里OSS等低成本存儲(chǔ)系統(tǒng)上運(yùn)行，它還具備ACID事務(wù)、批流一體以及Upsert等能力，可以對(duì)接多種商業(yè)或開(kāi)源的查詢(xún)計(jì)算引擎。這些能力使得湖倉(cāng)體系逐步成為了一種趨勢(shì)。

湖倉(cāng)系統(tǒng)有一定的學(xué)習(xí)成本，比如合理配置、小文件、清理策略、性能調(diào)優(yōu)等等。下面將從湖倉(cāng)系統(tǒng)設(shè)計(jì)上入手，了解三種格式的差異。以Spark計(jì)算引擎為例，去分析讀寫(xiě)計(jì)算過(guò)程中的一些主要的鏈路和影響性能的關(guān)鍵點(diǎn)。

二、核心設(shè)計(jì)

Delta Lake、Iceberg、Hudi三個(gè)數(shù)據(jù)湖格式在功能、特性、支持程度上基本一致，但是在具體設(shè)計(jì)上各有利弊和權(quán)衡，這些設(shè)計(jì)形成了支撐湖格式特性的基石，下文將主要分析元數(shù)據(jù)、MOR讀取這兩塊核心設(shè)計(jì)。

(資料圖片僅供參考)

1、元數(shù)據(jù)

元數(shù)據(jù)由schema、配置、有效的數(shù)據(jù)文件列表三個(gè)主要部分構(gòu)成。傳統(tǒng)數(shù)倉(cāng)系統(tǒng)有單獨(dú)服務(wù)來(lái)管理原數(shù)據(jù)和事務(wù)，三個(gè)數(shù)據(jù)湖格式都是將自己的元數(shù)據(jù)以自定義的數(shù)據(jù)結(jié)構(gòu)持久化到了文件系統(tǒng)中，放置在表的路徑下，但又和表數(shù)據(jù)分開(kāi)存儲(chǔ)。Hive表路徑或分區(qū)路徑下的所有數(shù)據(jù)文件都是有效的，而數(shù)據(jù)湖格式引入了多版本的概念，所以當(dāng)前版本的有效數(shù)據(jù)文件列表需要從元數(shù)據(jù)中挑選出來(lái)。三個(gè)湖格式都封裝了自身元數(shù)據(jù)的加載和更新的能力，這些可以方便的嵌入到不同的引擎，由各個(gè)引擎Plan和Execute自己的查詢(xún)。

以Spark為例來(lái)看看Delta Lake的元數(shù)據(jù)設(shè)計(jì)，它的元數(shù)據(jù)算是三個(gè)系統(tǒng)中最簡(jiǎn)潔的：每次對(duì)Delta Lake的寫(xiě)操作，或者添加字段等DDL操作，會(huì)生成一個(gè)新版本的json deltalog文件，這里面會(huì)記錄元數(shù)據(jù)的變更，包含一些schema的配置和file的信息，多次commit之后，會(huì)自動(dòng)產(chǎn)生一個(gè)checkpoint的parquet文件，這個(gè)parquet文件會(huì)包括前面所有版本的元數(shù)據(jù)信息，用于優(yōu)化查詢(xún)加載。

Delta Lake元數(shù)據(jù)加載流程：

定位最新的checkp元數(shù)據(jù)文件List后面的deltalogjson文件按版本號(hào)依次解析，得到表的schema、配置和有效數(shù)據(jù)文件列表

Iceberg也有一個(gè)統(tǒng)一的元數(shù)據(jù)集，與Delta Lake不同的是，Iceberg是三層的架構(gòu)。

其中metadata文件很像Delta Lake的checkpoint文件，包含了全部的信息，但是不同的是，metadata文件還包括了前幾個(gè)快照的信息，并且Iceberg是三層架構(gòu)，其manifest file能夠?qū)植康臄?shù)據(jù)文件做統(tǒng)計(jì)信息收集，因此也能用于分區(qū)之下、文件之上的裁剪。

Iceberg元數(shù)據(jù)加載流程：

定位到當(dāng)前metadata文件，得到表的schema和配置，和當(dāng)前數(shù)據(jù)文件快照snapshot的manifestlist文件解析manifestlist文件，得到一組manifest文件解析manifest文件，得到有效數(shù)據(jù)文件列表

Hudi和前面兩個(gè)很不一樣，其一是它沒(méi)有統(tǒng)一的元數(shù)據(jù)結(jié)構(gòu)，其二Hudi會(huì)對(duì)數(shù)據(jù)文件進(jìn)行分組，并對(duì)文件名進(jìn)行編碼，這是Hudi特有的file group概念。Hudi的數(shù)據(jù)必須有主鍵，主鍵可以映射到一個(gè)file group，后續(xù)對(duì)于這些主鍵所有的更新，都會(huì)寫(xiě)到這個(gè)file group，直到顯式的調(diào)用修改表的文件布局。也就是說(shuō)，一個(gè)file group隨著多次的commit，會(huì)產(chǎn)生多個(gè)版本。獲取當(dāng)前有效數(shù)據(jù)文件列表時(shí)，會(huì)先列出當(dāng)前分區(qū)下的所有文件，按照f(shuō)ile group分組，取出每個(gè)group的最新文件，再按照timeline篩選掉已經(jīng)被刪除的group，最后得到一個(gè)有效的文件列表。

Hudi元數(shù)據(jù)加載流程：

解析hoodie.Properties

得到表的schema和配置

獲取有效文件列表未開(kāi)啟metadata：List filesystem + timeline開(kāi)啟metadata：讀取metadata表

Delta Lake、Iceberg、Hude元數(shù)據(jù)對(duì)比如下：

2、Merge-On-Read（MOR）

在一般情況下，在更新一個(gè)Copy-On-Write表時(shí)，即使我們只想執(zhí)行一條更新操作，也需要將所有涉及到的數(shù)據(jù)文件加載進(jìn)來(lái)，然后應(yīng)用更新表達(dá)式，再將所有數(shù)據(jù)一起寫(xiě)出，這里就包含一些沒(méi)有更新的數(shù)據(jù)，這就是寫(xiě)放大的現(xiàn)象。為了解決寫(xiě)放大現(xiàn)象，三個(gè)數(shù)據(jù)湖格式中Hudi第一個(gè)實(shí)現(xiàn)了Merge-On-Read表。

MOR的設(shè)計(jì)思想是，只持久化需要寫(xiě)出的數(shù)據(jù)，再通過(guò)某種方式標(biāo)識(shí)出來(lái)，原來(lái)的數(shù)據(jù)文件里的數(shù)據(jù)成為過(guò)期數(shù)據(jù)，讀的時(shí)候進(jìn)行合并；為了提高效率，會(huì)定期進(jìn)行合并（Compaction），通常是按照Copy-On-Write的方式寫(xiě)一遍。不同的MOR實(shí)現(xiàn)的寫(xiě)入、合并策略會(huì)有所不同。

Hudi定義了一個(gè)filegroup概念，每個(gè)group包括最多1個(gè)原始數(shù)據(jù)文件和多個(gè)日志文件，數(shù)據(jù)文件運(yùn)行不存在。

數(shù)據(jù)通過(guò)主鍵映射到filegroup，如果更新數(shù)據(jù)將會(huì)追加寫(xiě)到映射到filegroup內(nèi)的日志文件，如果是刪除，則只需要做一個(gè)主鍵記錄，在合并的時(shí)候，首先讀取原始的數(shù)據(jù)，然后按照這部分?jǐn)?shù)據(jù)的主鍵去判斷在增量日志中有沒(méi)有相關(guān)的記錄，如果存在就做合并。

Delta Lake通過(guò)Deletion Vector的設(shè)計(jì)解決寫(xiě)放大的問(wèn)題。Delta Lake將需要更新、刪除的數(shù)據(jù)在原數(shù)據(jù)文件中的offset標(biāo)識(shí)出來(lái)，寫(xiě)入一個(gè)輔助文件。Iceberg V2表有兩個(gè)MOR的實(shí)現(xiàn)，其中基于position的設(shè)計(jì)和Delta Lake的DV是基本一樣，僅在具體實(shí)現(xiàn)上有些區(qū)別。DV的寫(xiě)入僅兩步：1）根據(jù)update或者delete的condition，找到文件中匹配的記錄，記錄他們?cè)谖募械膐ffset。持久化到一個(gè)bin文件中；2）將更新后的數(shù)據(jù)，寫(xiě)到普通的一個(gè)新文件中。

相較于Hudi允許存在多個(gè)日志文件， Delta Lake在查詢(xún)性能做了權(quán)衡，一個(gè)普通數(shù)據(jù)文件只允許伴隨至多一個(gè)DV文件，當(dāng)對(duì)一個(gè)已存在DV文件的數(shù)據(jù)文件再做一個(gè)更新的時(shí)候，最終寫(xiě)出時(shí)會(huì)把兩個(gè)DV合并的。由于DV文件中offset信息是通過(guò)位圖（RoaringBitMap）來(lái)保存的，合并操作是比較高效的。另外Hudi是將更新的數(shù)據(jù)也寫(xiě)入日志文件，Delta Lake是直接寫(xiě)入普通的parquet文件，然后在bin文件中做一個(gè)標(biāo)記。以及hudi日志文件是行存格式，Delta Lake的DV采用自定義的格式，而數(shù)據(jù)使用的parquet的列存。

Delta Lake在DV下的查詢(xún)方式我們可以直接看LogicalPlan，會(huì)更加清晰，即將原本的DeltaScan轉(zhuǎn)換成Project + Filter + Delta Scan的組合。在Parquet Scan某個(gè)數(shù)據(jù)文件時(shí)，追加了_skip_row的輔助字段，上層應(yīng)用_skip_row = false的過(guò)濾，然后通過(guò)Project的投影保證僅了無(wú)輔助字段的額外輸出。

顯然核心就是_skip_row的標(biāo)記，DeltaLake自定義了ParquetFileFormat，在讀取parquet文件后，對(duì)每個(gè)數(shù)據(jù)判斷roaringBitMap是否包含該offset，有標(biāo)記為true，沒(méi)有就是false。這樣就完成了Deletion Vector模式下DeltaLake表的查詢(xún)。

Delta Lake、Hudi、Iceberg的MOR實(shí)現(xiàn)對(duì)比：

基于offset或者position的MOR實(shí)現(xiàn)，由于會(huì)通過(guò)掃描文件來(lái)確定位置，因此寫(xiě)性能上會(huì)慢于iceberg的equality或hudi mor的實(shí)現(xiàn)，而由于該方案不需要類(lèi)似hash join的讀時(shí)合并策略，查詢(xún)性能會(huì)好一些。

三、性能優(yōu)化1、查詢(xún)

以Spark為例，一個(gè)完整的query鏈路如下：

其中與數(shù)據(jù)湖相關(guān)的有三點(diǎn)：元數(shù)據(jù)加載、優(yōu)化plan、Table Scan。

（1）元數(shù)據(jù)加載

元數(shù)據(jù)加載包括獲取schema、構(gòu)造fileindex等，分為單點(diǎn)加載和分布式加載兩種。單點(diǎn)加載的代表有Iceberg、Delta Lake-Standalone，適用于小表，有內(nèi)存壓力。分布式加載的實(shí)現(xiàn)有Hudi、Delta Lake，適用于大表，需提交Spark Job。

這里我們給出LHBench做了一個(gè)測(cè)試結(jié)果：使用的是TPC-H的store-sales表，設(shè)置的filesize都是10MB，單個(gè)圖表內(nèi)表的文件數(shù)量從1千到20W，三個(gè)圖表對(duì)應(yīng)的僅讀取一行，讀取一個(gè)分區(qū)，和普通字段作為過(guò)濾條件的三個(gè)場(chǎng)景。從趨勢(shì)上三個(gè)場(chǎng)景是一致的。我們分析第一個(gè)，最左側(cè)小表Iceberg的單點(diǎn)模式要稍好些，但與Delta Lake的分布式元數(shù)據(jù)加載差距不大。最右側(cè)單點(diǎn)方式整個(gè)Query的執(zhí)行時(shí)間中藍(lán)色執(zhí)行部分基本一致，很明顯被紅色startup部分限制，這部分在plan是一致的情況下可等同于元數(shù)據(jù)加載的時(shí)間。可見(jiàn)，如何智能的選擇合適的加載模式，是一個(gè)可選的優(yōu)化方向。

以下是兩個(gè)EMR的優(yōu)化案例。

案例1：EMR Manifest，是一個(gè)無(wú)服務(wù)化的優(yōu)化元數(shù)據(jù)加載方案。

阿里云EMR一客戶(hù)的核心ODS表，通過(guò)Spark Streaming寫(xiě)入Delta Lake，每天增量數(shù)據(jù)3TB，目前全表2.2PB，1500萬(wàn)個(gè)數(shù)據(jù)文件，僅元數(shù)據(jù)10GB。在正常情況下使用Hive/Presto查詢(xún)，使用的Delta-Standalone的單機(jī)加載方式，會(huì)完全卡住或者需要超高內(nèi)存。

該優(yōu)化方案是將數(shù)據(jù)文件的元數(shù)據(jù)按照分區(qū)結(jié)構(gòu)提前持久化到一個(gè)manifest文件，同時(shí)記錄manifest的元數(shù)據(jù)版本。在用戶(hù)查詢(xún)的時(shí)候，根據(jù)filter做分區(qū)裁剪，直接去讀分區(qū)下面的manifest文件，解析出本次查詢(xún)的有效數(shù)據(jù)文件，跳過(guò)了所有元數(shù)據(jù)加載的步驟。如果emr_manifest的版本有滯后，我們也會(huì)拿到滯后的元數(shù)據(jù)，合并得到正確的數(shù)據(jù)快照。另外manifest文件中還會(huì)保存一些size、stats這些信息，會(huì)應(yīng)用于一些文件級(jí)別的data-skipping優(yōu)化。該方案在該表體量?jī)H為300TB時(shí)提供，當(dāng)時(shí)需要10GB內(nèi)存90s加載完整的元數(shù)據(jù)，優(yōu)化后可以實(shí)現(xiàn)秒級(jí)返回，且內(nèi)存不需要額外調(diào)整。

案例2：EMR DataLake Metastore，有服務(wù)的中心化的優(yōu)化元數(shù)據(jù)加載方案。

這里不得不和HMS做一個(gè)對(duì)比，HMS僅存儲(chǔ)表的分區(qū)級(jí)信息，查詢(xún)普通表時(shí)根據(jù)分區(qū)位置去list路徑，拿到有效的數(shù)據(jù)文件列表。而數(shù)據(jù)湖格式具備多版本概念，所以針對(duì)湖格式的Metastore設(shè)計(jì)必須到文件級(jí)別。另外Hudi、Iceberg基于分布式鎖來(lái)實(shí)現(xiàn)事務(wù)性，而Delta則是基于所在文件系統(tǒng)的原子性和持久性，在某些場(chǎng)景下無(wú)法提供更強(qiáng)的一致性保障，這也是我們實(shí)現(xiàn)DataLake Metastore的另外一個(gè)原因。

EMR DataLake Metastore目前已經(jīng)支持了Hudi格式，完全兼容社區(qū)版本和元數(shù)據(jù)協(xié)議，同時(shí)支持元數(shù)據(jù)雙寫(xiě)。在提供了文件快照和事務(wù)的同時(shí)，后續(xù)也將繼續(xù)拓展data profiling和行級(jí)索引為查詢(xún)提供更精準(zhǔn)的裁剪優(yōu)化。

（2）優(yōu)化plan

數(shù)據(jù)湖的第二點(diǎn)優(yōu)化是plan優(yōu)化，在前面的Spark查詢(xún)例子里，調(diào)用Spark sql內(nèi)核結(jié)合各種的狀態(tài)統(tǒng)計(jì)信息去優(yōu)化logic plan。比如根據(jù)運(yùn)行時(shí)的一些統(tǒng)計(jì)信息，通過(guò)AQE去改變join的方式，基于表的靜態(tài)信息和一些cost模型來(lái)優(yōu)化的CBO等。

統(tǒng)計(jì)信息只有表級(jí)別和column級(jí)別兩類(lèi)，其中表級(jí)別有statistics信息包括bytes、rows等信息，字段級(jí)別有min、max等。

對(duì)于一個(gè)簡(jiǎn)單的join，僅使用表級(jí)別的rowcount信息就能夠去調(diào)整join順序，使得整體的plan過(guò)程中需要join的數(shù)據(jù)量達(dá)到最少。但是如果sql帶有aggregate或者filter，就需要結(jié)合列的信息去估算整個(gè)scan的cost。

關(guān)于通過(guò)統(tǒng)計(jì)指標(biāo)來(lái)優(yōu)化查詢(xún)，有幾下幾點(diǎn)思考：

如何更好的利用statistics來(lái)優(yōu)化查詢(xún)。比如Delta Lake中count整表的sql可以直接通過(guò)元數(shù)據(jù)層面的統(tǒng)計(jì)信息得到count值，將原本的aggregate操作轉(zhuǎn)換成LocalRelation，避免Scan全表。如何高效收集/實(shí)時(shí)更新statistics。如何打通湖格式自身和Spark需要的statistics。spark會(huì)從table properties獲取統(tǒng)計(jì)信息，而一般情況湖格式的統(tǒng)計(jì)信息都是記錄到自己的元數(shù)據(jù)中，這部分需要更好的打通和復(fù)用。（3）TableScan

在完整Logical Plan的優(yōu)化并轉(zhuǎn)化為Phyiscal Plan后，下一步就要執(zhí)行具體的數(shù)據(jù)文件讀取，即Table Scan階段。

這里小文件對(duì)查詢(xún)的影響是大家所熟悉的。三個(gè)湖格式都提供了相應(yīng)的合并小文件的功能，關(guān)鍵問(wèn)題其一是目標(biāo)文件設(shè)置多大是合適的，其二合并操作執(zhí)行的時(shí)機(jī)和方式，這點(diǎn)我們后續(xù)再展開(kāi)。默認(rèn)情況下parquet在各引擎都是128M左右，但這是否是最優(yōu)的還要看表的整體規(guī)模。過(guò)多的文件，將會(huì)導(dǎo)致元數(shù)據(jù)規(guī)模變大，影響元數(shù)據(jù)的讀寫(xiě)，databricks和阿里云emr都提供了自動(dòng)調(diào)教file size的功能。這里給出databricks的設(shè)置文件大小和表規(guī)模的映射關(guān)系。為了控制元數(shù)據(jù)的規(guī)模，對(duì)于10TB的表建議文件大小設(shè)置為1G。

在得到表的全部有效數(shù)據(jù)文件后，我們還是要根據(jù)查詢(xún)條件來(lái)盡可能的進(jìn)行裁剪，以減少最終讀取Parquet或者ORC文件的體量。Table scan優(yōu)化分為兩類(lèi)：一是元數(shù)據(jù)裁剪，二是行級(jí)索引。元數(shù)據(jù)裁剪又可以分為分區(qū)裁剪、Manifest裁剪（Iceberg特有）、File裁剪等不同粒度。

三個(gè)數(shù)據(jù)湖格式都支持Z-Order和DataSkipping：先將min-max信息寫(xiě)到元數(shù)據(jù)當(dāng)中，然后結(jié)合查詢(xún)過(guò)濾條件去實(shí)現(xiàn)過(guò)濾。

針對(duì)點(diǎn)查，傳統(tǒng)數(shù)據(jù)庫(kù)更多是通過(guò)行級(jí)索引來(lái)實(shí)現(xiàn)。Databricks商業(yè)版Delta Lake支持BloomFilter，使用單獨(dú)的目錄文件，保存數(shù)據(jù)文件的索引信息；Hudi也支持Multi-Modal Index多模索引。EMR也計(jì)劃在DataLake Metastore中嵌入Index System來(lái)支持點(diǎn)查加速。

最后就是實(shí)際的Reader來(lái)讀取數(shù)據(jù)文件了。以Parquet為例，各引擎集成parquet之后，讀寫(xiě)性能已經(jīng)非常不錯(cuò)。但是有一些具體的湖格式場(chǎng)景會(huì)關(guān)閉一些優(yōu)化參數(shù)，相關(guān)的比如謂詞下推，向量化等。另外文件的壓縮格式和壓縮比也會(huì)影響文件的加載，以及目前一些Native框架支持的Native Parquet Reader（如Arrow，Presto，Velox等）。

2、寫(xiě)入

以一個(gè)帶where條件的update SQL為例，首先要能找到匹配這些查詢(xún)條件的所在的數(shù)據(jù)文件，然后加載文件，對(duì)匹配到的數(shù)據(jù)應(yīng)用表達(dá)式完成更新，并最終寫(xiě)出。在完成提交后，也許還要執(zhí)行一些表的table service，包括clean、checkpoint、compaction等。

優(yōu)化寫(xiě)操作，選擇適合當(dāng)前場(chǎng)景的表類(lèi)型（如MOR表），并配置合適的參數(shù)。在湖格式場(chǎng)景下，這種后置的追加在湖格式寫(xiě)入后面的table service其實(shí)是逐漸成為了影響效率或者整體作業(yè)穩(wěn)定性的關(guān)鍵性因素。要解決這樣的問(wèn)題，比較通用的做法就是拆成兩個(gè)鏈路：一個(gè)鏈路就是正常寫(xiě)入，另外一個(gè)鏈路是用離線(xiàn)的方式通過(guò)調(diào)度一個(gè)作業(yè)來(lái)執(zhí)行table service任務(wù)，包括像clean、Hudi表管理等等。

在阿里云EMR場(chǎng)景下，有些同時(shí)使用Flink和Spark引擎的客戶(hù)，會(huì)采用Flink Hudi入湖和Spark離線(xiàn)管理的方案。另外EMR也提供了中心化的自動(dòng)化湖表管理，是結(jié)合阿里云的DLF服務(wù)實(shí)現(xiàn)的。如圖所示，湖格式自動(dòng)將commit信息同步到DLF湖表服務(wù)，該服務(wù)將根據(jù)預(yù)定義的策略和實(shí)際表的狀態(tài)及配置判斷是否要產(chǎn)生湖表管理類(lèi)的操作。

這里的核心其實(shí)是采取怎樣的策略來(lái)更好的管理湖表。EMR目前上線(xiàn)了兩大類(lèi)5個(gè)策略。在同一類(lèi)中，又設(shè)置優(yōu)先級(jí)避免多條策略的命中而造成計(jì)算資源的浪費(fèi)。舉個(gè)例子，大多數(shù)入湖的表是以時(shí)間分區(qū)的，DLF湖表管理能夠判斷到新分區(qū)的到達(dá)，而在第一時(shí)間對(duì)已完成寫(xiě)入的分區(qū)執(zhí)行小文件合并或者zorder的操作，以便快速提供這部分?jǐn)?shù)據(jù)的高效查詢(xún)。

關(guān)鍵詞：