一、湖倉架構(gòu)

騰訊大數(shù)據(jù)的湖倉架構(gòu)如下圖所示：

這里分為三個(gè)部分，分別是數(shù)據(jù)湖計(jì)算、數(shù)據(jù)湖管理和數(shù)據(jù)湖存儲(chǔ)。

數(shù)據(jù)湖計(jì)算部分，Spark 作為 ETL Batch 任務(wù)的主要批處理引擎，F(xiàn)link 作為準(zhǔn)實(shí)時(shí)計(jì)算的流處理引擎，StarRocks 和 Presto 作為即席查詢的 OLAP 引擎。數(shù)據(jù)湖管理層以 Iceberg 為核心，同時(shí)開放了一些簡(jiǎn)單的 API，支持用戶通過 SDK 的方式去調(diào)用。在 Iceberg 之上構(gòu)建了一套 Auto Optimize Service 服務(wù)，幫助用戶在使用 Iceberg 的過程中實(shí)現(xiàn)查詢性能的提升和存儲(chǔ)成本的降低。數(shù)據(jù)湖底層存儲(chǔ)基于 HDFS 和 COS，COS 是騰訊云的云對(duì)象存儲(chǔ)，可以滿足云上用戶的大規(guī)模結(jié)構(gòu)化/非結(jié)構(gòu)化存儲(chǔ)需求，在上層計(jì)算框架和底層存儲(chǔ)系統(tǒng)之間，也會(huì)引入 Alluxio 構(gòu)建了一個(gè)統(tǒng)一的存儲(chǔ) Cache 層，進(jìn)行數(shù)據(jù)緩存提速。本次分享的重點(diǎn)主要是圍繞智能優(yōu)化服務(wù)（Auto Optimize Service）展開。

二、智能優(yōu)化服務(wù)

智能優(yōu)化服務(wù)主要由六個(gè)部分組成，分別是：Compaction Service（合并小文件）、Expiration Service（淘汰過期快照）、Cleaning Service（生命周期管理和孤兒文件清理）、Clustering Service（數(shù)據(jù)重分布）、Index Service（二級(jí)索引推薦）和 Auto Engine Service（自動(dòng)引擎加速）。以下就各模塊近期做的重點(diǎn)工作展開介紹。

1. Compaction Service

（1）小文件合并優(yōu)化

小文件合并有讀和寫兩個(gè)階段，由于 Iceberg 主要以 PARQUET/ORC 列存格式為主，讀寫列存面臨著兩次行列轉(zhuǎn)換和編解碼，開銷非常大。針對(duì)這個(gè)痛點(diǎn)，我們對(duì) Parquet 存儲(chǔ)模型進(jìn)行了分析，主要由 RowGroup、Column Chunk、Page 以及 Footer 組成，相對(duì)位置如下圖所示，不同列的最小存儲(chǔ)單元以 Page 級(jí)別組織，數(shù)據(jù)水平方向上以 RowGroup 大小劃分?jǐn)?shù)據(jù)塊，以便上層引擎按照 RowGroup 級(jí)別分配 task 加載數(shù)據(jù)。

基于存儲(chǔ)模型的特點(diǎn)，我們針對(duì)性地采用了 RowGroup Level 和 Page Level 兩種拷貝優(yōu)化，對(duì)于大文件合并大文件且僅涉及重新壓縮、僅涉及列裁剪的場(chǎng)景，使用 RowGroup Copy；對(duì)于小文件合并大文件、不涉及列變化、不涉及 BloomFilter 的場(chǎng)景，使用 Page Copy。

下面是我們內(nèi)部全部升級(jí)優(yōu)化之后的落地效果，合并時(shí)間&資源減少 5 倍多。

（2）更多優(yōu)化

我們還增強(qiáng)了 Delete Files 合并優(yōu)化和增量 Rewrite 策略。

在大規(guī)模 Update 的場(chǎng)景下，會(huì)產(chǎn)生大量的 Delete Files，數(shù)據(jù)讀取時(shí)會(huì)頻繁地進(jìn)行 Delete File Apply Data File 的操作，這個(gè)過程是串行的，I/O 開銷巨大。當(dāng)合并的速度低于 Delete File Apply 的速度，就會(huì)因?yàn)榉e攢了大量的 Delete Files 導(dǎo)致合并失敗。針對(duì)這個(gè)痛點(diǎn)，我們使用 Left Anti Join 拆分出了關(guān)聯(lián) Delete File 的 DataFile 和未關(guān)聯(lián) Delete File 的 DataFile，然后將兩者進(jìn)行 Union All。此外還在 Delete File Apply Data File 的過程中使用了 Bloom Index 加速尋找，及時(shí)刪除未關(guān)聯(lián) Data File 的 Delete File。

增量 Rewrite 優(yōu)化會(huì)通過在 DataFile 中引入 Modify Time 來決策，進(jìn)行分區(qū)級(jí)別的增量更新。

2. Index Service

（1）Iceberg Core Framework

Iceberg 較 Hive 增加了 min-max 索引，記錄了 DataFile 所有 column 列的最大值和最小值，在執(zhí)行引擎計(jì)算時(shí)可以協(xié)助做文件級(jí)別的過濾，但是文件級(jí)別的索引粒度較粗，在隨機(jī)寫數(shù)據(jù)的時(shí)候 min-max 存在交叉，導(dǎo)致索引失效。所以我們?cè)谶@個(gè)基礎(chǔ)之上進(jìn)一步拓展了二級(jí)索引，來提高 Data Skipping 的能力，加速查詢。索引的構(gòu)建和加載過程在 Iceberg Core 層的框架支持實(shí)現(xiàn)如下：

（2）Iceberg scan metrics

對(duì)于專注于業(yè)務(wù)開發(fā)的用戶來說，索引的選擇往往是比較困難的，如何精準(zhǔn)的判斷是不是需要索引，需要什么索引，索引是否有效，索引是否會(huì)帶來副作用等，往往需要經(jīng)過一些額外的任務(wù)來進(jìn)行分析，如果靠用戶自己的決策選擇，獲得大規(guī)模的適配收益很難。基于這個(gè)想法，我們做了智能推薦索引的支持，而智能的推薦，首先是需要一套 metrics 框架的支持，能夠記錄表的 Scan，F(xiàn)ilter 等各種事件，收集 Partition Status 信息，然后對(duì)這些事件進(jìn)行分析，統(tǒng)計(jì)列的查詢頻次，過濾條件，根據(jù)規(guī)則區(qū)分高/低基數(shù)列等。最后根據(jù)分析結(jié)果，進(jìn)行 Index 的推薦。

（3）索引智能推薦流程

整個(gè)端到端的 Index Service 流程如下圖：1）首先是 SQL 提取，由于我們獲取到的 SQL 是引擎優(yōu)化后的，并不是原始 SQL，所以需要進(jìn)行 SQL 重構(gòu)。2）是索引粗篩，根據(jù)拿到的信息，比如列和分區(qū)的查詢頻度，初步判斷怎么建立索引是有效的。3）開始嘗試構(gòu)建索引，支持構(gòu)建分區(qū)級(jí)別增量索引。4）在用戶無感知的情況下進(jìn)行任務(wù)雙跑。5）根據(jù)雙跑結(jié)果進(jìn)行索引優(yōu)化的效果評(píng)估。6）將索引優(yōu)化數(shù)據(jù)輸出給用戶，推薦用戶使用。7）由于索引構(gòu)建是復(fù)雜的，一個(gè)表會(huì)被多任務(wù)引用，一個(gè)任務(wù)也會(huì)去訪問多張表，我們提供任務(wù)級(jí)別和表級(jí)別的索引構(gòu)建，盡可能實(shí)現(xiàn)表級(jí)和任務(wù)級(jí)的同步優(yōu)化。

3. Clustering Service

由于 Iceberg 的 min-max 索引在隨機(jī)寫的情況下是普遍失效的，導(dǎo)致 Data Skipping 能力較差，所以如果需要精確覆蓋 min-max，可以將數(shù)據(jù)進(jìn)行重排分布。當(dāng)用戶進(jìn)行單列查詢的時(shí)候，提前對(duì)數(shù)據(jù)列排序?qū)懭耄绻嵌嗔胁樵兊那闆r，由于無法保證多個(gè)列都分布在一個(gè)文件中，我們使用 Z-order，對(duì)每個(gè)列進(jìn)行數(shù)字化處理，采樣計(jì)算 Range-ID，生成交錯(cuò)位Z-Value，根據(jù) Z-Value 進(jìn)行重分區(qū)，可以保證不同列之間的相對(duì)有序性。

實(shí)際業(yè)務(wù)中，Data Clustering 和 Data Skipping 都實(shí)現(xiàn)了四倍以上的效果提升。

4. AutoEngine Service

相對(duì)于 OLAP 引擎來講，Iceberg 表，Hudi 表都是外表，這些外表基本都是 TB 級(jí)別，使用 StarRocks，Doris 查詢外表并不能發(fā)揮 OLAP 的查詢優(yōu)勢(shì)。AutoEngine Service 通過收集 OLAP 引擎的 Event Message，對(duì)相應(yīng)的分區(qū)進(jìn)行加熱，也就是將相關(guān)分區(qū)數(shù)據(jù)路由到 StarRocks 集群，上層引擎可以在 StarRocks 集群中發(fā)現(xiàn)該分區(qū)的元數(shù)據(jù)，由此實(shí)現(xiàn)基于存儲(chǔ)計(jì)算引擎的選擇優(yōu)化。

三、場(chǎng)景化能力

1. 多流拼接

關(guān)于多流拼接，這里舉個(gè)例子簡(jiǎn)單說明， 如圖所示，有兩個(gè) MQ 同時(shí)往下游寫數(shù)據(jù)，MQ1 更新列 data1，MQ2 更新列 data2，最終根據(jù) id 聚合，取時(shí)間戳 orderColumn 排序最靠前的一條，作為 join 之后的 source。要實(shí)現(xiàn)這個(gè)合并更新能力，往往需要外接各種臨時(shí)存儲(chǔ) Redis/Hbase/MQ 等組件。

那在 Iceberg 層面是怎么優(yōu)化的呢？由于 Iceberg 本身支持事務(wù)和列級(jí)的更新刪除操作，類似于代碼倉庫的 Branch 概念，因此可以通過打 tag 的方式去標(biāo)記狀態(tài)。具體實(shí)現(xiàn)是，初始化階段，數(shù)據(jù)寫入主流程，同時(shí)多流往其他 Merged Branch 去寫入，寫完之后的話會(huì)有一個(gè)異步的 Compaction 任務(wù)，定期和主流程合并，當(dāng)用戶在讀的時(shí)候，直接讀取 Merged Branch。

2. 主鍵表

通過多流 Join 的實(shí)現(xiàn)方法依賴 Compaction Service 的調(diào)度性能，當(dāng)數(shù)據(jù)規(guī)模不斷增加，多流 join 聚合計(jì)算更新的拼接方式可能存在性能瓶頸，所以我們也引入主鍵表作為行級(jí)更新的另一種實(shí)現(xiàn)方式。比如這里我們根據(jù) id 分成四個(gè)桶，存在多個(gè)任務(wù)往一個(gè)桶去寫數(shù)據(jù)，一個(gè)桶內(nèi)的數(shù)據(jù)是有序的，那么下游在讀取桶數(shù)據(jù)的時(shí)候會(huì)更輕松。但是當(dāng) id 的基數(shù)很大的時(shí)候，比如當(dāng) id 為 4/8/16 的時(shí)候，都會(huì)往一個(gè)桶內(nèi)寫數(shù)，會(huì)產(chǎn)生 DataFile 的重疊，在下游從桶內(nèi)讀數(shù)的時(shí)候，就需要合并一個(gè)桶內(nèi)的多個(gè) DataFile 到一個(gè) Reader 處理。如果分桶數(shù)量設(shè)置的不合適，單點(diǎn)壓力就會(huì)過大，此時(shí)可以使用 Rescale 實(shí)現(xiàn)桶的彈性擴(kuò)縮容。另外在桶的基礎(chǔ)上擴(kuò)展列族 Column Family 的概念，相當(dāng)于每個(gè)列都作為獨(dú)立的文件寫入，多個(gè) Column Family 行拼接 Full Outer Join 即可。

3. In Place 遷移

由于對(duì)數(shù)據(jù)湖的高階特性能力的需要，很多業(yè)務(wù)做了架構(gòu)的升級(jí)，同時(shí)也面臨著存量 Thive（騰訊自研 Hive）和 Hive 的數(shù)據(jù)遷移到 Iceberg。這里需要重點(diǎn)支持的工作包括：存儲(chǔ)數(shù)據(jù)的遷移，計(jì)算任務(wù)的遷移。

首先存儲(chǔ)數(shù)據(jù)的遷移，我們提供了 data in-place 的方案，不搬移原來的 data files，僅僅重新生成 Iceberg 新表所需的 metadata 即可，遷移的過程支持了 STRICT/APPEND/OVERWRITE 等三種模式。

其次是計(jì)算任務(wù)的遷移支持， 我們改進(jìn)支持了新的 Name Mapping 機(jī)制，增強(qiáng)支持了 Identity partition pruning 能力，使得對(duì)于場(chǎng)景的 built-in functions 裁剪能力取得數(shù)量級(jí)性能提升，優(yōu)化實(shí)現(xiàn)如下：

4. PyIceberg

Iceberg Table Spec 是開發(fā)性的實(shí)現(xiàn)，可以支持多種語言 API 接入，AI生態(tài)圈數(shù)據(jù)科學(xué)等主要以 Python 環(huán)境為主，要求高性能 Native 解碼，對(duì) JVM 環(huán)境無強(qiáng)依賴，PySpark 雖然具備接入 Iceberg 的能力，但是太重了。我們可以直接利用 PyIceberg 能力，無JVM 依賴，加載解碼一次即可，提供廣泛的機(jī)器學(xué)習(xí)類庫的優(yōu)勢(shì)，拓展 Python的技術(shù)棧到 Iceberg 元數(shù)據(jù)層面，構(gòu)造 Pandas，Tensorflow，Pytorch 等不同的 DataFrame，方便進(jìn)行數(shù)據(jù)分析和 AI 模型訓(xùn)練的編程探索，我們內(nèi)部也深度支持了 PyIceberg SQL 的列裁剪和謂詞下推能力，結(jié)合 DuckDB 做一些小數(shù)據(jù)集的算法快速調(diào)試。

四、總結(jié)和展望

未來還將從以下方面著手，進(jìn)行實(shí)時(shí)湖倉的優(yōu)化：

1. Auto Optimize Service

• 冷熱分離降本提效
• 物化視圖提速
• AE 服務(wù)智能化感知
• Compaction 能力打磨
• 更多 Transform UDF Partition Pruning 優(yōu)化

2. 主鍵表優(yōu)化

拓展 Deletion Vector，解決謂詞下推必須聯(lián)合去重的性能問題

3. AI 探索

• 落地適合模型訓(xùn)練的湖倉格式。
• 探索實(shí)現(xiàn)分布式 dataFrame，整合 metadata 和引擎。