01背景

字節(jié)跳動(dòng)特征存儲(chǔ)痛點(diǎn)

當(dāng)前行業(yè)內(nèi)的特征存儲(chǔ)整體流程主要分為以下四步：

特征存儲(chǔ)的整體流程

業(yè)務(wù)在線進(jìn)行特征模塊抽取;

抽取后的特征以行的格式存儲(chǔ)在 HDFS，考慮到成本，此時(shí)不存儲(chǔ)原始特征，只存抽取后的特征;

字節(jié)跳動(dòng)自研的分布式框架會(huì)將存儲(chǔ)的特征并發(fā)讀取并解碼發(fā)送給訓(xùn)練器;

訓(xùn)練器負(fù)責(zé)高速訓(xùn)練。

字節(jié)跳動(dòng)特征存儲(chǔ)總量為 EB 級(jí)別，每天的增量達(dá)到 PB 級(jí)別，并且每天用于訓(xùn)練的資源也達(dá)到了百萬(wàn)核心，所以整體上字節(jié)的存儲(chǔ)和計(jì)算的體量都是非常大的。在如此的體量之下，我們遇到了以下三大痛點(diǎn)：

特征抽取周期長(zhǎng)。在特征抽取上，當(dāng)前采用的是在線抽取的方式。大量的算法工程師，每天都在進(jìn)行大量的特征相關(guān)的試驗(yàn)。在當(dāng)前的在線抽取模式下，如果有算法工程師想要調(diào)研一個(gè)新的特征，那么他首先需要定義特征的計(jì)算方式，等待在線模塊的統(tǒng)一上線，然后需要等在線抽取的特征積累到一定的量級(jí)后才可以進(jìn)行訓(xùn)練，從而判斷這個(gè)特征是否有效果。這個(gè)過(guò)程通常需要2周甚至更長(zhǎng)的時(shí)間。并且，如果發(fā)現(xiàn)特征的計(jì)算邏輯寫(xiě)錯(cuò)或想要更改計(jì)算邏輯，則需重復(fù)上述過(guò)程。在線特征抽取導(dǎo)致當(dāng)前字節(jié)特征調(diào)研的效率非常低。基于當(dāng)前的架構(gòu)，離線特征調(diào)研的成本又非常高。

特征存儲(chǔ)空間占用大。字節(jié)的特征存儲(chǔ)當(dāng)前是以行存的形式進(jìn)行存儲(chǔ)。如果基于當(dāng)前的行存做特征調(diào)研，則需要基于原來(lái)的路徑額外生成新的數(shù)據(jù)集。一方面需要額外的空間對(duì)新的數(shù)據(jù)集進(jìn)行存儲(chǔ)，另一方面還需要額外的計(jì)算資源去讀取原來(lái)的全量數(shù)據(jù)生成新的數(shù)據(jù)，且很難做數(shù)據(jù)的管理和復(fù)用。行存對(duì)于特征存儲(chǔ)來(lái)說(shuō)，也很難進(jìn)行優(yōu)化，占用空間較大。

模型訓(xùn)練帶寬大，數(shù)據(jù)讀取有瓶頸。字節(jié)當(dāng)前將每個(gè)業(yè)務(wù)線的絕大部分特征都存儲(chǔ)在一個(gè)路徑下，訓(xùn)練的時(shí)候會(huì)直接基于這個(gè)路徑進(jìn)行訓(xùn)練。對(duì)于每個(gè)模型，訓(xùn)練所需的特征是不一樣的，每個(gè)業(yè)務(wù)線可能存有上萬(wàn)個(gè)特征，而大部分模型訓(xùn)練往往只需要幾百個(gè)特征，但因?yàn)樘卣魇且孕写娓袷竭M(jìn)行存儲(chǔ)，所以訓(xùn)練時(shí)需要將上萬(wàn)特征全部讀取后，再在內(nèi)存中進(jìn)行過(guò)濾，這就使得模型訓(xùn)練的帶寬需求非常大，數(shù)據(jù)的讀取成為了整個(gè)訓(xùn)練的瓶頸。

基于痛點(diǎn)的需求梳理

基于上述問(wèn)題，我們與業(yè)務(wù)方一同總結(jié)了若干需求：

存儲(chǔ)原始特征：由于在線特征抽取在特征調(diào)研上的低效率，我們期望能夠存儲(chǔ)原始特征;

離線調(diào)研能力：在原始特征的基礎(chǔ)上，可以進(jìn)行離線調(diào)研，從而提升特征調(diào)研效率;

支持特征回填：支持特征回填，在調(diào)研完成后，可以將歷史數(shù)據(jù)全部刷上調(diào)研好的特征;

降低存儲(chǔ)成本：充分利用數(shù)據(jù)分布的特殊性，降低存儲(chǔ)成本，騰出資源來(lái)存儲(chǔ)原始特征;

降低訓(xùn)練成本：訓(xùn)練時(shí)只讀需要的特征，而非全量特征，降低訓(xùn)練成本;

提升訓(xùn)練速度：訓(xùn)練時(shí)盡量降低數(shù)據(jù)的拷貝和序列化反序列化開(kāi)銷(xiāo)。

02字節(jié)跳動(dòng)海量特征存儲(chǔ)解決方案

在字節(jié)的整體架構(gòu)中，最上層是業(yè)務(wù)層，包括抖音、頭條、小說(shuō)等字節(jié)絕大部分業(yè)務(wù)線;

其下我們通過(guò)平臺(tái)層，給業(yè)務(wù)同學(xué)提供簡(jiǎn)單易用的 UI 和訪問(wèn)控制等功能;

在框架層，我們使用 Spark 作為特征處理框架(包括預(yù)處理和離線特征調(diào)研等)，字節(jié)自研的 Primus 作為訓(xùn)練框架;

在格式層，我們選用 Parquet 作為文件格式，Iceberg 作為表格式;

最下層是調(diào)度器 Yarn & K8s 以及存儲(chǔ) HDFS。

下面我們重點(diǎn)針對(duì)格式層進(jìn)行詳細(xì)介紹。

技術(shù)選型

為了滿足業(yè)務(wù)方提到的6個(gè)需求，我們首先想到的是通過(guò) Parquet 列存的格式，降低行存的存儲(chǔ)成本，節(jié)省的空間可用來(lái)存儲(chǔ)原始特征。同時(shí)由于 Parquet 選列可以下推到存儲(chǔ)層的特性，在訓(xùn)練時(shí)可以只讀需要的特征，從而降低訓(xùn)練時(shí)反序列化的成本，提升訓(xùn)練的速度。

但是使用 Parquet 引入了額外的問(wèn)題，原來(lái)的行存是基于 Protobuf 定義的半結(jié)構(gòu)化數(shù)據(jù)，不需要預(yù)先定義 Schema，而使用 Parquet 以后，我們需要先知道 Schema，然后才能進(jìn)行數(shù)據(jù)的存取，那么在特征新增和淘汰時(shí)，Schema 的更新就是一個(gè)很難解決的問(wèn)題。Parquet 并不支持?jǐn)?shù)據(jù)回填，如果要回填歷史幾年的數(shù)據(jù)，就需要將數(shù)據(jù)全量讀取，增加新列，再全量寫(xiě)回，這一方面會(huì)浪費(fèi)大量的計(jì)算資源，另一方面做特征回填時(shí)的 overwrite 操作，會(huì)導(dǎo)致當(dāng)前正在進(jìn)行訓(xùn)練的任務(wù)由于文件被替換而失敗。

為了解決這幾個(gè)問(wèn)題，我們引入了 Iceberg 來(lái)支持模式演進(jìn)、特征回填和并發(fā)讀寫(xiě)。

Iceberg 是適用于大型數(shù)據(jù)集的一個(gè)開(kāi)源表格式，具備模式演進(jìn)、隱藏分區(qū)&分區(qū)演進(jìn)、事務(wù)、MVCC、計(jì)算存儲(chǔ)引擎解耦等特性，這些特性匹配了我們所有的需求。因此，我們選擇了 Iceberg 作為我們的數(shù)據(jù)湖。

整體上 Iceberg 是一個(gè)分層的結(jié)構(gòu)，snapshot 層存儲(chǔ)了當(dāng)前表的所有快照;manifest list 層存儲(chǔ)了每個(gè)快照包含的 manifest 云數(shù)據(jù)，這一層的用途主要是為了多個(gè) snapshot 可以復(fù)用下一層的 manifest;manifest 層，存儲(chǔ)了下層 Data Files 元數(shù)據(jù);最下面的 Data File 是就是實(shí)際的數(shù)據(jù)文件。通過(guò)這樣的多層結(jié)構(gòu)，Iceberg 可以支持上述包括模式演進(jìn)等幾個(gè)特性。

下面我們來(lái)一一介紹 Iceberg 如何支持這些功能。

字節(jié)跳動(dòng)海量特征存儲(chǔ)解決方案

并發(fā)讀寫(xiě)

在并發(fā)讀取方面，Iceberg 是基于快照的讀取，對(duì) Iceberg 的每個(gè)操作都會(huì)生成新的快照，不影響正在讀取的快照，從而保證讀寫(xiě)互不影響。

在并發(fā)寫(xiě)入方面，Iceberg 是采用樂(lè)觀并發(fā)的方式，利用HDFS mv 的原子性語(yǔ)義保證只有一個(gè)能寫(xiě)入成功，而其他的并發(fā)寫(xiě)入會(huì)被檢查是否有沖突，若沒(méi)有沖突，則寫(xiě)入下一個(gè) snapshot。

模式演進(jìn)

Iceberg 的模式演進(jìn)原理

我們知道，Iceberg 元數(shù)據(jù)和 Parquet 元數(shù)據(jù)都有 Column，而中間的映射關(guān)系，是通過(guò) ID 字段來(lái)進(jìn)行一對(duì)一映射。

例如上面左圖中，Iceberg 和 Parquet 分別有 ABC 三列，對(duì)應(yīng) ID 1、2、3。那最終讀取出的 Dataframe 就是 和 Parquet 中一致包含 ID 為1、2、3的 ABC 三列。而當(dāng)我們對(duì)左圖進(jìn)行兩個(gè)操作，刪除舊的 B 列，寫(xiě)入新的 B 列后， Iceberg 對(duì)應(yīng)的三列 ID 會(huì)變成1、3、4，所以右圖中讀出來(lái)的 Dataframe，雖然也是 ABC 三列，但是這個(gè) B 列的 ID 并非 Parquet 中 B 列的 ID，因此最終實(shí)際的數(shù)據(jù)中，B 列為空值。

特征回填

寫(xiě)時(shí)復(fù)制

如上圖所示，COW 方式的特征回填通過(guò)一個(gè) Backfill 任務(wù)將原快照中的數(shù)據(jù)全部讀出，然后寫(xiě)入新列，再寫(xiě)出到新的 Data File 中，并生成新的快照。

這種方式的缺點(diǎn)在于雖然我們只需要寫(xiě)一列數(shù)據(jù)，但是需要將整體數(shù)據(jù)全部讀出，再全部寫(xiě)回，不僅浪費(fèi)了大量的計(jì)算資源用來(lái)對(duì)整個(gè) Parquet 文件進(jìn)行編碼解碼，還浪費(fèi)了大量的 IO 來(lái)讀取全量數(shù)據(jù)，且浪費(fèi)了大量的存儲(chǔ)資源來(lái)存儲(chǔ)重復(fù)的 ABC 列。

因此我們基于開(kāi)源 Iceberg 自研了 MOR 的 Backfill 方案。

讀時(shí)合并

如上圖所示，在 MOR 方案中，我們?nèi)匀恍枰粋€(gè) Backfill 任務(wù)來(lái)讀取原始的 Data File 文件，但是這里我們只讀取需要的字段。比如我們只需要 A 列通過(guò)某些計(jì)算邏輯生成 D 列，那么 Backfill 任務(wù)則只讀取 A 的數(shù)據(jù)，并且 Snapshot2 中只需要寫(xiě)包含 D 列的 update 文件。隨著新增列的增多，我們也需要將 Update 文件合并回 Data File 文件中。

為此，我們又提供了 Compaction 邏輯，即讀取舊的 Data File 和 Update File，并合并成一個(gè)單獨(dú)的 Data File。

MOR原理如上圖，假設(shè)原來(lái)有一個(gè)邏輯 Dataframe 是由兩個(gè) Data File 構(gòu)成， 現(xiàn)在需要回填一個(gè) ColD 的內(nèi)容。我們會(huì)寫(xiě)入一個(gè)包含 ColD 的 Update File，這樣 Snapshot2 中的邏輯 Dataframe 就會(huì)包含ABCD 四列。

實(shí)現(xiàn)細(xì)節(jié)：

Data File 和 Update File 都需要一個(gè)主鍵，并且每個(gè)文件都需要按照主鍵排序，在這個(gè)例子中是 ID;

讀取時(shí)，會(huì)根據(jù)用戶(hù)選擇的列，分析具體需要哪些 Update File 和 Data File;

根據(jù) Data File 中主鍵的 min-max 值去選擇與該 Data File 相對(duì)應(yīng)的 Update File;

MOR 整個(gè)過(guò)程是多個(gè) Data File 和 Update File 多路歸并的過(guò)程;

歸并的順序由 SEQ 來(lái)決定，SEQ 大的數(shù)據(jù)會(huì)覆蓋 SEQ 小的數(shù)據(jù)。

COW 與 MOR 特性比較

相比于 COW 方式全量讀取和寫(xiě)入所有列，MOR 的優(yōu)勢(shì)是只讀取需要的列，也只寫(xiě)入更新的列，沒(méi)有讀寫(xiě)放大問(wèn)題。在計(jì)算上節(jié)省了大量的資源，讀寫(xiě)的 IO 也大大降低，相比 COW 方式每次 COW 都翻倍的情況， MOR 只需要存儲(chǔ)新增列，也大大避免了存儲(chǔ)資源浪費(fèi)。

考慮到性能的開(kāi)銷(xiāo)，我們需要定期 Compaction，Compaction 是一個(gè)比較重的操作，和 COW 相當(dāng)。但是 Compaction 是一個(gè)異步的過(guò)程，可以在多次 MOR 后進(jìn)行一次 Compaction。那么一次 Compaction 的開(kāi)銷(xiāo)就可以攤銷(xiāo)到多次 MOR 上，例如10次 COW 和10次 MOR + 1次 Compaction 相比，存儲(chǔ)和讀寫(xiě)成本都從原來(lái)的 10x 降到當(dāng)前的 2x 。

MOR 的實(shí)現(xiàn)成本較高，但這可以通過(guò)良好的設(shè)計(jì)和大量的測(cè)試來(lái)解決。

而對(duì)于模型訓(xùn)練來(lái)說(shuō)，由于大多數(shù)模型訓(xùn)練只需要自己的列，所以大量的線上模型都不需要走 MOR 的邏輯，可以說(shuō)基本沒(méi)有開(kāi)銷(xiāo)。而少數(shù)的調(diào)研模型，往往只需讀自己的 Update File 而不用讀其他的 Update File ，所以整體上讀取的額外資源也并未增加太多。

訓(xùn)練優(yōu)化

從行存改為 Iceberg 后，我們也在訓(xùn)練上也做了大量的優(yōu)化。

在我們的原始架構(gòu)中，分布式訓(xùn)練框架并不解析實(shí)際的數(shù)據(jù)內(nèi)容，而是直接以行的形式把數(shù)據(jù)透?jìng)鹘o訓(xùn)練器，訓(xùn)練器在內(nèi)部進(jìn)行反序列化、選列等操作。

原始架構(gòu)

引入 Iceberg 后，我們要拿到選列帶來(lái)的 CPU 和 IO 收益就需要將選列下推到存儲(chǔ)層。最初為了保證下游訓(xùn)練器感知不到，我們?cè)谟?xùn)練框架層面，將選列反序列化后，構(gòu)造成原來(lái)的 ROW 格式，發(fā)送給下游訓(xùn)練器。相比原來(lái)，多了一層序列化反序列化的開(kāi)銷(xiāo)。

這就導(dǎo)致遷移到 Iceberg 后，整體訓(xùn)練速度反而變慢，資源也增加了。

列式改造為了提升訓(xùn)練速度，我們通過(guò)向量化讀取的方式，將 Iceberg 數(shù)據(jù)直接讀成 Batch 數(shù)據(jù)，發(fā)送給訓(xùn)練器，這一步提升了訓(xùn)練速度，并降低了部分資源消耗。

向量化讀取

為了達(dá)到最優(yōu)效果，我們與訓(xùn)練器團(tuán)隊(duì)合作，直接修改了訓(xùn)練器內(nèi)部，使訓(xùn)練器可以直接識(shí)別 Arrow 數(shù)據(jù)，這樣我們就實(shí)現(xiàn)了從 Iceberg 到訓(xùn)練器端到端的 Arrow 格式打通，這樣只需要在最開(kāi)始反序列化為 Arrow ，后續(xù)的操作就完全基于 Arrow 進(jìn)行，從而降低了序列化和反序列化開(kāi)銷(xiāo)，進(jìn)一步提升訓(xùn)練速度，降低資源消耗。

Arrow

優(yōu)化收益

最終，我們達(dá)到了最初的目標(biāo)，取得了離線特征工程的能力。在存儲(chǔ)成本上，普遍降低了40%以上;在同樣的訓(xùn)練速度下，CPU 降低了13%，網(wǎng)絡(luò)IO 降低40%。

03未來(lái)規(guī)劃

未來(lái)，我們規(guī)劃支持以下4種能力：

Upsert 的能力，支持用戶(hù)的部分?jǐn)?shù)據(jù)回流;

物化視圖的能力，支持用戶(hù)在常用的數(shù)據(jù)集上建立物化視圖，提高讀取效率;

Data Skipping 能力，進(jìn)一步優(yōu)化數(shù)據(jù)排布，下推更多邏輯，進(jìn)一步優(yōu)化 IO 和計(jì)算資源;

基于 Arrow 的數(shù)據(jù)預(yù)處理能力，向用戶(hù)提供良好的數(shù)據(jù)處理接口，同時(shí)將預(yù)處理提前預(yù)期，進(jìn)一步加速后續(xù)的訓(xùn)練。

審核編輯：湯梓紅