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

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

湖倉系統(tǒng)有一定的學(xué)習(xí)成本，比如合理配置、小文件、清理策略、性能調(diào)優(yōu)等等。下面將從湖倉系統(tǒng)設(shè)計(jì)上入手，了解三種格式的差異。以Spark計(jì)算引擎為例，去分析讀寫計(jì)算過程中的一些主要的鏈路和影響性能的關(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ù)倉系統(tǒng)有單獨(dú)服務(wù)來管理原數(shù)據(jù)和事務(wù)，三個(gè)數(shù)據(jù)湖格式都是將自己的元數(shù)據(jù)以自定義的數(shù)據(jù)結(jié)構(gòu)持久化到了文件系統(tǒng)中，放置在表的路徑下，但又和表數(shù)據(jù)分開存儲(chǔ)。Hive表路徑或分區(qū)路徑下的所有數(shù)據(jù)文件都是有效的，而數(shù)據(jù)湖格式引入了多版本的概念，所以當(dāng)前版本的有效數(shù)據(jù)文件列表需要從元數(shù)據(jù)中挑選出來。三個(gè)湖格式都封裝了自身元數(shù)據(jù)的加載和更新的能力，這些可以方便的嵌入到不同的引擎，由各個(gè)引擎Plan和Execute自己的查詢。

以Spark為例來看看Delta Lake的元數(shù)據(jù)設(shè)計(jì)，它的元數(shù)據(jù)算是三個(gè)系統(tǒng)中最簡潔的：每次對Delta Lake的寫操作，或者添加字段等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)化查詢加載。

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

• 定位最新的checkp元數(shù)據(jù)文件
• List后面的delta log json文件
• 按版本號(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的manifest list文件
• 解析manifest list文件，得到一組manifest文件
• 解析manifest文件，得到有效數(shù)據(jù)文件列表

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

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

• 解析hoodie.Properties

得到表的schema和配置

• 獲取有效文件列表
• 未開啟metadata：List filesystem + timeline
• 開啟metadata：讀取metadata表

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

2、Merge-On-Read（MOR）

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

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

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

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

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

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

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

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

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

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

三、性能優(yōu)化

1、查詢

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

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

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

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

該優(yōu)化方案是將數(shù)據(jù)文件的元數(shù)據(jù)按照分區(qū)結(jié)構(gòu)提前持久化到一個(gè)manifest文件，同時(shí)記錄manifest的元數(shù)據(jù)版本。在用戶查詢的時(shí)候，根據(jù)filter做分區(qū)裁剪，直接去讀分區(qū)下面的manifest文件，解析出本次查詢的有效數(shù)據(jù)文件，跳過了所有元數(shù)據(jù)加載的步驟。如果emr_manifest的版本有滯后，我們也會(huì)拿到滯后的元數(shù)據(jù)，合并得到正確的數(shù)據(jù)快照。另外manifest文件中還會(huì)保存一些size、stats這些信息，會(huì)應(yīng)用于一些文件級(jí)別的data-skipping優(yōu)化。該方案在該表體量僅為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è)對比，HMS僅存儲(chǔ)表的分區(qū)級(jí)信息，查詢普通表時(shí)根據(jù)分區(qū)位置去list路徑，拿到有效的數(shù)據(jù)文件列表。而數(shù)據(jù)湖格式具備多版本概念，所以針對湖格式的Metastore設(shè)計(jì)必須到文件級(jí)別。另外Hudi、Iceberg基于分布式鎖來實(shí)現(xiàn)事務(wù)性，而Delta則是基于所在文件系統(tǒng)的原子性和持久性，在某些場景下無法提供更強(qiáng)的一致性保障，這也是我們實(shí)現(xiàn)DataLake Metastore的另外一個(gè)原因。

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

（2）優(yōu)化plan

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

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

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

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

• 如何更好的利用statistics來優(yōu)化查詢。比如Delta Lake中count整表的sql可以直接通過元數(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）Table Scan

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

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

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

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

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

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

2、寫入

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

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

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

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