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

業(yè)務(wù)背景 

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

主要業(yè)務(wù)場(chǎng)景 

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

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

業(yè)務(wù)痛點(diǎn)和挑戰(zhàn) 

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

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

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

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

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

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

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

迭代演進(jìn)過(guò)程 

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

推薦場(chǎng)景特征服務(wù)演進(jìn)歷程 

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

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

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

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

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

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

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

實(shí)時(shí)特征分類體系 

整體架構(gòu) 

帶有窗口的特征，例如抖音視頻最近 1h 的點(diǎn)贊量（滑動(dòng)窗口）、直播間用戶最近一個(gè) session 的看播時(shí)長(zhǎng)（session 窗口）等；

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

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

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

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

業(yè)務(wù) DSL 

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

狀態(tài)存儲(chǔ)層 

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

Int：存儲(chǔ)簡(jiǎn)單的計(jì)數(shù)值類型 (多維度 counter)；

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

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

LinkedList

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

HashMap<int, List

• >：存儲(chǔ)二維 id_list；

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

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

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

當(dāng)然對(duì)于無(wú)狀態(tài)的 ETL 特征是不需要狀態(tài)存儲(chǔ)層的。

計(jì)算層 

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

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

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

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

?

無(wú)狀態(tài)特征計(jì)算接口

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

業(yè)務(wù)落地 

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

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

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

性能優(yōu)化 

Flink State Cache 

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

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

數(shù)據(jù)以追加的形式不斷寫入 RocksDB 中，RocksDB 后臺(tái)會(huì)不斷進(jìn)行 compaction 來(lái)刪除無(wú)效數(shù)據(jù)。

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

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

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

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

減少 CPU 開(kāi)銷 ：通過(guò)對(duì)熱點(diǎn)數(shù)據(jù)進(jìn)行緩存，減少和底層 StateBackend 的交互次數(shù)，達(dá)到減少序列化 / 反序列化開(kāi)銷的目的。

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

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

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

PB IDL 裁剪 

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

遇到的問(wèn)題 

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

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

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

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

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

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

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

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

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

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

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

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

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

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