2021 年，字節(jié)跳動旗下產(chǎn)品總 MAU 已超過 19 億。在以抖音、今日頭條、西瓜視頻等為代表的產(chǎn)品業(yè)務背景下，強大的推薦系統(tǒng)顯得尤為重要。Flink 提供了非常強大的 SQL 模塊和有狀態(tài)計算模塊。目前在字節(jié)推薦場景，實時簡單計數(shù)特征、窗口計數(shù)特征、序列特征已經(jīng)完全遷移到 Flink SQL 方案上。結(jié)合 Flink SQL 和 Flink 有狀態(tài)計算能力，我們正在構(gòu)建下一代通用的基礎特征計算統(tǒng)一架構(gòu)，期望可以高效支持常用有狀態(tài)、無狀態(tài)基礎特征的生產(chǎn)。   

業(yè)務背景 

對于今日頭條、抖音、西瓜視頻等字節(jié)跳動旗下產(chǎn)品，基于 Feed 流和短時效的推薦是核心業(yè)務場景。而推薦系統(tǒng)最基礎的燃料是特征，高效生產(chǎn)基礎特征對業(yè)務推薦系統(tǒng)的迭代至關(guān)重要。

主要業(yè)務場景 

抖音、火山短視頻等為代表的短視頻應用推薦場景，例如 Feed 流推薦、關(guān)注、社交、同城等各個場景，整體在國內(nèi)大概有 6 億 + 規(guī)模 DAU；

頭條、西瓜等為代表的 Feed 信息流推薦場景，例如 Feed 流、關(guān)注、子頻道等各個場景，整體在國內(nèi)大概有 1.5 億 + 規(guī)模 DAU；

業(yè)務痛點和挑戰(zhàn) 

目前字節(jié)跳動推薦場景基礎特征的生產(chǎn)現(xiàn)狀是“百花齊放”。離線特征計算的基本模式都是通過消費 Kafka、BMQ、Hive、HDFS、Abase、RPC 等數(shù)據(jù)源，基于 Spark、Flink 計算引擎實現(xiàn)特征的計算，而后把特征的結(jié)果寫入在線、離線存儲。各種不同類型的基礎特征計算散落在不同的服務中，缺乏業(yè)務抽象，帶來了較大的運維成本和穩(wěn)定性問題。

而更重要的是，缺乏統(tǒng)一的基礎特征生產(chǎn)平臺，使業(yè)務特征開發(fā)迭代速度和維護存在諸多不便。如業(yè)務方需自行維護大量離線任務、特征生產(chǎn)鏈路缺乏監(jiān)控、無法滿足不斷發(fā)展的業(yè)務需求等。

在字節(jié)的業(yè)務規(guī)模下，構(gòu)建統(tǒng)一的實時特征生產(chǎn)系統(tǒng)面臨著較大挑戰(zhàn)，主要來自四個方面：

巨大的業(yè)務規(guī)模：抖音、頭條、西瓜、火山等產(chǎn)品的數(shù)據(jù)規(guī)?？蛇_到日均 PB 級別。例如在抖音場景下，晚高峰 Feed 播放量達數(shù)百萬 QPS，客戶端上報用戶行為數(shù)據(jù)高達數(shù)千萬 IOPS。業(yè)務方期望在任何時候，特征任務都可以做到不斷流、消費沒有 lag 等，這就要求特征生產(chǎn)具備非常高的穩(wěn)定性。

較高的特征實時化要求：在以直播、電商、短視頻為代表的推薦場景下，為保證推薦效果，實時特征離線生產(chǎn)的時效性需實現(xiàn)常態(tài)穩(wěn)定于分鐘級別。

更好的擴展性和靈活性：隨著業(yè)務場景不斷復雜，特征需求更為靈活多變。從統(tǒng)計、序列、屬性類型的特征生產(chǎn)，到需要靈活支持窗口特征、多維特征等，業(yè)務方需要特征中臺能夠支持逐漸衍生而來的新特征類型和需求。

業(yè)務迭代速度快：特征中臺提供的面向業(yè)務的 DSL 需要足夠場景，特征生產(chǎn)鏈路盡量讓業(yè)務少寫代碼，底層的計算引擎、存儲引擎對業(yè)務完全透明，徹底釋放業(yè)務計算、存儲選型、調(diào)優(yōu)的負擔，徹底實現(xiàn)實時基礎特征的規(guī)模化生產(chǎn)，不斷提升特征生產(chǎn)力；

迭代演進過程 

在字節(jié)業(yè)務爆發(fā)式增長的過程中，為了滿足各式各樣的業(yè)務特征的需求，推薦場景衍生出了眾多特征服務。這些服務在特定的業(yè)務場景和歷史條件下較好支持了業(yè)務快速發(fā)展，大體的歷程如下：

推薦場景特征服務演進歷程 

在這其中 2020 年初是一個重要節(jié)點，我們開始在特征生產(chǎn)中引入 Flink SQL、Flink State 技術(shù)體系，逐步在計數(shù)特征系統(tǒng)、模型訓練的樣本拼接、窗口特征等場景進行落地，探索出新一代特征生產(chǎn)方案的思路。

新一代系統(tǒng)架構(gòu) 

結(jié)合上述業(yè)務背景，我們基于 Flink SQL 和 Flink 有狀態(tài)計算能力重新設計了新一代實時特征計算方案。新方案的定位是：解決基礎特征的計算和在線 Serving，提供更加抽象的基礎特征業(yè)務層 DSL。在計算層，我們基于 Flink SQL 靈活的數(shù)據(jù)處理表達能力，以及 Flink State 狀態(tài)存儲和計算能力等技術(shù)，支持各種復雜的窗口計算。極大地縮短業(yè)務基礎特征的生產(chǎn)周期，提升特征產(chǎn)出鏈路的穩(wěn)定性。新的架構(gòu)里，我們將 特征生產(chǎn)的鏈路分為數(shù)據(jù)源抽取 / 拼接、狀態(tài)存儲、計算三個階段，F(xiàn)link SQL 完成特征數(shù)據(jù)的抽取和流式拼接，F(xiàn)link State 完成特征計算的中間狀態(tài)存儲。

有狀態(tài)特征是非常重要的一類特征，其中最常用的就是帶有各種窗口的特征，例如統(tǒng)計最近 5 分鐘視頻的播放 VV 等。對于窗口類型的特征在字節(jié)內(nèi)部有一些基于存儲引擎的方案，整體思路是“輕離線重在線”，即把窗口狀態(tài)存儲、特征聚合計算全部放在存儲層和在線完成。離線數(shù)據(jù)流負責基本數(shù)據(jù)過濾和寫入，離線明細數(shù)據(jù)按照時間切分聚合存儲（類似于 micro batch），底層的存儲大部分是 KV 存儲、或者專門優(yōu)化的存儲引擎，在線層完成復雜的窗口聚合計算邏輯，每個請求來了之后在線層拉取存儲層的明細數(shù)據(jù)做聚合計算。

我們新的解決思路是“輕在線重離線”，即把比較重的 時間切片明細數(shù)據(jù)狀態(tài)存儲和窗口聚合計算全部放在離線層。窗口結(jié)果聚合通過 離線窗口觸發(fā)機制完成，把特征結(jié)果 推到在線 KV 存儲。在線模塊非常輕量級，只負責簡單的在線 serving，極大地簡化了在線層的架構(gòu)復雜度。在離線狀態(tài)存儲層。我們主要依賴 Flink 提供的 原生狀態(tài)存儲引擎 RocksDB，充分利用離線計算集群本地的 SSD 磁盤資源，極大減輕在線 KV 存儲的資源壓力。

對于長窗口的特征（7 天以上窗口特征），由于涉及 Flink 狀態(tài)層明細數(shù)據(jù)的回溯過程，F(xiàn)link Embedded 狀態(tài)存儲引擎沒有提供特別好的外部數(shù)據(jù)回灌機制（或者說不適合做）。因此對于這種“狀態(tài)冷啟動”場景，我們引入了中心化存儲作為底層狀態(tài)存儲層的存儲介質(zhì)，整體是 Hybrid架構(gòu)。例如 7 天以內(nèi)的狀態(tài)存儲在本地 SSD，7～30 天狀態(tài)存儲到中心化的存儲引擎，離線數(shù)據(jù)回溯可以非常方便的寫入中心化存儲。

除窗口特征外，這套機制同樣適用于其他類型的有狀態(tài)特征（如序列類型的特征）。

實時特征分類體系 

整體架構(gòu) 

帶有窗口的特征，例如抖音視頻最近 1h 的點贊量（滑動窗口）、直播間用戶最近一個 session 的看播時長（session 窗口）等；

數(shù)據(jù)源層 

在新的一體化特征架構(gòu)中，我們統(tǒng)一把各種類型數(shù)據(jù)源抽象為 Schema Table，這是因為底層依賴的 Flink SQL 計算引擎層對數(shù)據(jù)源提供了非常友好的 Table Format 抽象。在推薦場景，依賴的數(shù)據(jù)源非常多樣，每個特征上游依賴一個或者多個數(shù)據(jù)源。數(shù)據(jù)源可以是 Kafka、RMQ、KV 存儲、RPC 服務。對于多個數(shù)據(jù)源，支持數(shù)據(jù)源流式、批式拼接，拼接類型包括 Window Join 和基于 key 粒度的 Window Union Join，維表 Join 支持 Abase、RPC、HIVE 等。具體每種類型的拼接邏輯如下：

三種類型的 Join 和 Union 可以組合使用，實現(xiàn)復雜的多數(shù)據(jù)流拼接。例如 (A union B) Window Join (C Lookup Join D)。

另外，F(xiàn)link SQL 支持復雜字段的計算能力，也就是業(yè)務方可以基于數(shù)據(jù)源定義的 TableSchema 基礎字段實現(xiàn)擴展字段的計算。業(yè)務計算邏輯本質(zhì)是一個 UDF，我們會提供 UDF API 接口給業(yè)務方，然后上傳 JAR 到特征后臺加載。另外對于比較簡單的計算邏輯，后臺也支持通過提交簡單的 Python 代碼實現(xiàn)多語言計算。

業(yè)務 DSL 

從業(yè)務視角提供高度抽象的特征生產(chǎn) DSL 語言，屏蔽底層計算、存儲引擎細節(jié)，讓業(yè)務方聚焦于業(yè)務特征定義。業(yè)務 DSL 層提供：數(shù)據(jù)來源、數(shù)據(jù)格式、數(shù)據(jù)抽取邏輯、數(shù)據(jù)生成特征類型、數(shù)據(jù)輸出方式等。

狀態(tài)存儲層 

如上文所述，新的特征一體化方案解決的主要痛點是：如何應對各種類型（一般是滑動窗口）有狀態(tài)特征的計算問題。對于這類特征，在離線計算層架構(gòu)里會有一個狀態(tài)存儲層，把抽取層提取的 RawFeature 按照切片 Slot 存儲起來 (切片可以是時間切片、也可以是 Session 切片等)。切片類型在內(nèi)部是一個接口類型，在架構(gòu)上可以根據(jù)業(yè)務需求自行擴展。狀態(tài)里面其實存儲的不是原始 RawFeature（存儲原始的行為數(shù)據(jù)太浪費存儲空間），而是轉(zhuǎn)化為 FeaturePayload 的一種 POJO 結(jié)構(gòu)，這個結(jié)構(gòu)里面支持了常見的各種數(shù)據(jù)結(jié)構(gòu)類型：

Int：存儲簡單的計數(shù)值類型 (多維度 counter)；

HashMap<int, int>：存儲二維計數(shù)值，例如 Action Counter，key 為 target_id，value 為計數(shù)值；

SortedMap<int, int>: 存儲 topk 二維計數(shù) ；

LinkedList

• ：存儲 id_list 類型數(shù)據(jù)；

HashMap<int, List

• >：存儲二維 id_list；

自定義類型，業(yè)務可以根據(jù)需求 FeaturePayload 里面自定義數(shù)據(jù)類型

狀態(tài)層更新的業(yè)務接口：輸入是 SQL 抽取 / 拼接層抽取出來的 RawFeature，業(yè)務方可以根據(jù)業(yè)務需求實現(xiàn) updateFeatureInfo 接口對狀態(tài)層的更新。對于常用的特征類型內(nèi)置實現(xiàn)了 update 接口，業(yè)務方自定義特征類型可以繼承 update 接口實現(xiàn)。

/** *  特征狀態(tài) update 接口 */public interface FeatureStateApi extends Serializable {    /**     * 特征更新接口, 上游每條日志會提取必要字段轉(zhuǎn)換為 fields, 用來更新對應的特征狀態(tài)     *     * @param fields     *      context: 保存特征名稱、主鍵 和 一些配置參數(shù) ;     *      oldFeature: 特征之前的狀態(tài)     *      fields: 平臺 / 配置文件 中的抽取字段     * @return     */FeaturePayLoad assign(Context context,FeaturePayLoad feature, Map<String, Object> rawFeature);} 

當然對于無狀態(tài)的 ETL 特征是不需要狀態(tài)存儲層的。

計算層 

特征計算層完成特征計算聚合邏輯，有狀態(tài)特征計算輸入的數(shù)據(jù)是狀態(tài)存儲層存儲的帶有切片的 FeaturePayload 對象。簡單的 ETL 特征沒有狀態(tài)存儲層，輸入直接是 SQL 抽取層的數(shù)據(jù) RawFeature 對象，具體的接口如下：

/** *  有狀態(tài)特征計算接口 */public interface FeatureStateApi extends Serializable {    /**     * 特征聚合接口，會根據(jù)配置的特征計算窗口, 讀取窗口內(nèi)所有特征狀態(tài)，排序后傳入該接口     *     * @param featureInfos, 包含 2 個 field     *      timeslot: 特征狀態(tài)對應的時間槽     *      Feature: 該時間槽的特征狀態(tài)     * @return     */    FeaturePayLoad aggregate(Context context, List<Tuple2<Slot, FeaturePayLoad>> slotStates);} 

有狀態(tài)特征聚合接口

/** *  無狀態(tài)特征計算接口 */public interface FeatureConvertApi extends Serializable {    /**     * 轉(zhuǎn)換接口, 上游每條日志會提取必要字段轉(zhuǎn)換為 fields, 無狀態(tài)計算時，轉(zhuǎn)換為 gauss 內(nèi)的 feature 類型 ;     *     * @param fields     *      fields: 平臺 / 配置文件 中的抽取字段     * @return     */    FeaturePayLoad convert(Context context,  FeaturePayLoad featureSnapshot, Map<String, Object> rawFeatures);} 

?

無狀態(tài)特征計算接口

另外通過觸發(fā)機制來觸發(fā)特征計算層的執(zhí)行，目前支持的觸發(fā)機制主要有：

業(yè)務落地 

目前在字節(jié)推薦場景，新一代特征架構(gòu)已經(jīng)在抖音直播、電商、推送、抖音推薦等場景陸續(xù)上線了一些實時特征。主要是有狀態(tài)類型的特征，帶有窗口的一維統(tǒng)計類型、二維倒排拉鏈類型、二維 TOPK 類型、實時 CTR/CVR Rate 類型特征、序列類型特征等。

在業(yè)務核心指標達成方面成效顯著。在直播場景，依托新特征架構(gòu)強大的表達能力上線了一批特征之后，業(yè)務看播核心指標、互動指標收益非常顯著。在電商場景，基于新特征架構(gòu)上線了 400+ 實時特征。其中在直播電商方面，業(yè)務核心 GMV、下單率指標收益顯著。在抖音推送場景，基于新特征架構(gòu)離線狀態(tài)的存儲能力，聚合用戶行為數(shù)據(jù)然后寫入下游各路存儲，極大地緩解了業(yè)務下游數(shù)據(jù)庫的壓力，在一些場景中 QPS 可以下降到之前的 10% 左右。此外，抖音推薦 Feed、評論等業(yè)務都在基于新特征架構(gòu)重構(gòu)原有的特征體系。

值得一提的是，在電商和抖音直播場景，F(xiàn)link 流式任務狀態(tài)最大已經(jīng)達到 60T，而且這個量級還在不斷增大。預計不久的將來，單任務的狀態(tài)有可能會突破 100T，這對架構(gòu)的穩(wěn)定性是一個不小的挑戰(zhàn)。

性能優(yōu)化 

Flink State Cache 

目前 Flink 提供兩類 StateBackend：基于 Heap 的 FileSystemStateBackend 和基于 RocksDB 的 RocksDBStateBackend。對于 FileSystemStateBackend，由于數(shù)據(jù)都在內(nèi)存中，訪問速率很快，沒有額外開銷。而 RocksDBStateBackend 存在查盤、序列化 / 反序列化等額外開銷，CPU 使用量會有明顯上升。在字節(jié)內(nèi)部有大量使用 State 的作業(yè)，對于大狀態(tài)作業(yè)，通常會使用 RocksDBStateBackend 來管理本地狀態(tài)數(shù)據(jù)。RocksDB 是一個 KV 數(shù)據(jù)庫，以 LSM 的形式組織數(shù)據(jù)，在實際使用的過程中，有以下特點：

應用層和 RocksDB 的數(shù)據(jù)交互是以 Bytes 數(shù)組的形式進行，應用層每次訪問都需要序列化 / 反序列化；

數(shù)據(jù)以追加的形式不斷寫入 RocksDB 中，RocksDB 后臺會不斷進行 compaction 來刪除無效數(shù)據(jù)。

業(yè)務方使用 State 的場景多是 get-update，在使用 RocksDB 作為本地狀態(tài)存儲的過程中，出現(xiàn)過以下問題：

爬蟲數(shù)據(jù)導致熱 key，狀態(tài)會不斷進行更新 (get-update)，單 KV 數(shù)據(jù)達到 5MB，而 RocksDB 追加更新的特點導致后臺在不斷進行 flush 和 compaction，單 task 出現(xiàn)慢節(jié)點（抖音直播場景）。

電商場景作業(yè)多數(shù)為大狀態(tài)作業(yè) (目前已上線作業(yè)狀態(tài)約 60TB)，業(yè)務邏輯中會頻繁進行 State 操作。在融合 Flink State 過程中發(fā)現(xiàn) CPU 的開銷和原有~~ 的~~ 基于內(nèi)存或 abase 的實現(xiàn)有 40%~80% 的升高。經(jīng)優(yōu)化后，CPU 開銷主要集中在序列化 / 反序列化的過程中。

針對上述問題，可以通過在內(nèi)存維護一個對象 Cache，達到優(yōu)化熱點數(shù)據(jù)訪問和降低 CPU 開銷的目的。通過上述背景介紹，我們希望能為 StateBackend 提供一個通用的 Cache 功能，通過 Flink StateBackend Cache 功能設計方案達成以下目標：

減少 CPU 開銷 ：通過對熱點數(shù)據(jù)進行緩存，減少和底層 StateBackend 的交互次數(shù)，達到減少序列化 / 反序列化開銷的目的。

提升 State 吞吐能力 ：通過增加 Cache 后，State 吞吐能力應比原有的 StateBackend 提供的吞吐能力更高。理論上在 Cache 足夠大的情況下，吞吐能力應和基于 Heap 的 StateBackend 近似。

Cache 功能通用化 ：不同的 StateBackend 可以直接適配該 Cache 功能。目前我們主要支持 RocksDB，未來希望可以直接提供給別的 StateBackend 使用，例如 RemoteStateBackend。

經(jīng)過和字節(jié)基礎架構(gòu) Flink 團隊的合作，在實時特征生產(chǎn)升級 ，上線 Cache 大部分場景的 CPU 使用率大概會有高達 50% 左右的收益；

PB IDL 裁剪 

在字節(jié)內(nèi)部的實時特征離線生成鏈路當中，我們主要依賴的數(shù)據(jù)流是 Kafka。這些 Kafka 都是通過 PB 定義的數(shù)據(jù)，字段繁多。公司級別的大 Topic 一般會有 100+ 的字段，但大部分的特征生產(chǎn)任務只使用了其中的部分字段。對于 Protobuf 格式的數(shù)據(jù)源，我們可以完全通過裁剪數(shù)據(jù)流，mask 一些非必要的字段來節(jié)省反序列化的開銷。PB 類型的日志，可以直接裁剪 idl，保持必要字段的序號不變，在反序列化的時候會跳過 unknown field 的解析，這 對于 CPU 來說是更節(jié)省的，但是網(wǎng)絡帶寬不會有收益，預計裁剪后能節(jié)省非常多的 CPU 資源。在上線了 PB IDL 裁剪之后，大部分任務的 CPU 收益在 30% 左右。

遇到的問題 

新架構(gòu)特征生產(chǎn)任務本質(zhì)就是一個有狀態(tài)的 Flink 任務，底層的狀態(tài)存儲 StateBackend 主要是本地的 RocksDB。主要面臨兩個比較難解的問題，一是任務 DAG 變化 Checkpoint 失效，二是本地存儲不能很好地支持特征狀態(tài)歷史數(shù)據(jù)回溯。

實時特征任務不能動態(tài)添加新的特征：對于一個線上的 Flink 實時特征生產(chǎn)任務，我們不能隨意添加新的特征。這是由于引入新的特征會導致 Flink 任務計算的 DAG 發(fā)生改變，從而導致 Flink 任務的 Checkpoint 無法恢復，這對實時有狀態(tài)特征生產(chǎn)任務來說是不能接受的。目前我們的解法是禁止更改線上部署的特征任務配置，但這也就導致了線上生成的特征是不能隨便下線的。對于這個問題暫時沒有找到更好的解決辦法，后期仍需不斷探索。

特征狀態(tài)冷啟動問題：目前主要的狀態(tài)存儲引擎是 RocksDB，不能很好地支持狀態(tài)數(shù)據(jù)的回溯。

后續(xù)規(guī)劃 

當前新一代架構(gòu)還在字節(jié)推薦場景中快速演進，目前已較好解決了實時窗口特征的生產(chǎn)問題。

出于實現(xiàn)統(tǒng)一推薦場景下特征生產(chǎn)的目的，我們后續(xù)會繼續(xù)基于 Flink SQL 流批一體能力，在批式特征生產(chǎn)發(fā)力。此外也會基于 Hudi 數(shù)據(jù)湖技術(shù)，完成特征的實時入湖，高效支持模型訓練場景離線特征回溯痛點。規(guī)則引擎方向，計劃繼續(xù)探索 CEP，推動在電商場景有更多落地實踐。在實時窗口計算方向，將繼續(xù)深入調(diào)研 Flink 原生窗口機制，以期解決目前方案面臨的窗口特征數(shù)據(jù)退場問題。

支持批式特征：這套特征生產(chǎn)方案主要是解決實時有狀態(tài)特征的問題，而目前字節(jié)離線場景下還有大量批式特征是通過 Spark SQL 任務生產(chǎn)的。后續(xù)我們也會基于 Flink SQL 流批一體的計算能力，提供對批式場景特征的統(tǒng)一支持，目前也初步有了幾個場景的落地；

特征離線入湖：基于 Hudi On Flink 支持實時特征的離線數(shù)倉建設，主要是為了支持模型訓練樣本拼接場景離線特征回溯；

Flink CEP 規(guī)則引擎支持：Flink SQL 本質(zhì)上就是一種規(guī)則引擎，目前在線上我們把 Flink SQL 作為業(yè)務 DSL 過濾語義底層的執(zhí)行引擎。但 Flink SQL 擅長表達的 ETL 類型的過濾規(guī)則，不能表達帶有時序類型的規(guī)則語義。在直播、電商場景的時序規(guī)則需要嘗試 Flink CEP 更加復雜的規(guī)則引擎。

Flink Native Windowing 機制引入：對于窗口類型的有狀態(tài)特征，我們目前采用上文所述的抽象 SlotState 時間切片方案統(tǒng)一進行支持。另外 Flink 本身提供了非常完善的窗口機制，通過 Window Assigner、Window Trigger 等組件可以非常靈活地支持各種窗口語義。因此后續(xù)我們也會在窗口特征計算場景引入 Flink 原生的 Windowing 機制，更加靈活地支持窗口特征迭代。

Flink HybridState Backend 架構(gòu)：目前在字節(jié)的線上場景中，F(xiàn)link 底層的 StateBackend 默認都是使用 RocksDB 存儲引擎。這種內(nèi)嵌的存儲引擎不能通過外部機制去提供狀態(tài)數(shù)據(jù)的回灌和多任務共享，因此我們需要支持 KV 中心化存儲方案，實現(xiàn)靈活的特征狀態(tài)回溯。

靜態(tài)屬性類型特征統(tǒng)一管理：通過特征平臺提供統(tǒng)一的 DSL 語義，統(tǒng)一管理其他外部靜態(tài)類型的特征服務。例如一些其他業(yè)務團隊維度的用戶分類、標簽服務等。

本文來自微信公眾號 “InfoQ”（ID：infoqchina），作者：郭文飛，36氪經(jīng)授權(quán)發(fā)布。

關(guān)鍵詞： 5年 迭代 5次 抖音