?一、背景

在B站，每天都有PB級(jí)的數(shù)據(jù)注入到大數(shù)據(jù)平臺(tái)，經(jīng)過離線或?qū)崟r(shí)的ETL建模后，提供給下游的分析、推薦及預(yù)測(cè)等場(chǎng)景使用。面對(duì)如此大規(guī)模的數(shù)據(jù)，如何高效低成本地滿足下游數(shù)據(jù)的分析需求，一直是我們重點(diǎn)的工作方向。

我們之前的數(shù)據(jù)處理流程基本上是這樣的：采集端將客戶端埋點(diǎn)、服務(wù)端埋點(diǎn)、日志、業(yè)務(wù)數(shù)據(jù)庫等數(shù)據(jù)收集到HDFS、Kafka等存儲(chǔ)系統(tǒng)中，然后通過Hive、Spark、Flink等離線和實(shí)時(shí)引擎對(duì)數(shù)據(jù)進(jìn)行ETL處理及數(shù)倉(cāng)建模，數(shù)據(jù)存儲(chǔ)使用ORC列式存儲(chǔ)格式，用戶可以通過Presto、Spark等引擎對(duì)數(shù)倉(cāng)建模后的數(shù)據(jù)進(jìn)行數(shù)據(jù)探索以及構(gòu)建BI報(bào)表。對(duì)于大部分的數(shù)據(jù)服務(wù)和部分BI報(bào)表，Presto、Spark訪問ORC格式數(shù)據(jù)可能無法滿足用戶對(duì)于查詢響應(yīng)時(shí)間的要求，這時(shí)需要將數(shù)據(jù)寫入ClickHouse等這種專門的OLAP引擎或者進(jìn)一步處理數(shù)據(jù)后寫入HBase、Redis等KV存儲(chǔ)系統(tǒng)中等方式解決。

?

當(dāng)前的數(shù)據(jù)處理流程雖然在一定程度上可以滿足目前的業(yè)務(wù)需求，但是整個(gè)流程的效率和成本都還有很大的提升空間，主要體現(xiàn)在：

• 為了提升查詢效率，從Hive表出倉(cāng)到ClickHouse、HBase、Redis、ElasticSearch、Mysql等外部系統(tǒng)中，需要額外的數(shù)據(jù)開發(fā)工作，額外的存儲(chǔ)冗余，但同時(shí)擁有了更少的數(shù)據(jù)靈活性，復(fù)雜的組件支持增加了數(shù)據(jù)服務(wù)開發(fā)的成本，更長(zhǎng)的數(shù)據(jù)處理流程也降低了穩(wěn)定性和可靠性。
• 對(duì)于未出倉(cāng)的數(shù)據(jù)，用戶無論是進(jìn)行數(shù)據(jù)探索還是使用BI報(bào)表，都還受SQL on Hadoop本身性能所限，和用戶期望的交互式響應(yīng)有很大差距。

本文主要介紹為了應(yīng)對(duì)以上挑戰(zhàn)，我們?cè)诤}(cāng)一體方向上的一些探索和實(shí)踐。

二、為什么需要湖倉(cāng)一體

在討論這個(gè)問題前，我們可能首先要明確兩個(gè)概念：什么是數(shù)據(jù)湖？什么是數(shù)據(jù)倉(cāng)庫？這兩個(gè)概念在業(yè)界都有大量的討論，每個(gè)人的說法也不盡相同，我們嘗試總結(jié)如下，對(duì)于數(shù)據(jù)湖：

• 使用統(tǒng)一的分布式存儲(chǔ)系統(tǒng)，可假設(shè)為無限容量。
• 有統(tǒng)一的元數(shù)據(jù)管理系統(tǒng)。
• 使用開放的數(shù)據(jù)存儲(chǔ)格式。
• 使用開放的數(shù)據(jù)處理引擎對(duì)數(shù)據(jù)進(jìn)行加工和分析。

我們之前的大數(shù)據(jù)架構(gòu)基本上是一個(gè)典型的數(shù)據(jù)湖架構(gòu)，使用HDFS作為統(tǒng)一的存儲(chǔ)系統(tǒng)，Hive metastore提供統(tǒng)一的Schema元數(shù)據(jù)管理，數(shù)據(jù)以CSV、JSON、ORC等開放存儲(chǔ)格式存儲(chǔ)在HDFS上，用戶可以使用SQL、DataSet、FileSystem等各個(gè)層次的API使用Hive、Spark、Presto、Python等框架或語言訪問數(shù)據(jù)。

數(shù)據(jù)湖架構(gòu)的好處是有非常大的靈活性，結(jié)構(gòu)化、半結(jié)構(gòu)化、非結(jié)構(gòu)化數(shù)據(jù)都可以放在數(shù)據(jù)湖中，用戶可以使用任意合適的引擎對(duì)所有的數(shù)據(jù)進(jìn)行靈活的數(shù)據(jù)探索，幾乎沒有任何限制，但是它也存在很大的缺陷，最主要的就是數(shù)據(jù)管理和查詢效率的問題。

對(duì)于數(shù)據(jù)倉(cāng)庫：

• 自定義的數(shù)據(jù)存儲(chǔ)格式。
• 自己管理數(shù)據(jù)的組織方式。
• 強(qiáng)Schema數(shù)據(jù)，對(duì)外提供標(biāo)準(zhǔn)的SQL接口。
• 具有高效的計(jì)算存儲(chǔ)一體設(shè)計(jì)和豐富的查詢加速特性。

數(shù)據(jù)倉(cāng)庫（OLAP引擎）對(duì)于數(shù)據(jù)的要求相對(duì)更加嚴(yán)格，以ClickHouse為例，必須是預(yù)先定義的強(qiáng)Schema數(shù)據(jù)通過JDBC寫入ClickHouse中，ClickHouse使用自己的存儲(chǔ)格式存儲(chǔ)數(shù)據(jù)，并且會(huì)對(duì)數(shù)據(jù)文件進(jìn)行排序或者文件合并之類的數(shù)據(jù)組織優(yōu)化，對(duì)外提供SQL接口，不會(huì)暴露內(nèi)部的數(shù)據(jù)文件，提供索引等高級(jí)的查詢加速特性，內(nèi)部的計(jì)算引擎和存儲(chǔ)格式也會(huì)有很多的一體協(xié)同優(yōu)化，一般認(rèn)為專門的數(shù)據(jù)倉(cāng)庫查詢效率會(huì)優(yōu)于數(shù)據(jù)湖架構(gòu)，在B站的實(shí)踐上，大部分場(chǎng)景，像ClickHouse對(duì)比Spark、Presto也確實(shí)有量級(jí)上的性能提升。

在我們實(shí)際的數(shù)據(jù)處理場(chǎng)景中，除了AI和數(shù)據(jù)探索等場(chǎng)景，探索未知數(shù)據(jù)的未知問題，比較依賴數(shù)據(jù)湖架構(gòu)的靈活性，其實(shí)大部分的場(chǎng)景是基于已知數(shù)據(jù)的，即我們的數(shù)據(jù)開發(fā)同學(xué)，實(shí)際上是基于Hive表的強(qiáng)Schema數(shù)據(jù)，進(jìn)行從ODS，DWD，DWB到ADS等各個(gè)業(yè)務(wù)數(shù)倉(cāng)的分層建設(shè)，本質(zhì)上我們是主要是基于數(shù)據(jù)湖的架構(gòu)進(jìn)行業(yè)務(wù)數(shù)倉(cāng)的建設(shè)，如何提升這部分場(chǎng)景的查詢效率，使用成本和用戶體驗(yàn)是我們?cè)谶@方面工作的核心內(nèi)容。

湖倉(cāng)一體是近兩年大數(shù)據(jù)一個(gè)非常熱門的方向，如何在同一套技術(shù)架構(gòu)上同時(shí)保持湖的靈活性和倉(cāng)的高效性是其中的關(guān)鍵。常見的是兩條技術(shù)路線：一條是從分布式數(shù)倉(cāng)向湖倉(cāng)一體演進(jìn)，在分布式數(shù)倉(cāng)中支持CSV、JSON、ORC、PARQUET等開放存儲(chǔ)格式，將數(shù)據(jù)的處理流程從ETL轉(zhuǎn)換為ELT，數(shù)據(jù)注入到分布式數(shù)倉(cāng)后，在分布式數(shù)倉(cāng)中進(jìn)行業(yè)務(wù)數(shù)倉(cāng)的建模工作，比如AWS RedShift及SnowFlake等；另外一條是從數(shù)據(jù)湖向湖倉(cāng)一體演進(jìn)，基于開放的查詢引擎和新引入的開放表存儲(chǔ)格式達(dá)到分布式數(shù)倉(cāng)的處理效率，這方面閉源商業(yè)產(chǎn)品的代表是DataBricks SQL，他們基于兼容Spark API的閉源Photon內(nèi)核和DeltaLake存儲(chǔ)格式以及S3對(duì)象存儲(chǔ)的湖倉(cāng)一體架構(gòu)，宣稱在TPC-DS Benchmark上性能超過專門的云數(shù)據(jù)倉(cāng)庫SnowFlake。在開源社區(qū)領(lǐng)域，Iceberg、Hudi、DeltaLake等項(xiàng)目的出現(xiàn)也為在SQL on Hadoop的數(shù)據(jù)湖技術(shù)方案上實(shí)現(xiàn)湖倉(cāng)一體提供了基礎(chǔ)的技術(shù)儲(chǔ)備。在B站，基于我們之前的技術(shù)棧和實(shí)際的業(yè)務(wù)場(chǎng)景，我們選擇了第二個(gè)方向，從數(shù)據(jù)湖架構(gòu)向湖倉(cāng)一體演進(jìn)。

三、B站的湖倉(cāng)一體架構(gòu)

對(duì)于B站的湖倉(cāng)一體架構(gòu)，我們想要解決的問題主要有兩個(gè)：一是鑒于從Hive表出倉(cāng)到外部系統(tǒng)（ClickHouse、HBase、ES等）帶來的復(fù)雜性和存儲(chǔ)開發(fā)等額外代價(jià)，盡量減少這種場(chǎng)景出倉(cāng)的必要性。二是對(duì)于基于SQL on Hadoop的分析查詢場(chǎng)景，提升查詢效率，降低成本。我們基于Iceberg構(gòu)建了我們的湖倉(cāng)一體架構(gòu)，在具體介紹B站的湖倉(cāng)一體架構(gòu)之前，我覺得有必要先討論清楚兩個(gè)問題，為什么Iceberg可以構(gòu)建湖倉(cāng)一體架構(gòu)，以及我們?yōu)槭裁催x擇Iceberg？

1、為什么基于Iceberg可以構(gòu)建湖倉(cāng)一體架構(gòu)？

對(duì)比開放的SQL引擎、存儲(chǔ)格式如：Presto、Spark、ORC、Parquet和分布式數(shù)倉(cāng)如：ClickHouse、SnowFlake對(duì)應(yīng)層的實(shí)現(xiàn)，其實(shí)差別不大，開源分布式引擎一直在逐漸補(bǔ)足SQL Runtime和存儲(chǔ)層的一些影響性能的高級(jí)特性，比如Runtime CodeGen，向量化執(zhí)行引擎，基于statistic的CBO，索引等等，當(dāng)前兩者最大的一個(gè)不同在于對(duì)于數(shù)據(jù)組織的管理能力。

對(duì)于數(shù)據(jù)湖架構(gòu)來說，數(shù)據(jù)文件在HDFS的分布組織是由寫入任務(wù)決定的，而對(duì)于分布式數(shù)倉(cāng)來說，數(shù)據(jù)一般是通過JDBC寫入，數(shù)據(jù)的存儲(chǔ)組織方式是由數(shù)倉(cāng)本身決定的，所以數(shù)倉(cāng)可以按照對(duì)于查詢更加友好的方式組織數(shù)據(jù)的存儲(chǔ)，比如對(duì)數(shù)據(jù)文件定期compact到合適的大小或者對(duì)數(shù)據(jù)進(jìn)行合理排序和分組，對(duì)于大規(guī)模的數(shù)據(jù)來說，數(shù)據(jù)的優(yōu)化組織可以大大提高查詢的效率。

Iceberg、Hudi、DeltaLake等新的表存儲(chǔ)格式的出現(xiàn)，最主要的特性就是可以在HDFS上自組織管理表的metadata信息，從而提供了表數(shù)據(jù)的Snapshot及粗粒度的事務(wù)支持能力，基于此，我們可以在開放的查詢引擎之外，異步地，透明地對(duì)Iceberg、Hudi、DeltaLake格式的數(shù)據(jù)進(jìn)行重新的數(shù)據(jù)組織優(yōu)化，從而達(dá)到了分布式數(shù)倉(cāng)類似的效果。

2、為什么選擇Iceberg？

Iceberg、Hudi以及DeltaLake是基本同時(shí)期出現(xiàn)的開源表存儲(chǔ)格式項(xiàng)目，整體的功能和定位也是基本相同，網(wǎng)上已經(jīng)有很多相關(guān)對(duì)比介紹的文章，這里就不詳細(xì)比較了，我們選擇Iceberg的主要原因是：Iceberg在三個(gè)里面是表存儲(chǔ)格式抽象的最好的，包括讀寫引擎、Table Schema、文件存儲(chǔ)格式都是pluggable的，我們可以進(jìn)行比較靈活的擴(kuò)展，并保證和開源以及之前版本的兼容性，基于此我們也比較看好該項(xiàng)目的長(zhǎng)遠(yuǎn)發(fā)展。

下圖是我們整體的湖倉(cāng)一體架構(gòu)，支持開放的Spark、Flink等引擎從Kafka、HDFS接入數(shù)據(jù)，然后Magnus服務(wù)會(huì)異步地拉起Spark任務(wù)對(duì)Iceberg數(shù)據(jù)進(jìn)行重新的存儲(chǔ)組織優(yōu)化，我們主要是用Trino作為查詢引擎，并引入Alluxio做Iceberg的元數(shù)據(jù)和索引數(shù)據(jù)的緩存加速。

1）Magnus：Iceberg智能管理服務(wù)

Magnus是我們湖倉(cāng)一體架構(gòu)的核心組件，它負(fù)責(zé)管理優(yōu)化所有的Iceberg表中的數(shù)據(jù)。Iceberg本身是一個(gè)表存儲(chǔ)格式，雖然其項(xiàng)目本身提供了基于Spark、Flink等用于合并小文件，合并metadata文件或者清理過期Snapshot數(shù)據(jù)等Action Job，但是要依賴外部服務(wù)調(diào)度這些Action Job，而Magnus正是承擔(dān)這個(gè)角色。

我們對(duì)Iceberg進(jìn)行了擴(kuò)展，當(dāng)Iceberg表發(fā)生更新的時(shí)候，會(huì)發(fā)送一個(gè)event信息到Magnus服務(wù)中，Magnus服務(wù)維護(hù)一個(gè)隊(duì)列用于保存這些commit event信息，同時(shí)Magnus內(nèi)部的Scheduler調(diào)度器會(huì)持續(xù)消費(fèi)event隊(duì)列，并根據(jù)對(duì)應(yīng)Iceberg表的元數(shù)據(jù)信息及相關(guān)的策略決定是否及如何拉起Spark任務(wù)優(yōu)化Iceberg表的數(shù)據(jù)組織。

2）Iceberg內(nèi)核增強(qiáng)

對(duì)于豐富的多維分析場(chǎng)景，我們也有針對(duì)性的在Iceberg內(nèi)核和其他方面進(jìn)行了定制化增強(qiáng)，這里簡(jiǎn)要介紹兩個(gè)方面：Z-Order排序和索引。

3、Z-Order排序

Iceberg在表的metadata中記錄了文件級(jí)別每個(gè)列的MinMax信息，并且支持小文件合并以及全局Linear排序(即Order By)，這兩者配合起來，我們可以在很多查詢場(chǎng)景實(shí)現(xiàn)非常好的DataSkiping效果，比如我們對(duì)于某個(gè)Iceberg表的數(shù)據(jù)文件按照字段a進(jìn)行全局排序后，如果后續(xù)查詢帶有a的過濾條件，查詢引擎會(huì)通過PredictePushDown把過濾條件下推到文件訪問層，我們就可以根據(jù)MinMax索引把所有不需要的文件直接跳過，只訪問數(shù)據(jù)所在的文件即可。

在多維分析的實(shí)際場(chǎng)景中，一般都會(huì)有多個(gè)常用的過濾字段，Linear Order只對(duì)靠前字段有較好的Data Skip效果，通常會(huì)采用將低基數(shù)字段作為靠前的排序字段，從而才能保證對(duì)于后面的排序字段在過濾時(shí)也有一定的Data Skipping效果，但這無法從根本上解決問題，需要引入一種新的排序機(jī)制，使得多個(gè)常用的過濾字段均能夠獲得比較好的Data Skipping效果。

Interleaved Order(即Z-Order)是在圖像處理以及數(shù)倉(cāng)中使用的一種排序方式，Z-ORDER曲線可以以一條無限長(zhǎng)的一維曲線，穿過任意維度的所有空間，對(duì)于一條數(shù)據(jù)的多個(gè)排序字段，可以看作是數(shù)據(jù)的多個(gè)維度，多維數(shù)據(jù)本身是沒有天然的順序的，但是Z-Order通過一定規(guī)則將多維數(shù)據(jù)映射到一維數(shù)據(jù)上，構(gòu)建z-value，從而可以基于一維數(shù)據(jù)進(jìn)行排序，此外Z-Order的映射規(guī)則保證了按照一維數(shù)據(jù)排序后的數(shù)據(jù)同時(shí)根據(jù)多個(gè)排序字段聚集。

參考wikipedia中的Z-Order介紹，可以通過對(duì)兩個(gè)數(shù)據(jù)比特位的交錯(cuò)填充來構(gòu)建z-value，如下圖所示，對(duì)于(x, y)兩維數(shù)據(jù)，數(shù)據(jù)值 0 ≤ x ≤ 7, 0 ≤ y ≤ 7，構(gòu)建的z-values以及z-order順序如下：

可以看到，如果根據(jù)z-values的順序?qū)?shù)據(jù)進(jìn)行排序，并平均分為4個(gè)文件，無論我們?cè)诓樵冎惺褂脁還是y字段過濾進(jìn)行點(diǎn)查詢，都可以skip一半的不相干文件，如果數(shù)據(jù)量更大，效果會(huì)更好，也就是說，基于Z-Order分區(qū)存儲(chǔ)的文件，可以在多個(gè)字段上擁有比較好的Data Skipping效果。我們對(duì)Spark進(jìn)行了增強(qiáng)，支持Z-Order Range Partitioner用于對(duì)Iceberg數(shù)據(jù)進(jìn)行文件間的排序組織，擴(kuò)展了Iceberg表的元信息，用戶可以自定義期望的Iceberg表的Distribution信息，支持按照Hash、Range、Z-Order等方式進(jìn)行文件間數(shù)據(jù)排序，以及對(duì)應(yīng)的OptimizeAction用于拉起Spark任務(wù)，按照用戶定義的Distribution信息對(duì)Iceberg表進(jìn)行重組織。具體詳情可查詢參考文獻(xiàn)[1]（通過數(shù)據(jù)組織加速大規(guī)模數(shù)據(jù)分析）。

4、索引

Iceberg默認(rèn)存儲(chǔ)文件級(jí)別每列的Min、Max信息，并用于TableScan階段的文件過濾，基本等價(jià)于分布式數(shù)倉(cāng)中的MinMax索引，MinMax索引對(duì)于排序后的字段DataSkipping效果很好，但是對(duì)于非排序字段，數(shù)據(jù)隨機(jī)散布于各個(gè)文件，使用該字段過濾時(shí)，MinMax索引基本很難有文件Skip的效果，BloomFilter索引在這種場(chǎng)景下可以更好地發(fā)揮作用，尤其是當(dāng)字段基數(shù)較大的時(shí)候。布隆過濾器實(shí)際上是一個(gè)很長(zhǎng)的二進(jìn)制向量和多個(gè)Hash函數(shù)，數(shù)據(jù)通過多個(gè)函數(shù)映射到二進(jìn)制向量的比特位上，布隆過濾器的空間效率和查詢時(shí)間都非常高效，非常適合用于檢索一個(gè)元素是否存在于一個(gè)集合中。

布隆過濾器的空間效率和查詢時(shí)間都非常高效，但是在使用上也有局限之處，主要是它能夠支持的過濾條件是有限的，只適用于：=、IN、NotNull等等值表達(dá)式，對(duì)于常見的Range過濾，比如>、>=、<、<=等是不支持的。為了支持更豐富的過濾表達(dá)式，我們引入了BitMap索引。BitMap也是一個(gè)非常常見的數(shù)據(jù)結(jié)構(gòu)，將一組正整形數(shù)據(jù)映射到比特位，相比于BloomFilter，不存在Hash沖突的情況，所以不會(huì)出現(xiàn)False-Positive，但是一般需要更多的存儲(chǔ)空間。對(duì)于高基數(shù)字段的BitMap索引，落地實(shí)現(xiàn)主要的問題在于：

• 需要存儲(chǔ)字段基數(shù)對(duì)應(yīng)個(gè)BitMap，存儲(chǔ)代價(jià)太大。
• 在Range過濾時(shí)，使用BitMap判斷是否可以Skip文件時(shí)，需要訪問大量BitMap，讀取代價(jià)太大。

為了解決以上問題，我們引入了Bit-sliced Encoded Bitmap實(shí)現(xiàn)。具體詳情可查詢參考文獻(xiàn)[2]（通過索引加速湖倉(cāng)一體分析）。

1）在B站的落地

基于Iceberg的湖倉(cāng)一體方案在B站的數(shù)據(jù)分析場(chǎng)景正逐漸落地，我們目前已經(jīng)支撐PB級(jí)的數(shù)據(jù)量，每天響應(yīng)幾萬個(gè)查詢，其中P90的查詢可以在1s內(nèi)響應(yīng)，滿足了多個(gè)運(yùn)營(yíng)分析數(shù)據(jù)服務(wù)交互式分析的需求。接下來，我們希望能夠?qū)⒑}(cāng)一體架構(gòu)作為我們OLAP數(shù)倉(cāng)建模的基礎(chǔ)，統(tǒng)一大部分的業(yè)務(wù)數(shù)倉(cāng)分析層數(shù)據(jù)的存儲(chǔ)和查詢，簡(jiǎn)化技術(shù)架構(gòu)，提升查詢效率，節(jié)省資源成本。

四、總結(jié)和展望

相比于傳統(tǒng)的SQL on Hadoop技術(shù)棧，基于Iceberg的湖倉(cāng)一體架構(gòu)，在保證了和已有Hadoop技術(shù)棧的兼容性情況下，提供了接近分布式數(shù)倉(cāng)的分析效率，兼顧了湖的靈活性和倉(cāng)的高效性，從我們落地實(shí)踐的經(jīng)驗(yàn)看，對(duì)于用戶基本透明，只是一種新的Hive表存儲(chǔ)格式，沒有更多使用和認(rèn)知的門檻，和已有的大數(shù)據(jù)平臺(tái)工具和服務(wù)也能非常小代價(jià)地集成。為了進(jìn)一步提高在不同場(chǎng)景的查詢效率和使用體驗(yàn)，我們還在以下方向?qū)ceberg進(jìn)行進(jìn)一步的增強(qiáng)：

• 星型模型的數(shù)據(jù)分布組織，支持按照維度表字段對(duì)事實(shí)表數(shù)據(jù)進(jìn)行排序組織和索引。
• 預(yù)計(jì)算，通過預(yù)計(jì)算對(duì)固定查詢模式進(jìn)行加速。
• 智能化，自動(dòng)采集用戶查詢歷史，分析查詢模式，自適應(yīng)調(diào)整數(shù)據(jù)的排序組織和索引等。

后續(xù)的進(jìn)展我們會(huì)持續(xù)更新，歡迎感興趣的小伙伴來和我們一起交流溝通。?