歡迎關(guān)注公眾號“Tim在路上”
今天來閑談下數(shù)據(jù)湖三劍客中的iceberg。
Iceberg項目2017年由Netflix發(fā)起， 它是在2018年被Netflix捐贈給Apache基金會的項目。在2021年Iceberg的作者Ryan Blue創(chuàng)建Tabular公司，發(fā)起以Apache Iceberg為核心構(gòu)建一種新型數(shù)據(jù)平臺。
Ryan Blue 認(rèn)為我們不是齒輪——我們是工匠，Iceberg的哲學(xué)的核心是讓人們快樂：**數(shù)據(jù)基礎(chǔ)設(shè)施應(yīng)該在沒有令人不快的意外的情況下工作。
Iceberg最初的功能相比Delta或Hudi少一些，但是得益于底層架構(gòu)接口設(shè)計的優(yōu)雅通用，因此其較早的實現(xiàn)了Flink的讀寫，在國內(nèi)也獲得了不少的關(guān)注。今天就來談下Iceberg的優(yōu)勢與原理。
Hive數(shù)倉遇到的問題
首先我們回到Ryan Blue創(chuàng)建Iceberg的原因。起初是認(rèn)識到數(shù)據(jù)的組織方式（表格式）是許多數(shù)據(jù)基礎(chǔ)設(shè)施面臨挫折和問題的共同原因——這些問題因Netflix運行在 S3上的云原生數(shù)據(jù)平臺而加劇。
例如如果沒有原子提交，對 Hive 表的每次更改都會冒著其他地方出現(xiàn)正確性錯誤的風(fēng)險，因此自動化的修復(fù)問題也就是白日夢，很多維護(hù)工作留給了數(shù)據(jù)工程師，讓人不快樂。
所以說在Iceberg創(chuàng)建初期，它最核心希望解決的是Hive數(shù)倉遇到的問題。
具體來說，主要包括下面這些問題：
- 沒有acid保證，無法讀寫分離
- 只能支持partition粒度的謂詞下推
- 確定需要掃描哪些文件時使用文件系統(tǒng)的list操作
- partition字段必須顯式出現(xiàn)在query里面
1. 沒有acid保證
由于Hive數(shù)倉只是文件系統(tǒng)上一系列文件的集合（單純的采用目錄方式進(jìn)行管理），而數(shù)據(jù)讀寫只是對文件的直接操作，沒有關(guān)系型數(shù)據(jù)庫常有的事務(wù)概念和acid保證，所以會存在臟讀等問題。
2. partition粒度的謂詞下推
Hive的文件結(jié)構(gòu)只能通過partition和bucket對需要掃描哪些文件進(jìn)行過濾，無法精確到文件粒度。所以盡管parquet文件里保存了max和min值可以用于進(jìn)一步的過濾（即謂詞下推），但是Hive卻無法使用。
3. 文件系統(tǒng)的list操作
Hive在確定了需要掃描的partition和bucket之后，對于bucket下有哪些文件需要使用文件系統(tǒng)的list操作，而這個操作是O(n)級別的，會隨著文件數(shù)量的增加而變慢。特別是對于像s3這樣的對象存儲來說，一次list操作需要幾百毫秒，每次只能取1000條記錄，對性能的影響無法忽略。
4. query需要顯式地指定partition
在 Hive 中，分區(qū)需要顯示指定為表中的一個字段，并且要求在寫入和讀取時需要明確的指定寫入和讀取的分區(qū)。Iceberg將完全自行處理，并跳過不需要的分區(qū)和數(shù)據(jù)。在建表時用戶可以指定分區(qū)，無需為快速查詢添加額外的過濾，表布局可以隨著數(shù)據(jù)或查詢的變化而更新。
在上述例子中，Hive 表并不知道event_date 和event_time的對應(yīng)關(guān)系，需要用戶來跟蹤。
而在 Iceberg 中將分區(qū)進(jìn)行隱藏，由 Iceberg 來跟蹤分區(qū)與列的對應(yīng)關(guān)系。在建表時用戶可以指定date(event_time) 作為分區(qū)， Iceberg 會保證正確的數(shù)據(jù)總是寫入正確的分區(qū)，而且在查詢時不需要手動指定分區(qū)列，Iceberg 會自動根據(jù)查詢條件來進(jìn)行分區(qū)裁剪。
一種開放的表格式
上面講了創(chuàng)建Iceberg最初想要解決的問題，下面我們說下Iceberg的定位是什么，以及它在數(shù)據(jù)湖架構(gòu)中的位置。
Iceberg 的核心開發(fā)者Ryan Blue，將Iceberg定義為一種開放式的表格式為大數(shù)據(jù)分析，它的定位是在計算引擎之下，又在存儲之上，將其稱之為table format。
在大數(shù)據(jù)時代數(shù)據(jù)的存儲格式早已經(jīng)發(fā)生了翻天覆地的變化，從最初的txt file , 到后來的Sequence file , rcfile以及目前的parquet、orc 和 avro 等數(shù)據(jù)存儲文件。數(shù)據(jù)的存儲有了更好的性能、更高的壓縮比，但是對于數(shù)據(jù)的組織方式依然沒有太大的變化。目前Hive對于數(shù)據(jù)組織的方式任然是采用文件目錄的方式進(jìn)行組織方式，這種組織方式面臨上一節(jié)中遇到的問題。
Apache Iceberg is an open table format for huge analytic datasets. Iceberg adds tables to Presto and Spark that use a high-performance format that works just like a SQL table.
從上圖可以看出，Iceberg是在HDFS或S3存儲引擎上的又一層，用于管理在存儲引擎中的Parquet、ORC和avro等壓縮的大數(shù)據(jù)文件，使這些文件更便于管理維護(hù)，同時為其構(gòu)造出相應(yīng)的元數(shù)據(jù)文件。其上層是對接用于計算的Spark、Presto和Flink等計算引擎，并為其提供靈活的可插拔性。
自下而上的元數(shù)據(jù)
那么Iceberg是如何組織數(shù)據(jù)與元數(shù)據(jù)的呢？
在數(shù)據(jù)存儲層面上，Iceberg是規(guī)定只能將數(shù)據(jù)存儲在Parquet、ORC和Avro文件中的。像 Parquet 這樣的文件格式已經(jīng)可以讀取每個數(shù)據(jù)文件中的列子集并跳過行。
因此，如果可以跟蹤表中的每個數(shù)據(jù)文件，分區(qū)和列級指標(biāo)的主要信息，那么就可以根據(jù)數(shù)據(jù)文件的統(tǒng)計信息來更有效的進(jìn)行Data skip。
在Iceberg中對于每個數(shù)據(jù)文件，都會存在一個manifest清單文件來追蹤這個數(shù)據(jù)文件的位置，分區(qū)信息和列的最大最小，以及是否存在 null 或 NaN 值等統(tǒng)計信息。每個清單都會跟蹤表中的文件子集，以減少寫入放大并允許并行元數(shù)據(jù)操作。
每個清單文件追蹤的不只是一個文件，在清單文件中會為每個數(shù)據(jù)文件創(chuàng)建一個統(tǒng)計信息的json存儲。這樣可以使用這些統(tǒng)計信息檢查每個文件是否與給定的查詢過濾器匹配，如果當(dāng)前查詢的信息并不在當(dāng)前數(shù)據(jù)的范圍內(nèi)，還可以實現(xiàn)File skip， 避免讀取不必要的文件。
如下圖所示，每個清單文件追蹤多個數(shù)據(jù)文件，這樣的優(yōu)點是減少了元數(shù)據(jù)小文件的生成，同時可以允許跳過整個清單文件以及其關(guān)聯(lián)的數(shù)據(jù)文件。
在元數(shù)據(jù)層面上，Iceberg 將某個版本或快照的清單文件存貯在清單文件列表中，即manifest-list中。其是manifest-list并不是單獨的文件，而是snapshot快照文件中的一個list。
從manifest-list清單文件列表中讀取清單時，Iceberg 會將查詢的分區(qū)謂詞與每個分區(qū)字段的值范圍進(jìn)行比較，然后跳過那些沒有任何范圍重疊的清單文件。元數(shù)據(jù)中的min-max索引對查找查詢文件所需的工作量產(chǎn)生了巨大影響。當(dāng)表增長到數(shù)十或數(shù)百 PB 時，可能會有數(shù) GB 的元數(shù)據(jù)，如果對元數(shù)據(jù)進(jìn)行暴力掃描將需要長時間的等待作業(yè)——相反，使用min-max索引構(gòu)建的元數(shù)據(jù)存儲使得Iceberg 會跳過大部分。
回過頭來，我們在來看下Iceberg在其中是如何維護(hù)分區(qū)信息的。
Iceberg和Hive不同的是，Iceberg不是通過list出目錄來跟蹤分區(qū)和定位文件的。從上面的元數(shù)據(jù)文件可以看出，Iceberg的清單文件中會記錄每個數(shù)據(jù)文件所屬的分區(qū)值信息，同時在清單列表中會記錄每個清單文件的分區(qū)信息。除此以外在Iceberg的數(shù)據(jù)文件中也會存儲分區(qū)列的值，以進(jìn)行自動分區(qū)轉(zhuǎn)換的實現(xiàn)。
總而言之，Iceberg采用的是直接存儲分區(qū)值而不是作為字符串鍵，這樣無需像 Hive 中那樣解析鍵或 URL 編碼值，同時利用元數(shù)據(jù)索引來過濾分區(qū)選擇數(shù)據(jù)文件。
綜上，每次進(jìn)行數(shù)據(jù)的增刪改都會創(chuàng)建一系列的Data file 或 Delete file數(shù)據(jù)文件, 同時會生成多個追蹤和記錄每個數(shù)據(jù)文件的manifest file清單文件，每個清單文件中可能會記錄多個數(shù)據(jù)文件的統(tǒng)計信息；這些清單文件會被匯總記錄到snapshot文件中的manifest list清單文件列表中，同時在快照文件中記錄了每個清單文件的統(tǒng)計信息，方便跳過整個清單文件。而每次操作都會重新復(fù)制一份metadata.json 的元數(shù)據(jù)文件，文件匯總了所有快照文件的信息，同時在文件中追加寫入最新生成的快照文件。
高性能的查詢
Iceberg表格式的最主打的賣點正是其更快的查詢速度。
在Iceberg中自上而下實現(xiàn)了三層的數(shù)據(jù)過濾策略，分別是分區(qū)裁剪、文件過濾和RowGroup過濾。
分區(qū)剪裁：對于分區(qū)表來說，優(yōu)化器可以自動從where條件中根據(jù)分區(qū)鍵直接提取出需要訪問的分區(qū)，從而避免掃描所有的分區(qū)，降低了IO請求。Iceberg支持分區(qū)表和隱式分區(qū)技術(shù)，所以很自然地支持分區(qū)裁剪優(yōu)化。
如上一節(jié)所示，Iceberg實現(xiàn)分區(qū)剪枝并不依賴文件所在的目錄，而是利用了Iceberg特有的清單文件實現(xiàn)了一套更為復(fù)雜的分區(qū)系統(tǒng)及分區(qū)剪枝算法，名為Hidden Partition。首先每個snapshot中都存儲所有manifest清單文件的包含分區(qū)列信息，每個清單文件每個數(shù)據(jù)文件中存儲分區(qū)列值信息。這些元數(shù)據(jù)信息可以幫助確定每個分區(qū)中包含哪些文件。
這樣實現(xiàn)的好處是：1. 無需調(diào)用文件系統(tǒng)的list操作，可以直接定位到屬于分區(qū)的數(shù)據(jù)文件。2. partition的存儲方式是透明的，用戶在查詢時無需指定分區(qū)，Iceberg可以自己實現(xiàn)分區(qū)的轉(zhuǎn)換。3. 即使用戶修改分區(qū)信息后，用戶無需重寫之前的數(shù)據(jù)。
文件過濾：Iceberg提供了文件級別的統(tǒng)計信息，例如Min/Max等?？梢杂脀here語句中的過濾條件去判斷目標(biāo)數(shù)據(jù)是否存在于文件中。
Iceberg利用元數(shù)據(jù)中的統(tǒng)計信息，通過Predicate PushDown(謂詞下推)實現(xiàn)數(shù)據(jù)的過濾。
在講Iceberg前我們先來說下Spark是如何實現(xiàn)謂詞下推的：
在SparkSQL優(yōu)化中，會把查詢的過濾條件，下推到靠近存儲層，這樣可以減少從存儲層讀取的數(shù)據(jù)量。其次在真正讀取過濾數(shù)據(jù)時，Spark并不自己實現(xiàn)謂詞下推，而是交給文件格式的reader來解決。例如對于parquet文件，Spark使用PartquetRecordReader或VectorizedParquetRecordReader類來讀取parquet文件，分別對于非向量化讀和向量化的讀取。在構(gòu)造reader類時需要提供filter的參數(shù)，即過濾的條件。過濾邏輯稍后由RowGroupFilter調(diào)用，根據(jù)文件中塊的統(tǒng)計信息或存儲列的元數(shù)據(jù)驗證是否應(yīng)該刪除讀取塊。（Spark在3.1 支持avro, json, csv的謂詞下推）
相比于Spark, Iceberg會在snapshot層面，基于元數(shù)據(jù)信息過濾掉不滿足條件的data file。
RowGroup過濾：對于Parquet這類列式存儲文件格式，它也會有文件級別的統(tǒng)計信息，例如Min/Max/BloomFiter等等，利用這些信息可以快速跳過無關(guān)的RowGroup，減少文件內(nèi)的數(shù)據(jù)掃描。
Iceberg在data file層面過濾掉不滿足條件的RowGroup。這一點和Spark實際是類似的，但是作為存儲引擎的Iceberg，他使用了parquet更偏底層的ParquetFileReader接口，自己實現(xiàn)了過濾邏輯。
Iceberg通過調(diào)用更底層的API, 可以直接跳過整個RowGroup, 更進(jìn)一步的減少了IO量。
今天我們先簡單介紹了Iceberg, 后續(xù)再通過源碼去了解Iceberg是如何實現(xiàn)upsert, delete 以及如何與Spark進(jìn)行整合的。
