Apache Iceberg 中引入索引提升查詢性能

火山引擎EMR團隊 位元組跳動技術團隊 2023-05-17 12:04 發表于北京

Apache Iceberg 是一種開源資料 Lakehouse 表格式，提供強大的功能和開放的生态系統，如：Time travel，ACID 事務，partition evolution，schema evolution 等功能。

本文将讨論火山引擎EMR團隊針對 Iceberg 元件的優化思路，通過引入索引來提高查詢性能。

采用 Iceberg 建構資料湖倉

火山引擎 E-MapReduce（簡稱 EMR）是火山引擎數智平台（VeDI）旗下的雲原生開源大資料平台産品， 提供了企業級的 Hadoop、Spark、Flink、Hive、Presto、Kafka、StarRocks、Doris、Hudi、Iceberg 等大資料生态元件，100% 開源相容，可以幫助企業快速建構企業級大資料平台，降低運維門檻。秉承業界領先的 EMR Stateless 理念，火山引擎 EMR 可以實作叢集級别的彈性伸縮，即無業務需求時釋放叢集，有業務需求時再拉起叢集，配合智能化的冷熱資料分層存儲能力，助力企業在大資料基建領域進一步降本提效。

基于火山引擎 EMR 産品，可以建構資料湖倉、近實時數倉、實時數倉等場景。例如，使用 Iceberg 建構資料湖倉，從 ODS 到 DWD 等不同的分層進行模組化，将資料 HFDS 或 TOS（火山引擎對象存儲産品）上，然後采用 Trino 或者 Spark 去做分析。

如何加速查詢性能，使其盡可能接近專門的分布式數倉（如 ClickHouse 等），是需要思考和探究的問題。

索引是業界常用的提高查詢性能的手段之一，針對 Iceberg 我們也采用了增加索引的方式。對常用的列字段建構 Index，在進行 table scan 時利用 Index 隻傳回比對的資料，降低比對資料量，進而大大提高查詢性能。

Iceberg 介紹

介紹 Iceberg Index 功能之前，我們先簡單介紹下 Iceberg 的架構。Iceberg 具有分層的中繼資料架構，如下如所示。

Spark、Presto、Flink 等多種引擎讀取 Iceberg 的資料，就是利用分層的中繼資料找到 data file 清單。例如，Spark 引擎解析 SQL 語句，然後調用 Iceberg 的接口，擷取 data file 并進行 task 切分。

在 Manifest file 中記錄了 data file 中字段的最大值和最小值。

"data_file": {
        "content": 0,
        "file_path": "hdfs://emr-cluster/warehouse/hive/db.db/sample/data/ts_day=2020-12-31/category=diamond/00000-0-220aa9a6-4530-499f-9450-da946d667624-00001.parquet",
        "file_format": "PARQUET",
        ......
        "lower_bounds": {
                "array": [{
                        "key": 1,
                        "value": "\u0006\u0000\u0000\u0000"
                }, {
                        "key": 2,
                        "value": "diamond"
                }, {
                        "key": 3,
                        "value": "\u0000\u0004܍ŷ\u0005\u0000"
                }]
        },
        "upper_bounds": {
                "array": [{
                        "key": 1,
                        "value": "\u0007\u0000\u0000\u0000"
                }, {
                        "key": 2,
                        "value": "diamond"
                }, {
                        "key": 3,
                        "value": "\u0000¨odÆ·\u0005\u0000"
                }]
        },
        ......
}
           

利用這些資訊，可以進行 data file 級别的初步過濾，把不符合條件的 data file 過濾掉，進而減少一部分資料的讀取。

實作索引的必要性

既然 Iceberg 已經提供 data file 級别的過濾。為什麼我們還需要引入索引呢？以下面例子進行介紹，左邊兩個表格分别是 data file 檔案裡面的内容，右邊表格是 data file 對應的 manifest file。

針對SELECT * FROM table WHERE age > 50，利用 min-max 統計資訊，很容易發現 data file 1 中沒有滿足條件的資料，是以 data file 1 就不會參與計算。

但是針對多元分析，如name = 'LiLy' AND age > 30，利用name和age的min-max的統計資訊分别對條件name = 'LiLy'和age > 30進行判斷，得到 data file 1 和 data file 2 都滿足條件。然而，仔細分析 data file 1 和 data file 2 的資料，并不存在符合條件的資料，是以 min-max 過濾效果不太理想。是以通過引入合适的索引功能，可以提高 data skipping 的機率，提高查詢性能。

1. 首先探究索引類型

索引類型有多種，如 BloomFilter、Ribbon Filter、Dictionary Index、BitMap 等。為了滿足多元分析場景，我們選擇了[Range-Encoded BitMap]https://www.featurebase.com/blog/range-encoded-bitmaps （ Base-2, Bit-sliced Index），可适用于高基數場景，滿足=、<、>、IN、BETWEEN 等操作的多元分析。

例如，對上面的 name 和 age 兩列分别計算索引資訊。由于 name 屬于字元串類型，需要先進行字典編碼再進行計算索引資訊。采用 Range-Encoded 技術，根據資料的二進制相關資訊以及對應的 pos 資訊生成索引資料。利用索引資料分析得到，同時滿足name = 'LiLy' 和age > 30的資料不在同一行，恰好可利用 Range-Encoded 的交并運算将資料進行過濾掉，是以 data file 1 不用參與計算。

也就是說，BitMap 的交并運算可以更好地在複雜過濾條件的情況下過濾掉更多的資料檔案。

2. 接下來探究索引的粒度。

Iceberg 提供的 min-max，也是一種檔案級别的索引。檔案級别的索引就是根據 filter 條件過濾掉不符合條件的 data file。檔案級别的索引可适用于多種檔案類型，但這種粒度比較粗，隻要 data file 中有一條資料符合條件，該 data file 中的資料就會全部讀取出來參與計算，進而影響 SQL 的查詢性能。

對于 Parquet、ORC 的檔案格式，提供有 file chunk 的概念(row group or stripe)，我們完全可以按照 row group / stripe 粒度，對資料進行過濾。（為了友善描述，我們将 row group 和 stripe 統稱 split。）

如：SQL語句：SELECT * FROM table WHERE col_1> v1 AND col_2 = v2，其中對 col_1 字段和 col_2 字段已建構 Index 資訊。現在利用索引對 SQL 語句作用。

SQL 語句解析後，将符合條件的 data file 清單進行切分後，得到很多 split 的清單。利用索引，分析 split 中資料是否滿足條件，如果不滿足則跳過。如上圖 data file 清單切分後，得到數萬級别數量的 split 清單。将索引資料作用在 split1，發現 split1 中沒有同時col_1> v1 AND col_2 = v2滿足條件的資料，該 split1 中的資料就不會參與計算。最後處理後，隻得到了少量的 split 清單，資料過濾度達到 10% 以上，查詢性能有明顯提升。

是以，采用 row group / stripe 級别的細粒度索引，可以過濾大部分資料。

細粒度索引實作邏輯

Iceberg 中繼資料中 manifest file 中除了提供 min-max 等統計資訊，還提供有 split 相關資訊："split_offsets":{"array":[4,...]}，極大友善我們實作 row group / stripe 級别的細粒度索引。

1. 提供索引的建構 API

Iceberg 中提供建構索引的 API，引擎端調用該 API 即可實作索引建構功能。對于 Spark 3.3 及以上版本，已經提供有索引的 SQL 語句，在 Iceberg 的 Spark 子產品實作 Spark 提供的索引接口即可。

1. 建構索引

我們采用異步建構索引，不影響主線任務。也提供了增量建構索引功能，隻對 append 資料進行建構索引。調用 TableScan 讀取資料，按照 data file 的 split offset 切分資料，進行建構索引，并儲存索引資料和對應的中繼資料資訊。為了避免出現小檔案存在，我們會進行索引資料合并。

1. 索引檔案存儲

索引檔案格式采用[puffin]https://iceberg.apache.org/puffin-spec/格式，這是一種二進制格式。 Magic Blob₁ Blob₂ ... Blobₙ Footer

在 Footer 中儲存每個 blob 的中繼資料資訊。索引建構成功後，會生成類似于下面内容的檔案。

索引帶來的收益

Range-Encoded BitMap 适用于多元分析場景，且 Ranger 範圍較小時，效果非常明顯。下面我們基于 Spark 引擎性能測試。

1. 構造 1TB 的 SSB 測試資料，分别在建構 Index 前後，對以下用例進行測試。

Q1: SELECT count(*) FROM lineorder WHERE  lo_ordtotalprice = 19665277
Q2: SELECT count(*) FROM lineorder WHERE  lo_ordtotalprice = 19665277 AND lo_revenue  = 2141624
Q3: SELECT count(*) FROM lineorder WHERE  lo_ordtotalprice = 19665277 AND lo_revenue  >=10304000
Q4: SELECT count(*) FROM lineorder WHERE  lo_ordtotalprice = 21877827 AND lo_revenue  >= 83800  AND lo_revenue  <= 103800
Q5: SELECT count(*) FROM lineorder WHERE  lo_ordtotalprice > 21877827 AND lo_revenue  >= 83800  AND lo_revenue  <= 93800
Q6: SELECT count(*) FROM lineorder WHERE lo_ordtotalprice >= 93565 AND   lo_ordtotalprice < 93909
Q7: SELECT count(*) FROM lineorder WHERE   lo_ordtotalprice >= 93565 AND   lo_ordtotalprice < 91003562 AND lo_revenue    >=904300 AND lo_revenue    <= 9904300
           

左圖展示了 7 條 SQL 語句分别在沒有 Index 和采用 Index 情況下的執行時間。右圖展示采用 Index 後，7 條 SQL 語句讀資料的 split 數量。很明顯讀資料的 split 數量越少，Index 效果越好。最糟糕的情況，所有的 split 都參數計算，這時和沒有建構索引的效果類似。

1. 采用 SSB 基準測試

由于 SSB 提供的測試場景，和 Range-Encoded 有利的場景，不太比對，是以 Index 的效果并沒有明顯的效果。但也不會比不采用 Index 的效果差。如下面左圖，分别是建構索引前後，SQL 語句的執行時間，建構索引的優勢并沒有展現出來。右圖中，可以看到所有的 split 都參與了計算。

總結

根據上面的介紹，這裡總結下 Iceberg 中索引實作的一些特征：

• 細粒度索引級别：提供 RowGroup/Stripe 級别的索引，可以更加精确的定位資料的查詢範圍，減少不必要資料輸入，進而提高查詢性能；
• 索引作用于執行端：查詢任務被配置設定多個執行端，每個執行端隻判斷該節點上的 RowGroup/Stripe 資料是否符合即可；
• 适配多種引擎：索引建構後，可用于多種引擎；
• 提供異步建構 Index，進而不影響主業務的進行；
• 适用于高基數 & 低基數場景，且占有存儲空間小。滿足範圍查詢、等值查詢等場景。且範圍越小，收益效果越明顯。