一、愛奇藝 OLAP 簡介

首先簡單介紹一下愛奇藝 OLAP 的基本情況：

存儲方面，OLAP 目前支持三類存儲：

① 離線 HDFS：用于離線分析、批處理等場景；

② 實時 Kafka：用于實時分析、在線處理等場景；

③ 近實時 Iceberg：分鐘級延遲，是今天要重點介紹的數據湖產品。

存儲之上是查詢引擎，我們采用 SparkSQL 做 ETL 處理，采用 Trino 做 Ad-hoc 即席查詢，ClickHouse 用于查詢加速的場景。我們通過 Pilot 提供對外的統一查詢，支持各類應用場景。

二、為什么要數據湖

下面來介紹一下愛奇藝數據湖的建設背景。

1、數據湖技術加速數據流通

為什么要有數據湖？數據湖其實就是為了加速數據流通。

愛奇藝 Pingback 投遞的場景：

Pingback 是愛奇藝內部對端上埋點的習慣名稱，每個公司都會有類似的服務。在經典的 Lambda 架構解決方案里，Pingback 數據在投遞后，有離線和實時兩個通路。

離線通路寫到 HDFS 里，然后由離線開發平臺構建離線數倉。離線數倉的優點是成本很低，支持的容量也很大。缺點是延遲大，可能要 1 小時或者 1 天。為了解決這個時效性問題，往往會再構建一個實時數倉。通常用 Kafka 作為存儲，用 Flink 或者 Spark 這類的流計算任務處理 Kafka 數據，構建實時數倉。

實時數倉的延遲非常低，能做到秒級的延遲，但缺點是成本很高，只能放最近幾個小時的數據，要基于 Kafka 做明細查詢也是比較的困難的。

其實很多實時分析場景并不需要秒級的延遲，分鐘級的延遲就足夠了。譬如說廣告、會員的運營場景，或者監控大盤等。數據湖產品提供了性價比很高，容量很大的分鐘級延遲的解決方案。

2、Iceberg 定義-新型表格式

愛奇藝的數據湖選型用的是 Iceberg，Iceberg 是一種新設計的開源表格式，用于大規模數據分析。

① Iceberg 本質上不是存儲，因為它底層存儲復用了 HDFS，或者對象存儲。在存儲之上構建了 Iceberg 表級抽象，對標 Hive 的表設計。

② 它也不是查詢引擎或者流計算引擎，它支持各類計算引擎，比如 Hive、 Flink、 Spark，也支持各類的 SQL 查詢引擎。

（1）表格式-Hive 及其缺陷

為什么有了 Hive 表格式還要引入 Iceberg 表格式？ 

一個經典的 Hive 表可能會有天級分區、小時級分區，或者進一步的子分區。其設計核心是用目錄樹去組織數據，能夠很好地做分區級過濾。

但是它也有著以下缺點：

① 元數據統一存在 Metastore，通常底下是 MySQL，很容易成為瓶頸。

② 由于元信息是分區級別的，沒有文件級別的信息，因而當發起一個查詢時，制定執行計劃需要拿到分區下的文件列表。拿到文件列表本質上是對每一個分區請求 NameNode 做 List 請求。舉個例子，一天有 200 多個分區，查 7 天的數據，分區數就會非常多，會發起 O(N) 復雜度的 NameNode 的 List 請求調用，這個元數據的枚舉過程會非常的慢。

③ 由于它的最小單位是分區級別的，最大的原子操作就是分區級別的覆蓋，其他一些原子操作是不支持的。

（2）表格式-Iceberg

Iceberg 新定義的表結構有元數據層和數據層。數據層就是數據文件。元數據層是它很重要的設計點，可以復用 Hive 的 MetaStore，指向最新的快照。元數據里面分多層，記錄了具體的文件列表。

每次有新的 Commit，就會創建出新的快照，讀請求可以訪問舊的快照，寫請求寫新的。在寫的過程中，新創建的數據文件讀是不可見的，只有在提交后把最新的版本指過去，新寫入的文件才可見。做到了讀寫分離。同時修改操作是原子的，能夠支持細粒度的分區內部的修改。

（3）表格式-Hive VS Iceberg

簡單比較一下 Hive 和 Iceberg：兩者底層都采用 HDFS 或者對象存儲，都是 PB 級的廉價存儲方案。區別 Hive 元信息是分區級，Iceberg 是文件級。比如 Hive 分區原本有 100 個文件，加了 5 個文件，那么 Hive 下游任務就需要重新計算 Hive 分區下的全部數據。Iceberg 能夠獲取到修改的 5 個文件，可以做增量的下游計算。

時效性是 Iceberg 很明顯的優勢，能夠做到近實時，比如 5 分鐘級，如果每分鐘提交一次則可以做到分鐘級。

制定執行計劃時，Iceberg 是常數級的，它只讀取固定的元數據文件就能夠拿到文件列表。

Iceberg 還支持文件級別的過濾，比如基于統計信息或者字典做過濾。

三、數據湖平臺建設

為了方便用戶使用，愛奇藝在引入數據湖以后，首先要做平臺化建設。

1、平臺總覽

這是愛奇藝數據湖整體的產品架構圖：

最底下是數據源，比如前面提到的 Pingback、用戶 MySQL 的 Binlog 解析、日志和監控信息，會分別進到實時、離線和 Iceberg 通道。在 Iceberg 之上，通過 RCP 平臺、Babel 平臺分別做流式入湖和離線入湖。使用 Trino 和 Spark SQL 去做查詢。同時我們開發了數據湖平臺去完成元數據管理、權限管理等等。

2、流式入湖

愛奇藝通過實時計算平臺，能夠做到很簡單的入湖。一個 Kafka 的數據只需要三步，就可以完成配置流任務：首先配置從哪個 Kafka 開始讀；然后在里面做 Transform 邏輯，比如篩選、重命名，最后定義寫到哪個 Iceberg。

3、出湖查詢

入湖的下一步是查詢，也就是出湖。目前 Iceberg 有兩類文件格式，V1 格式支持 Append Only 數據，不支持行級修改。Iceberg 發布的最新版本 V2 格式能支持行級更新。

目前 V1 格式是通過 Trino 引擎查詢，V2 格式通過 SparkSQL 查詢。前端是通過 Pilot，我們的自研 SQL 引擎做分發，能夠基于文件格式自動地選擇引擎，支持各類用戶場景。

四、性能優化

下面介紹一些性能優化的工作。

1、小文件

說到數據湖，無論哪個產品都繞不開的一個問題就是小文件問題。Hive 可以批量，比如每小時做一次計算，可以寫出很大的文件。在 Iceberg 中，由于需要做到近實時，每分鐘或者每 5 分鐘寫文件，文件就比較小，必然會有小文件問題。我們主要通過兩個方面去解決小文件問題: 

（1）生命周期

根據表的生命周期做處理。比如一張表可能只需要保留一年，或者保留 30 天，歷史的數據可以刪除。

目前平臺會限制用戶建表必須配置生命周期，通過數據湖平臺自動地完成清理邏輯。

清理用的是 Iceberg 官方提供的解決方案，Spark 的 Procedure，先是 Drop 分區，然后 Expire 歷史的 Snapshot，再刪除孤兒文件，最后重寫元數據文件。

這套流程直接跑，有些環節是存在性能問題的，并不能夠滿足清理的效率：

① 第一：Spark 的使用模式，每次跑任務都需要提交一個 Spark 任務，需要先申請Yarn 資源，再啟動 Application，跑完這個任務后這個 Application 就釋放掉了。這里可以采用 Spark 的常駐模式，生命周期清理 SQL 可以跑得很快， 資源是不釋放的，避免了申請和啟動的耗時。

② 第二：天級的目錄刪除，Iceberg 官方的實現是比較慢的。它用的是孤兒文件刪除的策略，在文件數比較多的時候，掃描過程比較慢。我們做了改進，因為明確知道整個天級目錄都不需要，可以直接刪除整個目錄。

③ 第三：我們添加了回收站的機制，生命周期誤刪除時能有恢復的手段。

做了這些優化以后，線上大概幾千個表，都能夠按時完成生命周期的清理。比如 Venus 庫原先可能有 2 億個 iNode，清理完以后穩定在 4000 萬的數量級。

（2）智能合并

另外一個處理小文件問題的方式就是合并。最簡單的就是配置一個定時合并。

人工配置定時合并比較大的問題是：定時策略比較難配置。比如，什么時機應該做合并，這次合并應該要合并什么范圍的數據，如果讓業務去配這些信息，每一個 Iceberg 用戶就需要非常深入地去理解小文件產生的機理才能夠比較好地控制合并的范圍。

為了解決這個問題，我們參考了 Netflix 的文章，做了智能合并，它的核心思想是：

不再由用戶指定合并行為，而是統計 Iceberg 表每個分區下面的文件數，計算均方差，再結合表的權重因子，算出來哪些表合并以后效果是最好的，添加到待合并的分區列表里面。然后由合并任務按照優先級完成合并過程，用戶無需做配置。

（3）合并性能優化

有了智能合并以后，還要解決合并的性能優化問題，我們也一直跟隨社區的發展。在使用過程中，最初 Iceberg 在文件合并這塊做得還不是很好。最早的時候，有個問題，Delete File 在合并以后并沒有被真正地刪除，目前已經修復。舉個例子，如果 Delete 以后馬上有個 Rewrite Data File，那么相應的 Delete File 是不會被刪除的。這個問題目前有一些解決方案，但最標準的解決方案，社區還在跟進當中。

還有一些大表合并任務經常失敗。這里我們可以配置 Bucket 分區，將全表合并改為每次合并其中一個 Bucket 分區，減少單次合并的數據量。

還可以應用 Binpack 合并策略去控制合并選擇的邏輯。應用 Bucket 分區和 Binpack合并策略以后，如右上示意圖體現的是文件數的變化，可以判斷這個文件數一直在增長，這個小的下降是小時級分區合并，到一定時間做全表合并，它的文件數據減少得比較多，存在周期性的震蕩。

還有一個例子，我們發現在做合并的時候經常會和寫入任務沖突，會報一個錯誤，要合并的這個文件有一個 Position Delete 在引用，其實是一個誤判，因為在社區的默認的參數里面，去判斷這個 Data File 有沒有被新的 Delete File 引用的時候，有Upper bound 和 Lower bound，但這兩個 Bound 被截取了，這個 Data File 其實沒有被引用，但截取以后它就在這個區間里面了，解決方法修改表屬性控制相應行為。

（4）寫入參數控制

前文介紹了當小文件已經產生的時候如何優化，但我們更希望小文件最好不要產生，在寫入的時候就把文件數控制住。我們需要去了解 Flink 任務寫入的時候是怎么控制文件數量的。

左上角示意圖中這個 Flink 任務有 100 個并行度，在默認參數 Distribution-mode = None 時每一個并行度都會往分區下寫文件，就會寫入 100 個文件，一分鐘寫 100 個文件每個數據文件都很小。

如果配置 Distribution-mode = Hash，如左下角的圖中，在寫入的時候會先做 Shuffle，基于 Partition Key Shuffle 到特定的 Sink，這個 Flink 任務會把數據都集中到一個 Sink，寫到一個文件，就解決了小文件問題。

但又會引入新的問題，數據量比較大的時候，單個任務寫文件的效率跟不上，就會造成 Flink 任務反壓。這個時候我們用哈希策略結合 Bucket 分區。比如，可以控制 1 個 Hour 下面 10 個 Bucket，通過兩者結合起來就可以很精確地去控制 1 個分區到底要生產多少個文件。一般建議寫入文件大概在 100 MB 左右是比較合適的。上圖的表格中列出了各個參數配置下的文件數量。

2、查詢優化

解決了小文件問題，接下來是查詢的性能問題。在最初做 Iceberg 性能驗證的時候，我們發現它的批量 Scan 性能是非常好的，但是點查詢的性能就比較糟糕。

（1）ID 查詢慢

舉個例子，在訂單表中，用特定 ID，如訂單 ID 或者用戶 ID 去查詢明細，簡化后的SQL 就是 order_id = ‘555’。默認的情況下，Iceberg 會基于 MinMax 做過濾，但數據按照時間戳排序，MinMax 過濾其實是不生效的，比如 File A 的 MinMax 范圍包含 555，File N MinMax 321 到 987 也包含 555，其實是過濾不掉的。因而點查詢事實上就是全表掃描。

針對點查詢場景，BloomFilter 是非常適用的。最初社區沒有這個功能，Parquet 在 1.12 的時候支持 BloomFilter，Iceberg 的默認存儲格式也是 Parquet，所以我們考慮修改 Iceberg 引入這一功能。

（2）開啟 BloomFilter

先介紹一下 BloomFilter 的作用，在這個架構圖中，比如，針對 order_id 開啟了 BloomFilter，為每一個數據文件構建 BloomFilter，將 order_id 進行哈希后映射到對應 bit，如果值存在就把對應的位設為 1，如果不存在對應的位默認是 0。在 Bloom Filter 里面，如果標志位為 1，這個值不一定存在，但如果標志位為 0，這個值一定不存在。通過努力，我們在 Iceberg 的內核里面添加了相應的支持。在 Spark 讀取  Iceberg 和 Trino 讀取的時候也添加了相應的能力。

BloomFilter 支持 Equals 和 In 過濾。如果標志位為 0 是一定能過濾的。不支持 not equals、not in、比較符等過濾條件。

示意圖中 order_id = 555 這個條件，哈希后另外兩個文件對應的標志位值都是 0，在查詢的時候就可以很快地把其他文件過濾掉了，能夠精確命中訂單所在的數據文件。

（3）BloomFilter 效果

經過測試，在 Spark SQL 中的訂單 ID 查詢，原來全表掃描需要將近 1000 秒，開啟 BloomFilter 后只需要 10 秒鐘。Trino 開啟 BF 后，可以過濾 98.5% 的查詢，CPU 消耗只有以前的 5%。

BloomFilter 會帶來額外的空間開銷。經過簡單的測試，大概有 3% 的額外空間損耗。即 3% 的存儲代價可以帶來點查詢 100 倍的提升。

（4）Alluxio 緩存

查詢優化另外一個工作是緩存加速，如使用 Alluxio 做緩存加速。

這是愛奇藝 Trino 查數據湖的架構圖。業務通過 Pilot 引擎分發到 Trino 網關，自動地選擇使用哪個 Trino 集群執行查詢。原本 Trino Worker 上面的 SSD 存儲是浪費的，我們在之上混布了 Alluxio，復用了原本閑置的 SSD 存儲，幾乎沒有什么額外機器開銷。

以前去查 HDFS 可能會有性能抖動，比如，業務有一個大的批任務，導致 HDFS 性抖動，查詢性能會降得很厲害，Alluxio 緩存能夠很好地屏蔽這一點。經過測試 Venus 日志應用 Alluxio 以后，P90 從 18 秒可以降低到 1 秒。

（5）Trino 元數據讀取問題

在實際的使用過程中發現 Trino 查詢有個意想不到的問題，元數據讀取性能遠比我們想象中的要慢。比如，讀取一個 5 M 的元數據竟然要 3 秒鐘，后面查數據可能只需要 1 秒，元數據反而更慢。

通過火焰圖和阿里的 Arthas 做定位，發現 Read 的方法被調用了百萬次，文件總共 5 M，讀取 100 多萬次是非常不合理的。進一步跟蹤，定位原因是父類里面一個 Read 方法的默認實現會逐個 Byte 讀取，Trino 這邊沒有覆蓋這個方法的實現，就會降級到默認方法，每次讀 1 個 Byte ，所以調用次數非常多，導致很慢，優化以后耗時縮短到了 0.5 秒。

五、業務落地

最后來介紹業務落地的情況，在應用了上述優化后，業務能取得什么樣的效果。

1、廣告流批一體

第一個例子是廣告的流批一體場景。原來的實時鏈路中，實時數據通過 Kafka 寫到  Kudu，離線數據同步到 Hive，通過 Impala 來統一查詢，基于離線覆蓋的進度將查詢分發到 Kudu 和 Hive。

使用 Iceberg 以后，實時和離線數據都更新 Iceberg，不需要進度管理，直接查詢 Iceberg 表即可。Iceberg 實現了兩方面的統一，一是存儲統一，不需要有兩個類型的存儲，查詢不需要做拆分。二是任務開發統一為 SQL，原先離線是 HiveSQL，實時是 Spark Jar 包，統一為 SQL 開發。數據入湖后結合分布式改造，廣告智能出價全鏈路由 35 分鐘縮短到 7-10 分鐘。

2、Venus 日志入湖

Venus 是愛奇藝內部的日志分析平臺。之前的架構中 Kafka 數據往 ElasticSearch 里面存儲，如果業務流量較大就給它一個獨立集群，小流量業務則用公共集群。這個方案存在一些問題：一是流量調度很難做，當集群流量有瓶頸時，需要把流量拆分走；二是 ES 的存儲成本非常高。

存儲改用 Iceberg 方案后，所有業務的流量都寫到一個 Iceberg 集群，不需要拆分流量。Venus 接入層通過日志查詢平臺，數據存儲的切換對用戶是透明的。Iceberg 帶來的好處包括：

① 成本顯著下降。不需要獨立的 ES 集群了，Iceberg 和 Trino 都復用現有的資源，并沒有什么額外的成本。

② 穩定性大幅提升。因為 ES 的成本太貴，沒有配副本，一旦單個磁盤或節點有問題，都會引發用戶的報障。用 Iceberg 以后，寫入帶寬非常大而且穩定性很好，報障減少了 80% 以上。

3、審核場景

接下來是愛奇藝內部的審核場景，審核場景需要對一些歷史的行記錄做修改。沒有 Iceberg 以前，沒有很好的技術方案支持行級更新。 

原來解決方案里用 MongoDB 存全量的數據，做行級的更新，然后用 ES 構建二級索引，改用 Iceberg 以后兩個存儲都統一到 Iceberg 里面。對業務帶來的好處是：

① 原本的監控告警要定期查 ES 做聚合，用 MySQL 開發報表，現在不需要了，報表直接查 Iceberg 就可以，能夠支持實時告警。

② 數據湖大幅提高業務的效率。原本分析任務開發非常復雜，要從 Mongo 里面導數非常不方便。有了數據湖以后可以統一為 SQL 查詢。

4、CDC 訂單入湖

最后是 CDC 類數據入湖，此處以訂單為例。基于 MySQL 數據做大數據分析，有兩類解決方案：第一類是每天導出一份到 Hive，缺點是每次導出都是全量，延遲很大，只能看一天以前的數據。另外全量導的性能也很差，對 MySQL 壓力也比較大。第二類是實時解決方案，增量變更寫在 Kudu 里面，Kudu 是一個成本很高的解決方案。如果 Kudu 寫入帶寬波動，同步任務負責人需要去做運維操作。

使用數據湖方案，愛奇藝實時計算平臺，通過 Flink CDC 技術很方便地可以將 MySQL 數據入湖。數據湖方案具備如下優勢，一是近實時，數據延遲在分鐘級，遠優于之前的離線方案；二是成本低，相比于 Kudu 無需獨立節點，大幅降低機器成本；三是省運維，Iceberg 寫入帶寬大且穩定，大幅降低運維代價。

六、未來規劃

最后介紹一下未來規劃。愛奇藝未來會在流批一體里面有更多的落地，包括廣告的全面推廣、Pingback 在 BI 場景的落地。另外，我們計劃把數據湖落地在特征生產，可以由以前離線或者批的特征生產，變成近實時，能夠支持晚到數據，支持樣本的行級的修正。

在技術方面會嘗試把 Iceberg 的 Puffin 統計信息用于查詢加速的場景。還會對社區在做的 Branch 和 Tag 進行調研，尋找內部的落地場景。