一、湖倉系統(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ù)據(jù)湖的低成本、開放性。這些數(shù)據(jù)湖格式基于開源的Parquet和ORC構(gòu)建，能夠在AWS S3、阿里OSS等低成本存儲系統(tǒng)上運(yùn)行，它還具備ACID事務(wù)、批流一體以及Upsert等能力，可以對接多種商業(yè)或開源的查詢計算引擎。這些能力使得湖倉體系逐步成為了一種趨勢。

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

二、核心設(shè)計

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

(資料圖片僅供參考)

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

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

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

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

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

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

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

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

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

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

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

解析hoodie.Properties

得到表的schema和配置

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

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

2、Merge-On-Read（MOR）

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

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

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

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

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

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

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

顯然核心就是_skip_row的標(biāo)記，DeltaLake自定義了ParquetFileFormat，在讀取parquet文件后，對每個數(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)，由于會通過掃描文件來確定位置，因此寫性能上會慢于iceberg的equality或hudi mor的實(shí)現(xiàn)，而由于該方案不需要類似hash join的讀時合并策略，查詢性能會好一些。

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

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

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

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

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

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

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

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

EMR DataLake Metastore目前已經(jīng)支持了Hudi格式，完全兼容社區(qū)版本和元數(shù)據(jù)協(xié)議，同時支持元數(shù)據(jù)雙寫。在提供了文件快照和事務(wù)的同時，后續(xù)也將繼續(xù)拓展data profiling和行級索引為查詢提供更精準(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)計信息去優(yōu)化logic plan。比如根據(jù)運(yùn)行時的一些統(tǒng)計信息，通過AQE去改變join的方式，基于表的靜態(tài)信息和一些cost模型來優(yōu)化的CBO等。

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

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

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

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

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

這里小文件對查詢的影響是大家所熟悉的。三個湖格式都提供了相應(yīng)的合并小文件的功能，關(guān)鍵問題其一是目標(biāo)文件設(shè)置多大是合適的，其二合并操作執(zhí)行的時機(jī)和方式，這點(diǎn)我們后續(xù)再展開。默認(rèn)情況下parquet在各引擎都是128M左右，但這是否是最優(yōu)的還要看表的整體規(guī)模。過多的文件，將會導(dǎo)致元數(shù)據(jù)規(guī)模變大，影響元數(shù)據(jù)的讀寫，databricks和阿里云emr都提供了自動調(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ù)裁剪，二是行級索引。元數(shù)據(jù)裁剪又可以分為分區(qū)裁剪、Manifest裁剪（Iceberg特有）、File裁剪等不同粒度。

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

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

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

2、寫入

以一個帶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)鍵性因素。要解決這樣的問題，比較通用的做法就是拆成兩個鏈路：一個鏈路就是正常寫入，另外一個鏈路是用離線的方式通過調(diào)度一個作業(yè)來執(zhí)行table service任務(wù)，包括像clean、Hudi表管理等等。

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

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

關(guān)鍵詞：