一、背景

我們使用 Iceberg 構建湖倉一體平臺的初衷是希望解決業務方在使用 Hive 數倉時的一些痛點。主要包括以下幾大方面：

（1）Hive 的查詢性能達不到交互式分析的要求，所以經常需要把 Hive 的數據儲存到其它引擎當中。

（2）上一點造成了出倉鏈路越來越多，越來越復雜，維護成本高。

（3）另外，出倉的數據容易形成數據孤島，造成數據冗余，導致存儲成本上漲。

（4）最后，Hive 的時效性不好，即使用 FIink 流式的引擎寫入，延遲也會在小時級別。

我們希望我們的湖倉一體平臺能夠解決這些痛點，我們的目標是：

（1）首先，平臺要是互聯互通的，要支持各種引擎的訪問，避免數據孤島的出現。

（2）第二，查詢要高效，以滿足交互式分析的要求。

（3）第三，使用要盡可能的便捷，盡可能降低業務方的門檻。

我們的湖倉一體架構如上圖所示，采用 Iceberg 來存儲數據，數據是在 HDFS 上。入湖的幾條鏈路包括 FIink、Spark 引擎來寫入，也提供 java 的 API，業務方可以直接通過 API 來寫入數據，后臺有一個叫做 Magnus 的服務對 Iceberg 的數據進行不斷的優化。另外我們也用 Alluxio 來對數據進行緩存加速。我們使用 Trino 來進行交互式分析，對外提供查詢接口。寫入 Iceberg 的數據有一部分是要繼續寫入下游的 Iceberg 表。一般是數倉的分層建模的場景。雖然我們減少了 Hive 出倉的鏈路，但是有一些場景可能 Trino 的查詢還是達不到響應時間的要求。比如毫秒級的響應，可能還是會出倉到 ClickHouse、ES 等其它存儲中。

下面簡單介紹一下 Iceberg 的表結構，以及我們為什么選 Iceberg 作為存儲格式。

Iceberg 有文件級別的元數據管理。它基于 snapshot 來做多版本的控制。每一個 snapshot 對應一組 manifest，每一個 manifest 再對應具體的數據文件。我們選 Iceberg 的一個比較重要的原因是其開放的存儲格式。它有著比較好的 API 和存儲規范的定義，方便我們在后續對它做一些功能上的擴展。

二、查詢加速

接下來介紹我們目前的一些比較重要的工作。其中最核心的一項是查詢加速。

因為我們面對的是 OLAP 的場景，一般是會有過濾條件的。所以我們第一個思路是如何盡可能過濾掉不需要掃描的數據。Iceberg 在文件級別記錄了每一個列的一些統計信息，比如說 MinMax 值，這些統計可以在查詢計劃階段就把一些不需要的文件過濾掉。我們很直觀的一個想法是，如果對數據進行排序，就會讓相同的數據有更好的聚集效果，在過濾的時候就會過濾掉更多的文件。

所以我們最早是做了多維的排序。過濾字段可能有多個，不能用簡單的線性排序來做。線性排序只對靠前的排序字段有比較好的聚集效果。所以我們比較了 Z-ORDER 和 Hibert Curve 這兩種排序方式。從多維排序的實現來比較，發現 Hibert 的聚集性會更好一點。所以我們目前都是采用 Hibert 的方式。不管是 Z-ORDER 還是 Hibert ，都要求參與排序的字段是一個整型值。對于非整型的數據，我們用 Boundary Index 的方式來參與計算。

我們會把數據按照需要多少區間，來切出不同的 Boundary。根據它的 Boundary Index 來參與 Z-ORDER 和 Hibert Curve 的計算。

有了排序以后，另一個問題是多維的排序字段是不可以無限增加的。一般來說排序字段的個數越多，其聚集效果會越差。我們對業務方的建議是一般不要超過四個排序字段。如果有更多的過濾字段怎么辦？我們考慮到對于一些基數比較高的過濾字段，不去做排序，而是通過創建索引的方式，也能有一個比較好的過濾效果。

我們實現的索引是為了判斷一個數據文件是否滿足查詢條件的要求。所以我們的索引是文件級別的，一個表可以針對不同的列創建不同的索引。一個 DataFile 可能會關聯多個索引文件，我們把索引文件和 DataFile 的元數據一起存儲在 manifest 里。

下面介紹一下我們支持的索引種類：

（1）BloomFilter：計算比較簡單，占用空間也比較小。存在 false positive 的問題，只支持等值的查詢。

（2）Bitmap：功能更強大，支持等值和范圍查詢，匹配更精準，更精準是因為可以對多個條件匹配到的數據進行交并補計算，同時它返回的行號也可以幫助進一步 skip 數據。Bitmap 的缺點是占用空間比較大，尤其是對一些高基數的字段，創建 Bitmap 索引，可能加載索引的時間已經超過了過濾掉數據所節約的時間，甚至會產生一些負向的效果。

（3）BloomRF：我們參考一篇論文，實現了一種 BloomRF 索引，它與 BloomFilter 的原理類似，但是用了多段的有序哈希函數來支持等值和范圍的查詢。它的存儲開銷也與 BloomFilter 類似。其問題也是會有 false positive。

（4）TokenBloomFilter、NgramBloomFilter，TokenBitmap、NgramBitmap：是針對 token 的索引，是為日志場景設計的。相當于對日志做一些分詞的操作。分詞完成以后，構建 BloomFilter 或者 Bitmap 這樣的索引。TokenBloomFilter 和 TokenBitmap 針對的是英文的分詞，Ngram 針對的是中文的分詞。

除了索引以外，我們也在做對預計算的支持，內部叫做 Cube，或者 AggIndex，是針對聚合計算的加速。目前支持單表和星型模型的查詢。一個 Cube 的定義，主要定義兩個信息：一個是 Cube 的維度字段；另一個是 Cube 需要的聚合計算，常見的如 count、min、max、count distinct 等都是支持的。另外聚合是做在文件級別的。

舉一個例子：

它是一個星型模型，lineorder 表是事實表，會關聯 dates 、part 和 supplier 維表。如果要對這樣一個查詢場景去定義 Cube，所有需要在 group by 、where 語句中使用的字段都要作為維度字段。大家可以看到預計算是定義在事實表上的。它的預計算的定義是跟 lineorder 表關聯的。但是這里使用到的一些列可能是有維表當中的列。我們做了一個叫做關聯列的實現。事實表不僅可以用關聯列來定義 Cube，同時也能用關聯列對事實表的數據來進行排序和索引。像查詢里，p_brand 上有一個過濾條件，Cube 數據也可以用到索引來進行過濾。上面的過濾條件也可以用來過濾事實表的數據。

定義了 Cube 以后，Magnus 服務會在后臺去負責 Cube 文件的生成。因為是文件級別的聚合，所以生成的邏輯是每一個文件會去關聯其他的文件。比如這是事實表當中的一個 DataFile，它會去關聯三張維表。這三張維表關聯完以后會計算聚合值，最終會生成一個 CubeFile。CubeFile 與索引的情況類似，它會跟 DataFile 關聯起來，一起保存在 Manifest 當中。

對聚合值的處理，因為我們做的是文件級別的聚合。所以真正查詢的時候，還需要把文件級別的聚合再做 global merge, 才能得到最終的一個聚合效果。這里分兩種情況：

一種是可以直接累加的一些聚合值，如 min、max、count，在生成 Cube 文件的時候，可以直接存儲聚合結果；有一些不能直接累加，比如 Average，存儲的是中間狀態。查詢時需要判斷能否用 Cube 來響應，比如下圖中展示的查詢：

它是一個原始的邏輯計劃。我們會去找查詢當中的 aggregation 節點。對于 aggregation 節點，判斷其 source 表中是否存在一個 Cube 能滿足聚合計算的要求。如果找到，會把邏輯計劃進行轉換。轉換完以后，原來的 table scan 就會切換成 Cube 模式，就不去讀原始的數據了，而是去讀 Cube 文件的數據。因為 Cube 文件是異步生成的，所以就肯定會存在一種情況，可能有一些文件已經構建了 Cube，有一些文件可能還沒有生成 Cube。查詢改寫這一側會稍微有一點不一樣。對于這種情況，我們的處理思路是把有 Cube 的部分，保持跟原來一樣的改寫方式；沒有 Cube 的部分，現場把 Cube 的數據算出來，與已有 Cube 的數據做一次 union 以后，再做 global merge，這樣可以得到一個最終的結果。

當然這個做法只適用于只有少量文件還沒有 Cube 的情況。如果大部分文件都沒有 Cube，那么直接退化成原始的計算會更好。

Cube 做好之后，我們目前在探索用 star-tree index 對 Cube 來做一個增強。我們參考了 Apache Pinot 的實現。

要解決的問題是，Cube 是可以響應不同的維度組合的。比如 Cube 的定義可能選了三個維度，查詢的時候只用到了其中的兩個或者一個，Cube 也是可以響應的。所以從節省存儲的角度來說，用最細粒度的維度來定義 Cube。這樣只需要一個 Cube，就可以響應所有維度組合的查詢。

但是如果維度選的比較多，生成的 Cube，它的數據量也會比較大。而且維度多了以后，聚合效果會變差。如果用最細粒度定義的 Cube，去響應很少維度的查詢，中間還需要額外做很多聚合的計算。

如果針對每一個查詢都去定義特定的 Cube，可以保證查詢的時候 Cube 一定是最優的。但是它的問題是所需要的存儲成本就會比較高，所有不同的組合，都要實現，生成不同的 Cube 文件。

Star-Tree Index 希望在兩者之間做一個折中。針對我們的 Cube 生成 Star-Tree Index 這樣一個數據結構。

舉一個例子，比如我的 Cube 的定義是 Dim1、Dim2、Dim3 這三個字段，聚合值是 count。雖然維度一共有三個，但是常用的可能是 Dim1、Dim2 這兩個。這時候就可以按照 Dim1、Dim2 指定這兩個維度字段來生成 star tree。star tree 是一個多叉樹，每一層對應一個維度。每一層的節點是當前這一個維度的取值。比如 Dim1 的取值是 1、2、3，Dim2 的取值是 a、 b 、c。Star-Tee Index 會針對不同的取值來構造樹的節點。每一層還會有一個特殊的 star 節點，star 節點的含義是忽略掉這一層的取值，或者我們認為 star 是一個通配符。全部聚合在一起以后，它的聚合的結果是多少。對于 star 節點，會額外生成一些 star record。star 節點下面的這些節點都會生成具體的一個 star record。比如例子里面，Dim1 取值是 “*” 的時候，Dim2 可能有 a、d 這兩種。如果查詢當中只用到了 Dim2 這一個維度，那么可以通過 star record 來進行響應。因為我只需要考慮 Dim1 為 “*” 的情況。

三、智能優化

介紹完查詢加速以后，再來講一下我們目前做的智能優化的一些工作。

針對的是我們的 Magnus 服務。我們最根本的目標是希望盡可能降低用戶的使用門檻。比如 Hive 用戶，他可能需要了解一些大數據的原理；小文件多了，應該怎么處理，可能需要做一些合并；Hive 表應該怎么做分桶，文件內部怎么做排序。我們目前所處的一個階段，叫做自動化的階段。用戶不需要知道這么多底層的知識。但是他還是需要告訴我一些業務上的邏輯。比如常用的過濾字段是哪些，常用的聚合的模型是什么樣子的。我們再根據用戶提供的信息來自動幫他去創建索引，去創建 Cube。

最終我們是希望進一步簡化，用戶只是建表。表在建出來的使用過程當中，我們可以對它做一個智能的持續的優化。Magnus 服務就是以此為目的來開發的。它主要負責的功能包括：

（1）一個是自動的后臺優化，目前所有 Iceberg 表的寫入操作，Magnus 都會監聽，當監聽到寫入事件后，它會根據自己內部的一些調度邏輯，通過 spark 任務對表進行一些操作，比如排序、創建索引、構建 Cube 等。

（2）另一個比較重要的功能是，它可以幫助我們把 Iceberg 表的一些詳情做一個圖形化的展示，便于我們定位和排查問題。比如下圖中顯示的一張 Iceberg 表。

可以看到表是定義了排序字段的，在界面上可以看到它某一個分區下有多少個文件，這些文件有哪些已經按照用戶的要求做了排序，有哪些已經按照用戶的要求去構建了索引等等。

（3）第三個功能是智能化的推薦。實現方式是使用 Trino 把查詢明細全部落庫。

查詢明細當中包含了每張表用到的過濾字段，Magnus 服務會去定期去分析這些查詢明細，結合用戶的歷史查詢以及 Iceberg 表本身的統計信息。當然有一些統計信息可能是需要用 Trino 去現場計算出來的。結合這些信息以后，會給出一些優化建議。

上面的例子展示的是 Magnus 對某一張表的一次優化建議?？梢钥吹奖砝锩嬗脩粼臼嵌x了排序和索引字段的。Magnus 分析結果來看，首先是排序可以增加幾個字段，同時可以刪掉一些不必要的字段。索引也是可以去掉一些用不到的索引。后續我們會考慮根據推薦去驗證效果。如果效果好，后面可以考慮去自動幫助用戶進行修改。

四、現狀

最后來介紹一下我們目前落地的情況。

目前主要場景包括 BI 報表、指標服務、A/B Test、人群圈選和日志等。

Iceberg 表總量大約為 5PB，日增 75TB。Trino 查詢每天在 20 萬左右，P95 的響應時間是 5 秒。我們給自己的定位為秒級到 10 秒級。過濾的數據量（估算）為 500TB/ 天，占比約 100%~200%。