一、愛奇藝 OLAP 簡介

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

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

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

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

③ 近實時 Iceberg：分鐘級延遲，是今天要重點介紹的數(shù)據(jù)湖產(chǎn)品。

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

二、為什么要數(shù)據(jù)湖

下面來介紹一下愛奇藝數(shù)據(jù)湖的建設背景。

1、數(shù)據(jù)湖技術加速數(shù)據(jù)流通

為什么要有數(shù)據(jù)湖？數(shù)據(jù)湖其實就是為了加速數(shù)據(jù)流通。

愛奇藝 Pingback 投遞的場景：

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

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

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

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

2、Iceberg 定義-新型表格式

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

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

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

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

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

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

但是它也有著以下缺點：

① 元數(shù)據(jù)統(tǒng)一存在 Metastore，通常底下是 MySQL，很容易成為瓶頸。

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

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

（2）表格式-Iceberg

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

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

（3）表格式-Hive VS Iceberg

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

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

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

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

三、數(shù)據(jù)湖平臺建設

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

1、平臺總覽

這是愛奇藝數(shù)據(jù)湖整體的產(chǎn)品架構圖：

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

2、流式入湖

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

3、出湖查詢

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

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

四、性能優(yōu)化

下面介紹一些性能優(yōu)化的工作。

1、小文件

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

（1）生命周期

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

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

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

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

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

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

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

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

（2）智能合并

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

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

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

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

（3）合并性能優(yōu)化

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

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

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

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

（4）寫入?yún)?shù)控制

前文介紹了當小文件已經(jīng)產(chǎn)生的時候如何優(yōu)化，但我們更希望小文件最好不要產(chǎn)生，在寫入的時候就把文件數(shù)控制住。我們需要去了解 Flink 任務寫入的時候是怎么控制文件數(shù)量的。

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

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

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

2、查詢優(yōu)化

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

（1）ID 查詢慢

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

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

（2）開啟 BloomFilter

先介紹一下 BloomFilter 的作用，在這個架構圖中，比如，針對 order_id 開啟了 BloomFilter，為每一個數(shù)據(jù)文件構建 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，在查詢的時候就可以很快地把其他文件過濾掉了，能夠精確命中訂單所在的數(shù)據(jù)文件。

（3）BloomFilter 效果

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

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

（4）Alluxio 緩存

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

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

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

（5）Trino 元數(shù)據(jù)讀取問題

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

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

五、業(yè)務落地

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

1、廣告流批一體

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

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

2、Venus 日志入湖

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

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

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

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

3、審核場景

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

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

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

② 數(shù)據(jù)湖大幅提高業(yè)務的效率。原本分析任務開發(fā)非常復雜，要從 Mongo 里面導數(shù)非常不方便。有了數(shù)據(jù)湖以后可以統(tǒng)一為 SQL 查詢。

4、CDC 訂單入湖

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

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

六、未來規(guī)劃

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

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