Hudi 實(shí)踐 | 字節(jié)跳動(dòng)基于 Apache Hudi 構(gòu)建 EB 級(jí)數(shù)據(jù)湖實(shí)踐

來(lái)自字節(jié)跳動(dòng)的管梓越同學(xué)一篇關(guān)于Apache Hudi在字節(jié)跳動(dòng)構(gòu)建EB級(jí)數(shù)據(jù)湖實(shí)踐的分享。

接下來(lái)將分為場(chǎng)景需求、設(shè)計(jì)選型、功能支持、性能調(diào)優(yōu)、未來(lái)展望五部分介紹Hudi在字節(jié)跳動(dòng)推薦系統(tǒng)中的實(shí)踐。

在推薦系統(tǒng)中，我們?cè)趦蓚€(gè)場(chǎng)景下使用數(shù)據(jù)湖

1.我們使用BigTable作為整個(gè)系統(tǒng)近線處理的數(shù)據(jù)存儲(chǔ)，這是一個(gè)公司自研的組件TBase，提供了BigTable的語(yǔ)義和搜索推薦廣告場(chǎng)景下一些需求的抽象，并屏蔽底層存儲(chǔ)的差異。為了更好的理解，這里可以把它直接看做一個(gè)HBase。在這過(guò)程中為了能夠服務(wù)離線對(duì)數(shù)據(jù)的分析挖掘需求，需要將數(shù)據(jù)導(dǎo)出到離線存儲(chǔ)中。在過(guò)去用戶或是使用MR/Spark直接訪問(wèn)存儲(chǔ)，或是通過(guò)掃庫(kù)的方式獲取數(shù)據(jù)，不符合OLAP場(chǎng)景下的數(shù)據(jù)訪問(wèn)特性。因此我們基于數(shù)據(jù)湖構(gòu)建BigTable的CDC，提高數(shù)據(jù)時(shí)效，減少近線系統(tǒng)訪問(wèn)壓力，提供高效的OLAP訪問(wèn)和用戶友好的SQL消費(fèi)方式。2.除此之外，我們還在特征工程與模型訓(xùn)練的場(chǎng)景中使用數(shù)據(jù)湖。我們從內(nèi)部和外部分別獲得兩類實(shí)時(shí)數(shù)據(jù)流，一個(gè)是來(lái)自系統(tǒng)內(nèi)部回流的Instance，包含了推薦系統(tǒng)Serving時(shí)獲得的Feature。另一種是來(lái)自端上埋點(diǎn)/多種復(fù)雜外部數(shù)據(jù)源的反饋，這類數(shù)據(jù)作為L(zhǎng)abel，和之前的feature共同組成了完整的機(jī)器學(xué)習(xí)樣本。針對(duì)這個(gè)場(chǎng)景，我們需要實(shí)現(xiàn)一個(gè)基于主鍵的拼接操作，將Instance和Label Merge到一起。開(kāi)窗范圍可能長(zhǎng)達(dá)數(shù)十天，千億行量級(jí)。需要支持高效的列式選取和謂詞下推。同時(shí)還需要支持并發(fā)Update等相關(guān)能力。

在這兩個(gè)場(chǎng)景下存在如下挑戰(zhàn)

1.數(shù)據(jù)的非常不規(guī)整。相比Binlog，WAL沒(méi)法獲得一行的全部信息，同時(shí)數(shù)據(jù)大小變化非常大。2.吞吐量比較大，單表吞吐超百GB/s，單表PB級(jí)存儲(chǔ)。3.數(shù)據(jù)Schema 復(fù)雜。數(shù)據(jù)存在高維、稀疏等現(xiàn)象。表列數(shù)從1000-10000+都有。并且有大量復(fù)雜數(shù)據(jù)類型。

在引擎選型時(shí)，我們考察過(guò)Hudi，Iceberg，DeltaLake三個(gè)最熱門的數(shù)據(jù)湖引擎。三者在我們的場(chǎng)景下各有優(yōu)劣，最終基于Hudi對(duì)上下游生態(tài)的開(kāi)放，對(duì)全局索引的支持，對(duì)若干存儲(chǔ)邏輯提供了定制化的開(kāi)發(fā)接口等原因，選擇了Hudi作為存儲(chǔ)引擎。

?針對(duì)實(shí)時(shí)寫(xiě)入，選擇了時(shí)效性更好的MOR。?考察了索引類型，首先因?yàn)閃AL不能每次都獲取到數(shù)據(jù)的分區(qū)，所以必須要全局索引。在幾種全局索引實(shí)現(xiàn)中，為了實(shí)現(xiàn)高性能的寫(xiě)入，HBase是唯一的選擇。另外兩種的實(shí)現(xiàn)決定了都和HBase在性能有本質(zhì)上的差距。?在計(jì)算引擎上和API上，當(dāng)時(shí)Hudi對(duì)Flink的支持還不是特別完善，所以選擇了更為成熟的Spark，為了能靈活實(shí)現(xiàn)一些定制功能和邏輯，也因?yàn)镈ataFrame的API語(yǔ)義限制比較多，所以選擇了更底層的RDD API。

功能支持包括存儲(chǔ)語(yǔ)義的MVCC和Schema注冊(cè)系統(tǒng)。

首先為了支持WAL語(yǔ)義的寫(xiě)入，我們實(shí)現(xiàn)了針對(duì)MVCC的Payload，基于Avro自定義了一套帶時(shí)間戳的數(shù)據(jù)結(jié)構(gòu)實(shí)現(xiàn)。并通過(guò)視圖訪問(wèn)的方式對(duì)用戶屏蔽了這套邏輯。除此之外還實(shí)現(xiàn)了HBase Append的語(yǔ)義，可以實(shí)現(xiàn)對(duì)List類型的追加寫(xiě)而非覆蓋寫(xiě)。

由于Hudi本身的Schema從Write的數(shù)據(jù)中獲取，這種方式和其他系統(tǒng)對(duì)接不是很方便，以及我們需要一些基于Schema的擴(kuò)展功能，所以我們構(gòu)建了一個(gè)元數(shù)據(jù)中心來(lái)提供元數(shù)據(jù)相關(guān)的操作。

?首先我們基于一種內(nèi)部的存儲(chǔ)提供的語(yǔ)義實(shí)現(xiàn)了原子變更和異地多活。用戶可以通過(guò)接口原子地觸發(fā)Schema變更并立刻獲得結(jié)果。?并通過(guò)加入版本號(hào)的方法實(shí)現(xiàn)了Schema的多版本，有了版本號(hào)之后可以方便的使用Schema而不是把Json傳來(lái)傳去。有了多版本也可以實(shí)現(xiàn)Schema更靈活的演進(jìn)。?我們還支持了列級(jí)別的額外信息編碼，來(lái)幫助業(yè)務(wù)實(shí)現(xiàn)一些場(chǎng)景下特有的擴(kuò)展功能。并把列名替換成了數(shù)字來(lái)節(jié)約使用過(guò)程中的開(kāi)銷。?Hudi的Spark Job在使用的時(shí)候會(huì)在JVM級(jí)別構(gòu)建一個(gè)local cache并通過(guò)pull的方式和元數(shù)據(jù)中心同步數(shù)據(jù)，實(shí)現(xiàn)Schema的快速訪問(wèn)和進(jìn)程內(nèi)Schema的單例。

在我們場(chǎng)景下性能挑戰(zhàn)比較大，最大單表數(shù)據(jù)量達(dá)400PB+，日增PB級(jí)數(shù)據(jù)量，總數(shù)據(jù)量達(dá)EB級(jí)別，因此我們針對(duì)性能和數(shù)據(jù)特性開(kāi)發(fā)做了一些工作來(lái)提高性能。

序列化方面包括如下優(yōu)化

1.Schema：數(shù)據(jù)使用Avro序列化開(kāi)銷特別大，而且消耗資源也非常多。針對(duì)這個(gè)問(wèn)題，我們首先借助Schema的JVM單例，規(guī)避了序列化過(guò)程中很多費(fèi)CPU的比較操作。2.通過(guò)優(yōu)化Payload邏輯，減少了需要序列化的次數(shù)。3.借助了第三方的Avro序列化實(shí)現(xiàn)，通過(guò)將序列化過(guò)程編譯成字節(jié)碼的方式來(lái)提高SerDe的速度以及降低內(nèi)存占用。對(duì)這種序列化形式做了修改，以保證我們的復(fù)雜Schema也能夠正常編譯。

對(duì)于Compaction流程優(yōu)化如下

?Hudi除了默認(rèn)的Inline/Async compaction選項(xiàng)之外，還支持Compaction的靈活部署。Compaction Job的作業(yè)特性和Ingestion作業(yè)其實(shí)有較大區(qū)別。在同一個(gè)Spark Application當(dāng)中不僅不能針對(duì)性設(shè)置，也存在資源彈性不足的問(wèn)題。我們首先構(gòu)建了獨(dú)立部署的腳本，讓Compaction作業(yè)可以獨(dú)立觸發(fā)運(yùn)行。使用了低成本的混部隊(duì)列并可以針對(duì)此次Compaction的Plan做資源申請(qǐng)。除此之外還做了基于規(guī)則和啟發(fā)式的Compaction Strategy，用戶的需求通常是保證天級(jí)別或者小時(shí)級(jí)別的SLA，并針對(duì)性地壓縮某些分區(qū)的數(shù)據(jù)，所以提供了針對(duì)性壓縮的能力。?為了能縮短關(guān)鍵Compaction的時(shí)間，我們通常會(huì)提前做Compaction來(lái)避免所有工作都在一個(gè)Compaction Job中完成。但是如果一個(gè)Compact過(guò)的FileGroup又有新的Update，就不得不再次Compact。為了優(yōu)化整體的效率，我們根據(jù)業(yè)務(wù)信息對(duì)一個(gè)FileGroup該在何時(shí)被壓縮做了啟發(fā)式的調(diào)度以減少額外的壓縮損耗。該特性的具體收益還在評(píng)估中。?最后我們對(duì)Compaction做了一些流程的優(yōu)化，比如不使用WriteStatus的Cache等等。

HDFS作為一種面向吞吐設(shè)計(jì)的存儲(chǔ)，在集群水位比較高的情況下，實(shí)時(shí)寫(xiě)入毛刺比較嚴(yán)重。通過(guò)和HDFS團(tuán)隊(duì)的溝通與合作，做了相關(guān)的一些工作。

?首先把原有的數(shù)據(jù)HSync操作替換為HFlush，避免了分散性u(píng)pdate導(dǎo)致的磁盤IO寫(xiě)放大。?針對(duì)場(chǎng)景調(diào)參做了激進(jìn)的pipeline切換設(shè)置，并且HDFS團(tuán)隊(duì)開(kāi)發(fā)了靈活的可以控制pipeline的api，來(lái)實(shí)現(xiàn)這個(gè)場(chǎng)景下靈活的配置。?最后還通過(guò)logfile獨(dú)立IO隔離的方式保證了實(shí)時(shí)寫(xiě)入的時(shí)效性。

還有一些零零碎碎的性能提升，流程修改和Bug Fix，大家感興趣可以找我交流。

未來(lái)我們會(huì)在以下幾個(gè)方面持續(xù)迭代。

?產(chǎn)品化問(wèn)題：目前使用的API和調(diào)參調(diào)優(yōu)方式都對(duì)用戶要求很高，尤其是調(diào)參和運(yùn)維，需要對(duì)Hudi原理有相當(dāng)?shù)牧私獠趴梢酝瓿?，不利于用戶推廣使用。?生態(tài)對(duì)接問(wèn)題：在我們的場(chǎng)景中，技術(shù)棧以Flink為主，未來(lái)會(huì)探索Flink的使用。除此之外上下游使用的應(yīng)用和環(huán)境也比較復(fù)雜，非常需要跨語(yǔ)言和通用的接口實(shí)現(xiàn)。目前和Spark綁定過(guò)于嚴(yán)重。?成本和性能問(wèn)題：老生常談的話題，由于我們場(chǎng)景比較大，所以在這塊優(yōu)化上的收益非常可觀。?存儲(chǔ)語(yǔ)義：我們把Hudi當(dāng)做一種存儲(chǔ)來(lái)使用而非一種表格式。所以未來(lái)會(huì)拓展Hudi的使用場(chǎng)景，需要更豐富的存儲(chǔ)語(yǔ)義，會(huì)在這方面做更多的工作。

最后打個(gè)廣告，我們推薦架構(gòu)團(tuán)隊(duì)負(fù)責(zé)抖音、今日頭條、西瓜視頻等產(chǎn)品的推薦架構(gòu)設(shè)計(jì)和開(kāi)發(fā)，挑戰(zhàn)大、成長(zhǎng)快。現(xiàn)在正在招人，工作地包括：北京/上海/杭州/新加坡/山景城，有興趣的小伙伴歡迎添加微信 qinglingcannotfly 或發(fā)送簡(jiǎn)歷至郵箱: [email protected]